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

Question

我知道subscribeOn在序列被subscribe时会切换执行线程，但我发现它不能与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

但让我困惑的是

@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());
    }
}

subscribeOn

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或Publishon()上包含单独的subscribeOn()可以完成我想要的工作

Answer 1

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

但是，您可以在链中包含flatmap()调用，您可以为此指定一个单独的subscribeon()，该调用不会被覆盖。根据您的用例，这可能是一个选项，因为您可以在flatmap()调用中定义的发布服务器中完成大部分工作。