如何在Reactor中无需Thread.sleep等待Flux执行完成？

你当前用Thread.sleep循环判断Disposable是否释放的方式既低效又不可靠，而且直接用ArrayList收集元素还可能存在线程安全问题。下面分两种场景给出更合理的方案：

1. 必须阻塞等待的场景（如测试、独立程序入口） #

如果代码运行在非WebFlux的阻塞线程环境中（比如main方法、JUnit测试），可以直接用Reactor提供的block()方法，它会阻塞当前线程直到Flux完成并返回收集好的列表：

// 先并行处理其他任务
doSomeOtherStuff();

// 收集Flux所有元素并阻塞等待完成
List<MyObject> results = myFlux.collectList().block();

return results;

如果需要避免无限阻塞，可设置超时时间：

List<MyObject> results = myFlux.collectList().block(Duration.ofSeconds(10));

2. WebFlux非阻塞流程内的并行处理 #

如果是在WebFlux的请求处理链中，绝对不能用阻塞方法（会破坏非阻塞模型），应该用Reactor的操作符实现并行任务组合：

// 将其他任务包装为非阻塞的Mono（如果任务是阻塞型，指定单独的调度器）
Mono<Void> otherTask = Mono.fromRunnable(() -> doSomeOtherStuff())
                           .subscribeOn(Schedulers.boundedElastic());

// 收集Flux结果为Mono<List>
Mono<List<MyObject>> fluxResult = myFlux.collectList();

// 并行执行两个任务，全部完成后返回Flux的结果
return Mono.zip(otherTask, fluxResult)
           .map(tuple -> tuple.getT2());

这种方式全程非阻塞，完全符合WebFlux的设计理念，不需要手动管理线程或等待逻辑。

备选阻塞方案（不推荐但适用特殊场景） #

如果一定要保留手动订阅的方式，也可以用CountDownLatch替代Thread.sleep，避免无效循环：

// 用线程安全的列表收集元素
List<MyObject> results = Collections.synchronizedList(new ArrayList<>());
CountDownLatch latch = new CountDownLatch(1);

myFlux.doOnComplete(latch::countDown)
      .subscribe(results::add);

// 处理其他任务
doSomeOtherStuff();

// 等待Flux完成，可设置超时时间
latch.await(10, TimeUnit.SECONDS);

return results;

内容的提问来源于stack exchange，提问作者geanakuch