为何在reactor-http-nio-*线程内Mono.toFuture().get()无法执行？

我需要实现阻塞调用，为了避免IllegalStateException: block/blockFirst/blockLast异常，改用.toFuture().get()替代.block()，但发现在reactor-http-nio-*线程执行的Mono内部调用这个方法时，请求始终无法执行。

测试代码如下：

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import okhttp3.mockwebserver.Dispatcher;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class ReactiveTest {

  static MockServer server = new MockServer();

  public static void main(String[] args) throws InterruptedException, IOException {

    server.setUp();
    WebClient.create("http://localhost:1234/primary")
        .get()
        .exchangeToMono(x -> x.bodyToMono(String.class))
        .subscribe(
            response -> {
              System.out.println("First Mono Executing Thread: " + Thread.currentThread());
              try {
                CompletableFuture<String> res =
                    WebClient.create("http://localhost:1234/secondary")
                        .get()
                        .exchangeToMono(x -> x.bodyToMono(String.class))
                        .log()
                        .map(
                            x -> {
                              System.out.println("Second Mono Executing Thread: " + Thread.currentThread());
                              return x;
                            })
                        .toFuture();
                System.out.println("Second Mono with .toFuture().get() is never executed!: " + res.get());
              } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
              }
            });
    Thread.sleep(10000);
    server.tearDown();
    System.out.println("Main ended! But Second Mono with .toFuture().get() is never executed after waiting for 10 seconds!");
  }

  static class MockServer {
    final Dispatcher dispatcher =
        new Dispatcher() {

          @Override
          public MockResponse dispatch(RecordedRequest request) throws InterruptedException {

            switch (request.getPath()) {
              case "/primary":
                return new MockResponse().setResponseCode(200).setBody("{\"hello\":\"primary\"}");
              case "/secondary":
                return new MockResponse().setResponseCode(200).setBody("{\"hello\":\"secondary\"}");
            }
            return new MockResponse().setResponseCode(404);
          }
        };


    MockWebServer server;
    void setUp() {
      server = new MockWebServer();
      server.setDispatcher(dispatcher);
      try {
        server.start(1234);
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }

    void tearDown() throws IOException {
      server.shutdown();
    }
  }
}

依赖配置：

<dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>okhttp</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>mockwebserver</artifactId>
  <scope>test</scope>
</dependency>

原因分析 #

这是线程死锁导致的：

• reactor-http-nio-*是Reactor HTTP客户端的事件循环线程，这类线程专为非阻塞IO设计，绝对不能被阻塞——一旦阻塞会占用事件循环资源，甚至导致整个线程池瘫痪。
• 当你在该线程中调用CompletableFuture.get()时，会直接阻塞当前的reactor-http-nio线程；而第二个WebClient的请求又依赖这个线程池处理IO事件，于是形成了线程等待自身释放资源的死锁，请求自然永远无法完成。
• 主线程不属于事件循环线程，直接调用toFuture().get()时，第二个请求的IO操作可以正常由reactor-http-nio线程处理，因此不会触发死锁。

解决方案 #

要避免死锁，必须让阻塞操作脱离事件循环线程：

• 在第二个Mono上添加.subscribeOn(Schedulers.parallel())，将订阅逻辑切换到独立的并行线程池。这样get()阻塞的是并行线程，reactor-http-nio线程可以正常处理第二个请求的IO操作，不会出现死锁。
• 在Spring响应式系统线程（如org.springframework.statemachine相关线程）中，推荐使用Mono.subscribeOn(Schedulers.parallel()).toFuture().get()作为安全的阻塞调用方式，既避开了直接调用.block()抛出的异常，又不会阻塞事件循环线程。

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