subscribeOn 在 ServerRequest 的 bodyToMono 上使用时不起作用

Question

我知道 subscribeOn 用于在订阅序列时切换执行线程，但我发现它不适用于 ServerRequest.bodyToMono/Flux就像是Flux.just(1,2,3)          .doOnNext(integer -> log.info("test {}",integer))          .subscribeOn(Schedulers.elastic())          .subscribe();会使执行线程改变INFO 23313 --- [      elastic-2] c.a.p.m.f.service.router.TestService     : test 1INFO 23313 --- [      elastic-2] c.a.p.m.f.service.router.TestService     : test 2INFO 23313 --- [      elastic-2] c.a.p.m.f.service.router.TestService     : test 3但令我困惑的是假设我有一个 Spring WebFlux 路由器：@Configurationpublic class TestRouter {    @Bean    public RouterFunction<ServerResponse> testRouterFunction(TestService testService) {        return route().path("/test", builder -> builder.nest(accept(MediaType.ALL),          route -> route.PUT("/", req -> {              Mono<String> valueMono = req.bodyToMono(String.class);              return ServerResponse.ok().body(testService.test(valueMono), String.class);          }))).build();    }}和一个服务：@Service@Slf4jpublic class TestService {    public Mono<String> test(Mono<String> mono) {        return mono          .doOnSubscribe(subscription -> log.info("on subscribe"))          .subscribeOn(Schedulers.elastic())          .doOnNext(s -> log.info("received {}", s))          .subscribeOn(Schedulers.elastic());    }}基本逻辑是 http put 请求到 localhost:port/test 将接收以纯文本形式发送到服务器的内容我尝试让 doOnNext 在其他线程而不是 Spring WebFlux 的 NIO 线程上运行，无论我放在哪里subscribeOn执行线程始终是 NIO 线程：INFO 23200 --- [ctor-http-nio-4] c.a.p.m.f.service.router.TestService     : on subscribeINFO 23200 --- [ctor-http-nio-4] c.a.p.m.f.service.router.TestService     : received test感谢@MichaelBerry @SimonBaslé，你们俩都帮了我很多忙，对你们的答案都投赞成票简而言之，reactor-netty 将覆盖 http 订阅的 subscribeOn，使用 aflatMap()包含一个单独的subscribeOn()on 不同的Mono/Flux或publishOn()可以完成我想要的工作

Answer 1

subscribeOn()这不是您可以更改的内容 - 它只是调用之前调用的链中的最后一个调用subscribe()，因此由 WebFlux 决定使用它想要的任何调度程序。在本例中，它看起来像是在 NIO 驱动的事件循环或类似循环中处理请求。

但是，您可以flatMap()在链中包含一个调用，并为其指定subscribeOn()一个不会被覆盖的单独调用。这可能是一个选项，具体取决于您的用例，因为您可以在调用中定义的发布者中完成大部分工作flatMap()。