Reactor中Mono顺序执行报错：block()阻塞操作不被支持

场景与问题 #

现有两个响应式函数：

Mono<Void> messageEvent(MessageEvent messageEvent);
Mono<Victim> getPersonById(String personId);

需求：messageEvent执行成功后，调用getPersonById，并将返回的Victim存入数据库。

尝试代码：

spi.messageEvent(message).doOnSuccess(it -> {
    spi.getPersonById(evt.getVictimId()).doOnSuccess(victim -> {
        repository.save(victim);
    }).block();
}).block();

触发错误：

2023-03-09 21:56:20.191 ERROR 21080 --- [nister-group]-0]
c.s.e.v.s.q.i : on()方法中出现异常，原因='NULL'，异常信息='block()/blockFirst()/blockLast()是阻塞操作，在reactor-http-nio-2线程中不被支持'
java.lang.IllegalStateException: block()/blockFirst()/blockLast()是阻塞操作，在reactor-http-nio-2线程中不被支持
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83)

错误原因 #

你在Reactor的非阻塞IO线程（如reactor-http-nio-2）中调用了block()，违反了响应式编程的非阻塞原则：

• IO线程是Reactor专门处理非阻塞任务的资源，数量有限
• 阻塞调用会占用线程，导致线程池耗尽，服务无法处理后续请求

正确实现 #

情况1：数据库操作是响应式的（如Spring Data Reactive） #

用响应式操作符链式调用，完全避免阻塞：

spi.messageEvent(message)
    // 等待messageEvent执行完成，再触发getPersonById
    .then(spi.getPersonById(evt.getVictimId()))
    // 将查询结果存入数据库
    .flatMap(victim -> repository.save(victim));

情况2：数据库操作是阻塞的（如Spring Data JPA） #

需将阻塞操作包装到弹性线程池执行，避免阻塞IO线程：

import reactor.core.scheduler.Schedulers;

spi.messageEvent(message)
    .then(spi.getPersonById(evt.getVictimId()))
    // 用fromCallable包装阻塞的save方法，指定弹性线程池执行
    .flatMap(victim -> Mono.fromCallable(() -> repository.save(victim))
                            .subscribeOn(Schedulers.boundedElastic()));

注意事项 #

如果这段代码用于WebFlux控制器，直接返回该Mono对象即可，框架会自动处理订阅和线程调度，绝对不要手动调用block()。

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