WebTestClient 返回 IllegalStateException：block()/blockFirst()/blockLast() 正在阻塞，线程不支持

问题描述

以下函数：

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。

解决方法

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

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

• 在您将阻止的发布商上使用 share 运算符。请注意，共享运算符会在内部缓存该值。
• 使用 block() 或 scheduleOn 强制在阻塞兼容调度程序上执行 publishOn 调用。请注意，不应在直接调用 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);
    }
}

,

假设您不能修改 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();
    }