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；

• IO密集型（如HTTP调用、文件读写）：线程数可设为CPU核心数的2～4倍，甚至更高，需结合压测调整；

• 混合型任务：优先按IO密集型预估，再通过监控线程活跃度和任务排队时长动态调优。

手动构造ThreadPoolExecutor

第一步：定义核心参数

int corePoolSize = 8;

int maxPoolSize = 16;

long keepAliveTime = 60L;

BlockingQueue workQueue = new ArrayBlockingQueue(100); // 有界队列，防内存爆炸

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> tasks = new ArrayList();

for (int i = 0; i

tasks.add(() -> { Thread.sleep(100); return "task-" + i; });

}

Java Maven Code Review

审查Java Maven项目（ZIP压缩包或GitLab仓库URL），检查代码规范、命名、模块边界、可维护性问题以及重复代码。

下载

List> futures = executor.invokeAll(tasks); // 阻塞直到全部完成

for (Future future : futures) { System.out.println(future.get()); }

方法二：submit + 循环get —— 手动控制超时与异常

适合需要逐个处理结果、且不能容忍单个任务无限阻塞的情况。

List> futures = new ArrayList();

for (int i = 0; i

futures.add(executor.submit(() -> "result-" + i));

}

for (Future future : futures) {

try {

System.out.println(future.get(3, TimeUnit.SECONDS)); // 单任务超时3秒

} catch (TimeoutException e) {

System.err.println("任务超时，已跳过");

future.cancel(true);

}

}

方法三：CompletionService —— 按完成顺序实时消费结果

这是处理大批量异步任务最优雅的方式，尤其适合“哪个先完成就先处理哪个”的场景，比如爬虫聚合、价格比对。

CompletionService cs = new ExecutorCompletionService(executor);

for (int i = 0; i

cs.submit(() -> { Thread.sleep(new Random().nextInt(500)); return "done-" + i; });

}

for (int i = 0; i

Future future = cs.take(); // 阻塞获取最先完成的任务

System.out.println("收到：" + future.get()); // 此处可立即做业务处理

}

Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南