多线程处理List数据

• 按照线程数量对List数据切分后构建线程执行业务操作
• • ThreadUtil
  • ThreadFunction
  • 测试方法

按照线程数量对List数据切分后构建线程执行业务操作

ThreadUtil

/**
 * @description: 线程工具
 * @author: wmh
 * @createDate: 2021-12-31
 * @version: 1.0
 */
public class ThreadUtil {

    public ThreadUtil() {

    }

    /**
     * @param list 数据
     * @param function 线程执行方法
     * @param perThreadHandleCount 每个线程处理多少个
     * @param maxThreadCount 最大线程数
     * @throws Exception
     */
    public  ThreadUtil(List<?> list,ThreadFunction function,int perThreadHandleCount,int maxThreadCount) throws Exception{
        //处理的总数
        int total = list.size();
        int threadCount = total%perThreadHandleCount==0?total/perThreadHandleCount:total/perThreadHandleCount+1;
        threadCount = Math.min(threadCount, maxThreadCount);
        perThreadHandleCount = total%threadCount==0?total/threadCount:total/threadCount+1;
        //定义线程池中线程数量
        ExecutorService execPool = Executors.newFixedThreadPool(threadCount);
        //按照线程数量进行数据切分
        List<List<?>> handleDatas = this.getLists(perThreadHandleCount, list);
        //构建线程
        for(List<?> handleData : handleDatas){
            HttpThread serviceThread = new HttpThread(handleData, function);
            execPool.submit(serviceThread);
        }
        execPool.shutdown();
    }

    static class HttpThread extends Thread {
        List<?> list ;
        ThreadFunction function ;
        public HttpThread(List<?> listHandleData,ThreadFunction function){
            this.list = listHandleData;
            this.function = function;
        }
        public void run() {
            try {
                function.apply(list);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 数组切分函数，将List对象的数据按照threadHandleCount处理到多个List中
     * @param threadHandleCount 每个线程处理多少数据
     * @param mList 需要拆分的List
     * @return
     */
    private List<List<?>> getLists(int threadHandleCount, List<?> mList) {
        List<List<?>>mEndList=new ArrayList<>();
        if( mList.size()%threadHandleCount!=0) {
            for (int j = 0; j < mList.size() / threadHandleCount + 1; j++) {
                if ((j * threadHandleCount + threadHandleCount) < mList.size()) {
                    mEndList.add(mList.subList(j * threadHandleCount, j * threadHandleCount + threadHandleCount));//0-3,4-7,8-11    j=0,j+3=3   j=j*3+1
                } else if ((j * threadHandleCount + threadHandleCount) > mList.size()) {
                    mEndList.add(mList.subList(j * threadHandleCount, mList.size()));
                } else if (mList.size() < threadHandleCount) {
                    mEndList.add(mList.subList(0, mList.size()));
                }
            }
        }else if(mList.size()%threadHandleCount==0){
            for (int j = 0; j < mList.size() / threadHandleCount; j++) {
                if ((j * threadHandleCount + threadHandleCount) <= mList.size()) {
                    mEndList.add(mList.subList(j * threadHandleCount, j * threadHandleCount + threadHandleCount));//0-3,4-7,8-11    j=0,j+3=3   j=j*3+1
                } else if ((j * threadHandleCount+ threadHandleCount) > mList.size()) {
                    mEndList.add(mList.subList(j * threadHandleCount, mList.size()));
                } else if (mList.size() < threadHandleCount) {
                    mEndList.add(mList.subList(0, mList.size()));
                }
            }
        }
        return mEndList;
    }


}

ThreadFunction

/**
 * @description: 线程执行业务
 * @author: wmh
 * @createDate: 2021-12-31
 * @version: 1.0
 */
public interface ThreadFunction {

    void apply(List<?> list) throws Exception;
}

测试方法

	public static void main(String[] args) throws Exception {
		List<String> list = new ArrayList<>();
		for (int i = 0; i < 100; i++) {
			list.add("测试"+i);
		}

		ThreadUtil threadUtil = new ThreadUtil(list, new ThreadFunction() {
			@Override
			public void apply(List<?> list) {
				for (Object o : list) {
					System.out.println(o.toString());
				}
			}
		},25,10);

	}