WebTestClient 返回 IllegalStateException: block()/blockFirst()/blockLast() 是阻塞的，线程不支持

Question

如下函数：

private Boolean canDoIt(Parameter param) {
  return myService
      .getMyObjectInReactiveWay(param)
      .map(myObject -> myService.checkMyObjectInImperativeWay(myObject))
      .block();
}

在运行时工作正常，但是在使用 WebTestClient 测试使用它的流时，我收到以下错误：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-1
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83) ~[reactor-core-3.4.1.jar:3.4.1]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoFlatMap] :
    reactor.core.publisher.Mono.flatMap

我知道我不应该使用block()，但我别无选择：该函数必须返回Boolean（而不是Mono<Boolean>）。也许有另一种不使用block()的编写方式。

有没有办法让WebTestClient 不抛出该错误？

使用 Reactor Core 版本3.4.6。

Answer 1

假设您无法修改 checkMyObjectInImperativeWay 以返回 Mono：

   private Boolean canDoIt(Parameter param) {
        final AtomicBoolean result= new AtomicBoolean();
        myService.getMyObjectInReactiveWay(param)
                .map(myObject -> myService.checkMyObjectInImperativeWay(myObject))
                .subscribe((mono) -> result.set(mono));
        return result.get();
    }

Answer 2

我验证我的评论。 block() 检查调用线程是否与阻塞代码兼容（反应器外部的线程，或特定反应器调度程序的线程，如Schedulers.boundedElastic()）。

有两种方法可以在响应式堆栈中间处理阻塞调用：

• 在您将阻止的发布者上使用 share 运算符。请注意，共享运算符会在内部缓存该值。
• 使用scheduleOn 或publishOn 在阻塞兼容调度程序上强制执行block() 调用。请注意，不应在直接调用 block() 的发布者上调用此运算符，而应在将“包装”块调用的发布者上调用此运算符（参见下面的示例）。

一些参考资料：

• The schedulers API doc 总结了可用的调度程序类型，并说明哪些是为阻塞调用而设计的。
• Reference guide section about blocking call wrapping
• share() operator documentation

还有一个最小的可重现示例（在 v3.4.6 上测试）给出了这个输出：

Ok context: not running from reactor Threads
value is true
Problematic stack: working with scheduler not compatible with blocking call
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-2
Bad way to subscribe on a blocking compatible scheduler
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-4
Bad way to publish on blocking compatible scheduler
ERROR: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-6
Possible workaround: share the reactive stream before blocking on it
It worked
Right way to subscribe on blocking compatible scheduler
It worked
Right way to publish on blocking compatible scheduler
It worked

代码来了：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.Callable;
import java.util.function.Supplier;

public class BlockingWorkaround {

    public static void main(String[] args) throws Exception {
        System.out.println("Ok context: not running from reactor Threads");
        System.out.println("value is "+blockingFunction());

        System.out.println("Problematic stack: working with scheduler not compatible with blocking call");
        executeAndWait(() -> blockingFunction());

        System.out.println("Bad way to subscribe on a blocking compatible scheduler");
        executeAndWait(() -> blockingFunctionUsingSubscribeOn());

        System.out.println("Bad way to publish on blocking compatible scheduler");
        executeAndWait(() -> blockingFunctionUsingPublishOn());

        System.out.println("Possible workaround: share the reactive stream before blocking on it");
        executeAndWait(() -> blockingFunctionShared());

        System.out.println("Right way to subscribe on blocking compatible scheduler");
        subscribeOnAndWait(() -> blockingFunction());

        System.out.println("Right way to publish on blocking compatible scheduler");
        publishOnAndWait(() -> blockingFunction());
    }

    static Boolean blockingFunction() {
        return delay()
                .flatMap(delay -> Mono.just(true))
                .block();
    }

    static Boolean blockingFunctionShared() {
        return delay()
                .flatMap(delay -> Mono.just(true))
                .share() // Mono result is cached internally
                .block();
    }

    static Boolean blockingFunctionUsingSubscribeOn() {
        return delay()
                .subscribeOn(Schedulers.boundedElastic())
                .flatMap(delay -> Mono.just(true))
                .block();
    }

    static Boolean blockingFunctionUsingPublishOn() {
        return delay()
                .flatMap(delay -> Mono.just(true))
                .publishOn(Schedulers.boundedElastic())
                .block();
    }

    static Mono<Long> delay() {
        return Mono.delay(Duration.ofMillis(10));
    }

    private static void executeAndWait(Supplier<Boolean> blockingAction) throws InterruptedException {
        delay()
                .map(it -> blockingAction.get())
                .subscribe(
                        val -> System.out.println("It worked"),
                        err -> System.out.println("ERROR: " + err.getMessage())
                );

        Thread.sleep(100);
    }

    private static void subscribeOnAndWait(Callable<Boolean> blockingAction) throws InterruptedException {
        final Mono<Boolean> blockingMono = Mono.fromCallable(blockingAction)
                .subscribeOn(Schedulers.boundedElastic()); // Upstream is executed on given scheduler

        delay()
                .flatMap(it -> blockingMono)
                .subscribe(
                        val -> System.out.println("It worked"),
                        err -> System.out.println("ERROR: " + err.getMessage())
                );

        Thread.sleep(100);
    }

    private static void publishOnAndWait(Supplier<Boolean> blockingAction) throws InterruptedException {
        delay()
                .publishOn(Schedulers.boundedElastic()) // Cause downstream to be executed on given scheduler
                .map(it -> blockingAction.get())
                .subscribe(
                        val -> System.out.println("It worked"),
                        err -> System.out.println("ERROR: " + err.getMessage())
                );

        Thread.sleep(100);
    }
}