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

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 1
INFO 23313 --- [      elastic-2] c.a.p.m.f.service.router.TestService     : test 2
INFO 23313 --- [      elastic-2] c.a.p.m.f.service.router.TestService     : test 3

但让我困惑的是

假设我有一个 Spring WebFlux 路由器:

@Configuration
public 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
@Slf4j
public 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 将请求发送到 localhost:port/test 将接收它以纯文本形式发送到服务器的内容

我尝试让 doOnNext 在其他线程上运行，而不是在 Spring WebFlux 的 NIO 线程上运行，无论我放在哪里

subscribeOn

执行线程总是NIO线程:

INFO 23200 --- [ctor-http-nio-4] c.a.p.m.f.service.router.TestService     : on subscribe
INFO 23200 --- [ctor-http-nio-4] c.a.p.m.f.service.router.TestService     : received test

---

感谢@MichaelBerry @SimonBaslé，你们帮了我很多，给你们的答案都点赞

简而言之，reactor-netty 将为 http 订阅覆盖 subscribeOn，使用 flatMap() 在不同的 Mono/Flux 上包含一个单独的 subscribeOn() 或 publishOn() 可以完成我想要的工作

Answer 1

这不是您可以更改的东西 - 它只是在 subscribe() 被调用之前链中的最后一个 subscribeOn() 调用是有效的，所以它取决于WebFlux 使用它想要的任何调度程序。在这种情况下，它看起来像是在 NIO 驱动的事件循环或类似事件中处理请求。

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