Spring Boot 3.0.2中如何在Reactor线程内阻塞实现事件校验？

使用Spring Boot 3.0.2，需要在Reactor线程内完成事件校验，校验不通过则终止后续逻辑，但直接调用.block()时抛出如下异常：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-kqueue-7
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83) ~[reactor-core-3.5.2.jar:3.5.2]
at reactor.core.publisher.Mono.block(Mono.java:1710) ~[reactor-core-3.5.2.jar:3.5.2]

现有处理代码：

Flux.fromIterable(messages).map((message) -> {
            try {
                return this.readValue(message);
            } catch (Exception var3) {
                throw new RuntimeException(var3);
            }
        }).flatMap(this::process).subscribeOn(Schedulers.boundedElastic()).doOnError((error) -> {
            this.messageErrorHandler.handle(error);
        }).blockLast(Duration.ofMillis(Long.parseLong((String)this.properties.get("blockLast.timeout.ms"))));

校验逻辑在process()方法内：

eventValidator.validate(eventPayload);

其中validate方法发起IO调用，返回Mono<Data>类型结果：

Mono<Data> data = repo.getData();

需要实现获取校验结果并完成校验，同时避免上述异常。

核心原则是不要在Reactor的非阻塞线程中调用阻塞方法，而是将校验的Mono整合到反应式流的链式调用中，利用Reactor操作符实现校验逻辑，而非阻塞获取结果。

1. 修改process方法，用反应式方式串联校验与后续逻辑 #

将原来的阻塞获取逻辑改为通过flatMap操作符串联校验流程，在校验不通过时抛出异常，自然终止后续处理：

public Mono<Void> process(EventPayload eventPayload) {
    // 串联校验逻辑与后续业务处理
    return eventValidator.validate(eventPayload)
            .flatMap(data -> {
                // 执行校验判断，不通过则抛出异常
                if (!checkValidation(data)) {
                    return Mono.error(new IllegalStateException("事件校验不通过"));
                }
                // 校验通过，执行后续业务逻辑
                return executeBusinessLogic(eventPayload);
            });
}

2. 原理说明 #

Reactor中标记为非阻塞的线程（如reactor-http-*）严格禁止调用block()类阻塞方法，这会破坏反应式框架的非阻塞模型。通过将校验的Mono融入反应式流，整个流程保持非阻塞特性，既避免了异常，又能在校验失败时通过Mono.error()终止后续逻辑，完全匹配需求。

3. 特殊场景备选方案（不推荐） #

如果因历史代码限制无法重构为纯反应式逻辑，需确保阻塞操作在支持阻塞的线程池中执行（如Schedulers.boundedElastic()），可以在校验Mono上指定线程池后再调用block()：

public Mono<Void> process(EventPayload eventPayload) {
    // 切换到支持阻塞的线程池执行校验
    Data data = eventValidator.validate(eventPayload)
            .subscribeOn(Schedulers.boundedElastic())
            .block(); // 此时阻塞不会触发异常
    
    if (!checkValidation(data)) {
        throw new IllegalStateException("事件校验不通过");
    }
    return executeBusinessLogic(eventPayload);
}

注意：此方案会打破反应式流的非阻塞特性，仅建议在极端场景下临时使用。

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