Reactor线程中block()调用报错，需返回实体对象而非Mono的解决方案

问题根源是在Reactor的非阻塞IO线程（reactor-http-nio-3）中调用block()会阻塞事件循环，破坏响应式非阻塞特性，因此需要将阻塞操作转移到专门的阻塞线程池中执行。以下是具体实现：

1. 使用Reactor的boundedElastic调度器 #

Reactor提供了Schedulers.boundedElastic()，专门用于处理阻塞操作，它会创建有界的线程池，避免线程资源耗尽。

修改后的callmyOrderSvc方法 #

private MyOrderResponse callmyOrderSvc(String path, Log log) {
    return WebClient.create().get()
            .uri(myOrderSvcUrl + OrderConstant.SEP + path)
            .header(OrderConstant.SECRETKEY, apiKey)
            .retrieve()
            .onStatus(HttpStatusCode::isError, response -> 
                    response.bodyToMono(MyOrderResponse.class)
                            .flatMap(error -> {
                                LogUtil.logError(logger, "Error occurred", error.getMessage());
                                // 注意：HttpStatus需要传入具体的错误状态码，比如HttpStatus.BAD_REQUEST
                                return Mono.error(new CustomException(error.getMessage(), log.getId(), HttpStatus.BAD_REQUEST));
                            })
            )
            .bodyToMono(MyOrderResponse.class)
            // 将整个操作切换到boundedElastic线程池执行
            .subscribeOn(Schedulers.boundedElastic())
            .block();
}

2. 自定义阻塞线程池（可选，更可控） #

如果需要更精细地控制线程池参数（比如核心线程数、队列大小），可以自定义Reactor调度器：

// 在配置类中定义自定义调度器
@Bean
public Scheduler customBlockingScheduler() {
    // 参数：核心线程数，最大线程数，线程池名称
    return Schedulers.newBoundedElastic(10, 100, "order-service-blocking-pool");
}

然后在方法中注入使用：

@Autowired
private Scheduler customBlockingScheduler;

private MyOrderResponse callmyOrderSvc(String path, Log log) {
    return WebClient.create().get()
            .uri(myOrderSvcUrl + OrderConstant.SEP + path)
            .header(OrderConstant.SECRETKEY, apiKey)
            .retrieve()
            .onStatus(HttpStatusCode::isError, response -> 
                    response.bodyToMono(MyOrderResponse.class)
                            .flatMap(error -> {
                                LogUtil.logError(logger, "Error occurred", error.getMessage());
                                return Mono.error(new CustomException(error.getMessage(), log.getId(), HttpStatus.BAD_REQUEST));
                            })
            )
            .bodyToMono(MyOrderResponse.class)
            .subscribeOn(customBlockingScheduler)
            .block();
}

关键说明 #

• subscribeOn(Schedulers.boundedElastic())会将WebClient的请求和后续的block()操作转移到专门的阻塞线程，不会占用Reactor的IO线程，从而避免错误。
• 不要使用Schedulers.elastic()，它没有线程数上限，可能导致线程泛滥耗尽资源，boundedElastic是更安全的替代方案。
• 虽然此方法满足“必须返回实体对象”的要求，但在响应式环境中应尽量避免使用block()，建议业务层面逐步向响应式模型迁移。

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