Java并发处理批量任务需手动构造ThreadPoolExecutor,按CPU/IO密集型配置线程数与有界队列,并用invokeAll、submit+get或CompletionService提交任务。

Java中设置并发处理批量任务,核心是选择合适的线程池类型、配置合理参数,并匹配任务特性提交任务。直接用Executors默认工厂方法创建线程池,在生产环境极易引发OOM或线程饥饿,必须手动构造ThreadPoolExecutor并显式控制队列与拒绝策略。
选对线程池类型
不要用Executors.newFixedThreadPool(5),它底层使用无界LinkedBlockingQueue,任务积压会导致内存溢出。【必须用ThreadPoolExecutor手动构造】
根据任务类型选型:
• CPU密集型(如数据计算、加密解密):线程数 ≈ CPU核心数 × (1 + 平均等待时间 / 平均工作时间),通常设为Runtime.getRuntime().availableProcessors()或加1~2;
立即学习“Java免费学习笔记(深入)”;
• IO密集型(如HTTP调用、文件读写):线程数可设为CPU核心数的2~4倍,甚至更高,需结合压测调整;
• 混合型任务:优先按IO密集型预估,再通过监控线程活跃度和任务排队时长动态调优。
手动构造ThreadPoolExecutor
第一步:定义核心参数
int corePoolSize = 8;
int maxPoolSize = 16;
long keepAliveTime = 60L;
BlockingQueue
ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("batch-task-pool-%d").build();
RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy(); // 重要:避免丢任务,由调用线程兜底执行
第二步:创建实例
ThreadPoolExecutor executor = new ThreadPoolExecutor(
corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
workQueue, threadFactory, handler);
注意:构造后不要立即submit任务,先调用executor.prestartAllCoreThreads()预热核心线程,避免首波任务触发扩容延迟。
提交批量任务的三种方式
方法一:invokeAll —— 等所有任务完成再统一取结果
适用于“必须等全部返回才能继续”的场景,比如批量校验后生成汇总报告。
List
for (int i = 0; i
tasks.add(() -> { Thread.sleep(100); return "task-" + i; });
}
List
for (Future
方法二:submit + 循环get —— 手动控制超时与异常
适合需要逐个处理结果、且不能容忍单个任务无限阻塞的情况。
List
for (int i = 0; i
futures.add(executor.submit(() -> "result-" + i));
}
for (Future
try {
System.out.println(future.get(3, TimeUnit.SECONDS)); // 单任务超时3秒
} catch (TimeoutException e) {
System.err.println("任务超时,已跳过");
future.cancel(true);
}
}
方法三:CompletionService —— 按完成顺序实时消费结果
这是处理大批量异步任务最优雅的方式,尤其适合“哪个先完成就先处理哪个”的场景,比如爬虫聚合、价格比对。
CompletionService
for (int i = 0; i
cs.submit(() -> { Thread.sleep(new Random().nextInt(500)); return "done-" + i; });
}
for (int i = 0; i
Future
System.out.println("收到:" + future.get()); // 此处可立即做业务处理
}



















