功能简介
- 固定线程、限制最大队列长度的自定义线程池；
- 定制线程池加载任务、子线程各种参数，如分页大小、是否子线程出错就停止；
- 根据分页大小分页得到分页集，根据队列大小分配分页集得到线程处理的数据；
- 返回线程池分解、加载任务结果、子线程处理结果以及耗时（估量）；
- 监控线程池执行状况（可选是否监控，可自定义刷新时间）
示例
（ThreadPoolBuilder中设置的均是可选参数，可设置0-N个参数）
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("耗时" + result.getTimeCost());
关键代码：
/**
* 多线程处理数据
*
* @param data 数据
* @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("{}线程池处理结束...耗时：{}", 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("{}待处理的数据为空,线程池结束", 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("{}线程池实际队列长度: {}", biz, actualQueueSize);
this.threadResults = Lists.newArrayListWithCapacity(sliceSize);
final ThreadPoolExecutor threadPoolExecutor = initThreadPool(actualQueueSize);
try {
for (final List<List<T>> slice : slices) {
// 遇错即止可能使线程池提前结束
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("{}子线程执行异常: ", 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("{}当前已执行任务数量：{}, 总数量：{}, 进度：{}", biz, executed, actualQueueSize, NumberFormat.getPercentInstance().format((double) executed / actualQueueSize));
try {
TimeUnit.MILLISECONDS.sleep(monitorInterval);
} catch (InterruptedException e) {
log.info("{}任务执行状况睡眠异常: ", biz);
}
}
} else {
try {
threadPoolExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.info("{}任务等待执行结束被中断: ", 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
