Java Reactor中subscribe方法何时阻塞？求解答

这本质是Reactor线程模型和操作执行方式的问题，核心看你的序列里有没有触发异步线程调度的操作：

先拆解你的两个例子差异 #

例子1（立即返回主线程） #

Flux.range(1, 1000)
    .delayElements(Duration.ofNanos(1))
    .map(integer -> integer + 1)
    .subscribe(System.out::println);
System.out.println("after");

输出：

after
2
3
4
...

这里的关键是delayElements操作符——它会把每个元素的处理逻辑提交到Reactor的异步线程池（默认是Schedulers.parallel()）执行。调用subscribe后，主线程只完成了订阅流程，不会等待异步线程的处理，所以直接往下走打印"after"，元素处理在后台线程异步执行。

例子2（同步执行到序列结束） #

Flux.range(1, 1000)
    .map(integer -> integer + 1)
    .subscribe(System.out::println);
System.out.println("after");

输出：

2
3
...
1001
after

这个序列里的所有操作都是同步的：

• Flux.range在调用线程上同步生成1到1000的元素
• map在同一个线程上同步执行+1操作
• 订阅后的元素消费（打印）同样在调用线程上完成

整个过程没有任何线程切换，所以调用subscribe后，主线程会把整个序列的生成、处理、消费全部做完，才会执行后面的println("after")——这不是subscribe本身阻塞，而是整个序列在调用线程上同步执行完了。

怎么判断subscribe后会不会同步执行？ #

看你的序列里有没有以下几类操作：

• 显式切换线程的操作符：publishOn、subscribeOn
• 天然异步的操作符：delayElements、delaySequence、timeout等需要调度的操作
• 异步数据源：比如从消息队列、异步IO获取数据的Flux/Mono

如果没有这些，且数据源是同步生成的（比如range、just），那么整个序列会在调用subscribe的线程上同步执行，直到所有元素处理完毕，才会继续执行后续代码。

批处理场景怎么控制执行顺序？ #

如果需要确保整个Flux处理完成后再执行后续逻辑（比如批处理结束后做收尾），别依赖"是否阻塞"的表现，推荐用Reactor原生方法或编排方式：

1. 使用blockLast()：等待整个Flux完成，比如：

Flux.range(1, 1000)
    .map(integer -> integer + 1)
    .doOnNext(System.out::println)
    .blockLast(); // 等待最后一个元素处理完成
System.out.println("after");

2. 用then()编排后续逻辑：把收尾操作作为Flux的一部分，避免手动阻塞：

Flux.range(1, 1000)
    .map(integer -> integer + 1)
    .doOnNext(System.out::println)
    .then(Mono.fromRunnable(() -> System.out.println("after")))
    .subscribe();

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