Java 中 CyclicBarrier 怎么配合 CompletableFuture 使用

在 Java 并发编程中，CyclicBarrier 和 CompletableFuture 是两种截然不同的工具。前者是同步协调的「老将」，让多个线程在屏障点互相等待；后者是异步编排的「新贵」，擅长非阻塞链式处理。两者设计哲学对立，强行混用很容易踩坑——线程阻塞、线程池耗尽、甚至死锁，这些场景在社区里已经被反复讨论过。但话说回来，开发中确实会遇到「等待多个异步任务全部完成，再统一触发后续动作」的需求，这正是 CyclicBarrier 的典型应用场景。那么，在异步编程的语境下，该怎么优雅地实现这个语义？

用 CompletableFuture.allOf 替代 CyclicBarrier 的核心语义

CyclicBarrier 最经典的用法是「等 N 个线程都准备好，再一起继续」。在异步世界里，这直接对应「等待 N 个异步任务全部完成」。CompletableFuture 原生支持这种场景：

• CompletableFuture.allOf(CompletableFuture...)：返回一个新的 Future，当所有入参 Future 都完成时（无论成功还是异常），它才完成。
• 如果需要收集所有结果，可以在 thenApply 中调用 join()（注意：此时所有 Future 都已完成，join 不会阻塞），或者用 thenCombine/thenCompose 链式收集。
• 来看一个典型例子：启动三个异步任务，全部完成后打印汇总信息：

CompletableFuture task1 = CompletableFuture.supplyAsync(() -> {
    sleep(1000); return "A";
});
CompletableFuture task2 = CompletableFuture.supplyAsync(() -> {
    sleep(800);  return "B";
});
CompletableFuture task3 = CompletableFuture.supplyAsync(() -> {
    sleep(1200); return "C";
});

CompletableFuture.allOf(task1, task2, task3)
    .thenRun(() -> {
        List results = Arrays.asList(
            task1.join(), task2.join(), task3.join()
        );
        System.out.println("All done: " + results);
    })
    .join();

避免在 CompletableFuture 中调用 await() 或 awaitUninterruptibly

一个很容易犯的错误：在 supplyAsync 或 thenApply 的回调里直接调用 CyclicBarrier.await()。这会让工作线程（比如 ForkJoinPool.commonPool() 中的线程）陷入阻塞，吞吐量瞬间下降，严重时甚至拖垮整个线程池。原因很简单：CompletableFuture 的回调默认运行在异步线程池中，而这些线程的设计初衷是执行计算密集型任务，不是长期阻塞等待。

• 如果必须集成遗留的 CyclicBarrier 同步代码，一个折中方案是明确指定自定义线程池，并且确保线程池容量足够大、支持阻塞操作。
• 更推荐的思路：把同步协调逻辑迁移到主线程或专用调度线程中，让 CompletableFuture 只负责纯粹的异步计算，互不干扰。

需要「分阶段协同」？考虑用 CompletableFuture 链式编排

有没有遇到这样的需求：先并行执行阶段一，全部完成后，再并行执行阶段二？这其实不是 CyclicBarrier 的典型场景，而是异步流水线的最佳实践。用 CompletableFuture 的链式编排可以轻松实现：

• 阶段一用 allOf 汇总所有异步任务；
• 阶段二在 thenCompose 或 thenApplyAsync 中触发新的异步任务；
• 还可以嵌套、异常处理、超时控制，比手动维护 barrier 更健壮、更灵活。

// 阶段一：并发获取数据
CompletableFuture> stage1 = CompletableFuture.allOf(f1, f2, f3)
    .thenApply(v -> Arrays.asList(f1.join(), f2.join(), f3.join()));

// 阶段二：基于 stage1 结果并发处理
CompletableFuture> stage2 = stage1.thenCompose(dataList ->
    CompletableFuture.allOf(
        dataList.stream()
                .map(d -> CompletableFuture.supplyAsync(() -> process(d)))
                .toArray(CompletableFuture[]::new)
    ).thenApply(v -> dataList.stream()
        .map(d -> process(d))
        .collect(Collectors.toList()))
);

真要混合使用？封装成异步友好的 Barrier

极少数场景下（比如测试、模拟多线程协作），确实需要保留 CyclicBarrier 的阻塞等待行为，同时又想接入 CompletableFuture 的异步流程。这里有一个技巧：

• 创建一个 CompletableFuture，由 CyclicBarrier.await() 成功执行后触发 complete()；
• 使用独立的线程来调用 await()，避免污染异步线程池；
• 务必设置超时，防止 await 永久阻塞。

CyclicBarrier barrier = new CyclicBarrier(3);
CompletableFuture barrierFuture = new CompletableFuture<>();

Thread barrierThread = new Thread(() -> {
    try {
        barrier.await(5, TimeUnit.SECONDS);
        barrierFuture.complete(null);
    } catch (Exception e) {
        barrierFuture.completeExceptionally(e);
    }
});
barrierThread.start();

// 后续链式操作
barrierFuture.thenRun(() -> System.out.println("Barrier passed"))
             .join();

这样一来，CyclicBarrier 的同步等待被隔离在独立线程中，既保留了它的行为，又不影响异步编排的流畅性。当然，能不用的话还是尽量别用——毕竟，在异步世界里，allOf + 链式编排才是更自然、更安全的做法。