Consumer本身不支持线程同步,但可封装进Runnable或CompletableFuture,配合CountDownLatch实现“全部完成再继续”的汇聚同步;核心是用latch.await()阻塞主线程,各子线程执行consumer后调用countDown()。

在 Java 中,Consumer 本身只是函数式接口,不直接支持线程同步;但你可以把它作为任务逻辑封装进多线程执行单元(比如 Runnable 或 CompletableFuture),再配合 CountDownLatch 实现“所有子任务完成后再继续主流程”的汇聚同步效果。
核心思路:用 CountDownLatch 控制主线程等待
CountDownLatch 是一个一次性同步辅助类,初始化时指定计数(如任务数),每个子线程完成时调用 countDown(),主线程调用 await() 阻塞等待计数归零。
关键点:
- Consumer 负责处理单个数据或执行某段业务逻辑
- 每个子线程把 Consumer 的逻辑包装成 Runnable 执行
- 主线程在启动所有子线程后,立即 await(),等全部完成才往下走
典型写法示例(固定线程池 + Consumer + CountDownLatch)
假设你有一组数据,想并发处理,每条数据交给一个 Consumer<string></string> 处理,等全部处理完再打印汇总信息:
CountDownLatch latch = new CountDownLatch(dataList.size());
ExecutorService executor = Executors.newFixedThreadPool(4);
<p>Consumer<String> processor = item -> {
try {
// 模拟耗时处理
Thread.sleep(100);
System.out.println("Processed: " + item);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown(); // 必须放在 finally,确保计数一定减少
}
};</p><p>for (String item : dataList) {
executor.submit(() -> processor.accept(item));
}</p><p>try {
latch.await(); // 主线程阻塞,直到所有子任务调用 countDown()
System.out.println("All done.");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}</p><p>executor.shutdown();
注意事项和常见坑
-
countDown() 必须确保执行:建议包在
finally块里,避免异常导致计数没减,主线程永久阻塞 -
latch 不能重用:
CountDownLatch是一次性用品,用完即废;需要重复使用请换CyclicBarrier - 不要在 Consumer 内部直接 new Thread:容易失控线程数,推荐用线程池统一管理
-
异常传播需手动处理:子线程中抛出的异常不会自动传给主线程,如需收集错误,可配合
AtomicInteger或ConcurrentLinkedQueue记录失败项
替代方案提示(更现代/灵活的选择)
如果项目已用上 Java 8+ 的 CompletableFuture,可更简洁地实现类似效果:
立即学习“Java免费学习笔记(深入)”;
List<CompletableFuture<Void>> futures = dataList.stream()
.map(item -> CompletableFuture.runAsync(() -> processor.accept(item), executor))
.collect(Collectors.toList());
<p>CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
System.out.println("All done.");
它天然支持异常传播、组合编排,且无需手动管理计数器,适合复杂依赖场景。


















