我试图理解响应式编程是如何真正工作的。为此，我准备了简单的演示：来自Spring Framework的reactiveWebClient向简单的rest api发送请求，该客户端在每次操作中打印线程名称。

rest api：

@RestController
@SpringBootApplication
public class RestApiApplication {

    public static void main(String[] args) {
        SpringApplication.run(RestApiApplication.class, args);
    }

    @PostMapping("/resource")
    public void consumeResource(@RequestBody Resource resource) {
        System.out.println(String.format("consumed resource: %s", resource.toString()));
    }
}

@Data
@AllArgsConstructor
class Resource {
    private final Long id;
    private final String name;
}

和最重要的-响应式Web客户端：

@SpringBootApplication
public class ReactorWebclientApplication {

    public static void main(String[] args) {
        SpringApplication.run(ReactorWebclientApplication.class, args);
    }

    private final TcpClient tcpClient = TcpClient.create();

    private final WebClient webClient = WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))
        .baseUrl("http://localhost:8080")
        .build();

    @PostConstruct
    void doRequests() {
        var longs = LongStream.range(1L, 10_000L)
            .boxed()
            .toArray(Long[]::new);

        var longsStream = Stream.of(longs);

        Flux.fromStream(longsStream)
            .map(l -> {
                System.out.println(String.format("------- map [%s] --------", Thread.currentThread().getName()));
                return new Resource(l, String.format("name %s", l));
            })
            .filter(res -> {
                System.out.println(String.format("------- filter [%s] --------", Thread.currentThread().getName()));
                return !res.getId().equals(11_000L);
            })
            .flatMap(res -> {
                System.out.println(String.format("------- flatmap [%s] --------", Thread.currentThread().getName()));
                return webClient.post()
                    .uri("/resource")
                    .syncBody(res)
                    .header("Content-Type", "application/json")
                    .header("Accept", "application/json")
                    .retrieve()
                    .bodyToMono(Resource.class)
                    .doOnSuccess(ignore -> System.out.println(String.format("------- onsuccess [%s] --------", Thread.currentThread().getName())))
                    .doOnError(ignore -> System.out.println(String.format("------- onerror [%s] --------", Thread.currentThread().getName())));
            })
            .blockLast();
    }

}

@JsonIgnoreProperties(ignoreUnknown = true)
class Resource {
    private final Long id;
    private final String name;

    @JsonCreator
    Resource(@JsonProperty("id") Long id, @JsonProperty("name")  String name) {
        this.id = id;
        this.name = name;
    }

    Long getId() {
        return id;
    }

    String getName() {
        return name;
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("Resource{");
        sb.append("id=").append(id);
        sb.append(", name='").append(name).append('\'');
        sb.append('}');
        return sb.toString();
    }
}

问题是行为与我预测的不同。

我期望. map（）、.filter（）和.flitMap（）的每个调用都将在main线程上执行，并且.doOn成功（）或.doOnError的每个调用都将在nio线程池的线程上执行。所以我期望日志如下所示：

------- map [main] --------
------- filter [main] --------
------- flatmap [main] --------
(and so on...)
------- onsuccess [reactor-http-nio-2] --------
(and so on...)

但我得到的日志是：

------- map [main] --------
------- filter [main] --------
------- flatmap [main] --------
------- map [main] --------
------- filter [main] --------
------- flatmap [main] --------
------- onsuccess [reactor-http-nio-2] --------
------- onsuccess [reactor-http-nio-6] --------
------- onsuccess [reactor-http-nio-4] --------
------- onsuccess [reactor-http-nio-8] --------
------- map [reactor-http-nio-2] --------
------- filter [reactor-http-nio-2] --------
------- flatmap [reactor-http-nio-2] --------
------- map [reactor-http-nio-2] --------

接下来登录. map（）、.filter（）和.plitMap（）都是在reactor-超文本传输协议-nio的线程上完成的。

接下来不可理解的事实是，在主线程执行的操作和reactor-超文本传输协议-nio之间的比率总是不同的，有时所有的操作. map（），.filter（）和.plitMap（）都是在主线程执行的。