功能簡介

• 固定線程、限制最大隊(duì)列長度的自定義線程池；
• 定制線程池加載任務(wù)、子線程各種參數(shù)，如分頁大小、是否子線程出錯(cuò)就停止；
• 根據(jù)分頁大小分頁得到分頁集，根據(jù)隊(duì)列大小分配分頁集得到線程處理的數(shù)據(jù)；
• 返回線程池分解、加載任務(wù)結(jié)果、子線程處理結(jié)果以及耗時(shí)（估量）；
• 監(jiān)控線程池執(zhí)行狀況（可選是否監(jiān)控，可自定義刷新時(shí)間）

示例

（ThreadPoolBuilder中設(shè)置的均是可選參數(shù)，可設(shè)置0-N個(gè)參數(shù)）

ThreadPoolProcessor<Integer> threadPoolProcessor = ThreadPoolBuilder.<Integer>newBuilder()
                .setThreadNum(3)
                .setMaxQueueNum(2)
                .setPageSize(2)
                .setBiz("test")
                .setMonitorTaskStatus(false)
                .build();
        ThreadPoolResult result = threadPoolProcessor.execute(Arrays.asList(1, 2, 3, 4, 5, 6), new ThreadProcessor<Integer>() {
            @Override
            public void execute(List<List<Integer>> pages) {
                for (List<Integer> page : pages) {
                    for (Integer item : page) {
                        System.out.println(item);
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });
System.out.println(result.isAllThreadsSuccess());
System.out.println("耗時(shí)" + result.getTimeCost());

關(guān)鍵代碼：

  /**
     * 多線程處理數(shù)據(jù)
     *
     * @param data            數(shù)據(jù)
     * @param threadProcessor 子線程處理邏輯
     * @return
     */
    public ThreadPoolResult execute(List<T> data, ThreadProcessor<T> threadProcessor) {
        log.error("{}線程池處理開始...", biz);
        long start = System.currentTimeMillis();
        long timeCost = 0;
        try {
            process(data, threadProcessor);
            timeCost = (System.currentTimeMillis() - start) / 1000;
            log.info("{}線程池處理結(jié)束...耗時(shí)：{}", biz, timeCost);
            return new ThreadPoolResult(true, null, this.threadResults, timeCost);
        } catch (Throwable e) {
            log.error("{}線程池開啟異常：", biz, e);
            return new ThreadPoolResult(false, e, this.threadResults, timeCost);
        }
    }

    private void process(List<T> data, final ThreadProcessor<T> threadProcessor) {
        if (CollectionUtils.isEmpty(data)) {
            log.info("{}待處理的數(shù)據(jù)為空,線程池結(jié)束", biz);
            return;
        }
        List<List<T>> pageList = ListUtils.divideList(data, pageSize);
        List<List<List<T>>> slices = ListUtils.groupListBalancedly(pageList, maxQueueNum + maxThreadNum);
        int sliceSize = slices.size();
        int actualQueueSize = sliceSize > maxThreadNum ? sliceSize - maxThreadNum : 1;
        log.info("{}線程池實(shí)際隊(duì)列長度: {}", biz, actualQueueSize);
        this.threadResults = Lists.newArrayListWithCapacity(sliceSize);
        final ThreadPoolExecutor threadPoolExecutor = initThreadPool(actualQueueSize);
        try {
            for (final List<List<T>> slice : slices) {
                // 遇錯(cuò)即止可能使線程池提前結(jié)束
                if (!threadPoolExecutor.isTerminated()) {
                    threadPoolExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                if (hasError && stopWhenError) {
                                    threadPoolExecutor.shutdownNow();
                                    return;
                                }
                                threadProcessor.execute(slice);
                                executedTaskNums.incrementAndGet();
                                threadResults.add(new ThreadPoolResult.ThreadResult(true, null));
                            } catch (Exception e) {
                                e.printStackTrace();
                                log.error("{}子線程執(zhí)行異常: ", biz, e);
                                hasError = true;
                                threadResults.add(new ThreadPoolResult.ThreadResult(false, e));
                            }
                        }
                    });
                }
            }
        } finally {
            threadPoolExecutor.shutdown();
        }
        if (monitorTaskStatus) {
            while (!threadPoolExecutor.isTerminated()) {
                int executed = executedTaskNums.get();
                log.info("{}當(dāng)前已執(zhí)行任務(wù)數(shù)量：{}, 總數(shù)量：{}, 進(jìn)度：{}", biz, executed, actualQueueSize, NumberFormat.getPercentInstance().format((double) executed / actualQueueSize));
                try {
                    TimeUnit.MILLISECONDS.sleep(monitorInterval);
                } catch (InterruptedException e) {
                    log.info("{}任務(wù)執(zhí)行狀況睡眠異常: ", biz);
                }
            }
        } else {
            try {
                threadPoolExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                log.info("{}任務(wù)等待執(zhí)行結(jié)束被中斷: ", biz);
            }
        }

    }

    private ThreadPoolExecutor initThreadPool(int queueNum) {
        String threadNameFormat = this.biz + "-thread-pool-%d";
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(threadNameFormat).build();
        BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(queueNum);
        RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                log.error("{} BlockingQueue is full!", biz);
            }
        };
        return new ThreadPoolExecutor(this.maxThreadNum, this.maxThreadNum, 10L,
                TimeUnit.SECONDS, taskQueue, namedThreadFactory, rejectedExecutionHandler);
    }

源碼見：https://github.com/hzhqk/java/tree/master/util/threadpool