SubscribeOn或PublishOn不使用ReactiveCassandraCrudRepository

Question

我们在Spring WebFlux中使用反应性Spring数据存储库，我对SubscribeOn的理解是，它决定SubscribeOn之前的操作符将在flux中执行哪个线程池，而PublishOn则决定订阅将在哪个线程池上执行。然而，在下面的代码中，即使使用PublishOn和SubscribeOn，代码也不会在主线程上执行，而是返回到cluster-nio-worker-1。

    System.out.println("Current Thread :- "+Thread.currentThread().getName()); //Current Thread :- main
    personRepository.findAll().log()
            .map(document -> mapDocumentToSomethingElse(document)) //Current thread cluster-nio-worker-1
            .subscribeOn(Schedulers.immediate())
            .publishOn(Schedulers.immediate())
            .subscribe(trackingevent -> System.out.println("Got Item "+item +" inside thread "+Thread.currentThread()), //Thread[cluster-nio-worker-1,5,main]
                    excp -> excp.printStackTrace(),
                    () -> System.out.println("Completed processing Thread:- "+Thread.currentThread().getName())); //cluster-nio-worker-1

另外，thread[cluster-nio-worker-1，5，main]是什么意思？为什么这些方法调用不使用主线程执行。

Answer 1

subscribeOn方法使发布服务器使用给定的线程池来发布值。管道中可能有N个subscribeon方法。最衣着的就会生效。PersonRepository.findAll().log()是一个包装器，返回一个流量。因此，如果它在内部使用任何调度程序，那么您就不能使用subscribeon更改它。例如，interval方法使用parallel并且我不能将其更改为boundedElastic（如图所示）。

    Flux.interval(Duration.ofSeconds(1))
            .subscribeOn(Schedulers.boundedElastic())

schedulers.immediate只是将管道执行保持在同一线程中。它不是主的，在您的示例中，它将是cluster-nio-worker线程池。

我们可以从主线程切换到任何调度器线程池。但是我们不能将执行切换回主线程。它不是工程Reactor的限制。它应该是Java本身的一个限制。