我验证我的评论。block()检查调用线程是否与阻塞代码兼容（反应器外部的线程，或特定反应器调度程序的线程，如Schedulers.boundedElastic()）。
有两种方法可以在反应堆栈中间处理阻塞调用：
- 在您将阻止的发布者上使用共享运算符。请注意，共享运算符会在内部缓存该值。
- 使用或强制
block()在阻塞兼容调度程序上执行调用。请注意，不应在直接调用 的发布者上调用此运算符，而应在将“包装”块调用的发布者上调用此运算符（请参见下面的示例）。scheduleOnpublishOnblock()
一些参考资料：
还有一个最小的可重现示例（在 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);
}
}