应该并行 API 调用使用 Schedulers.parallel() 或 Schedulers.boundedElastic()

Question

老实说，我不知道调度器在 reactor 中是如何工作的。我读过一些，这是我发现的。

Schedulers.parallel() is good for CPU-intensive but short-lived tasks. It can execute N such tasks in parallel (by default N == number of CPUs)

Schedulers.elastic() and Schedulers.boundedElastic() are good for more long-lived tasks (eg. blocking IO tasks). The elastic one spawns threads on-demand without a limit while the recently introduced boundedElastic does the same with a ceiling on the number of created threads.

所以在我的 API 调用中有一个任务，我必须一遍又一遍地轮询请求，直到它的状态准备就绪。

Flux.just(order1, order2)
    .parallel(4)
    .runOn(Schedulers.parallel())
    .flatMap { order -> createOrder(order) }
    .flatMap { orderId -> 
       pollConfirmOrderStatus(orderId)
          .retryWhen(notReady)
    }
    .collectList()
    .subscribe()

如您所见，我使用 Schedulers.parallel() 并且工作正常，但我担心阻止 CPU 使用，因为我的服务器没有那么多 CPU核心。 pollConfirmOrderStatus 大约需要 1-2 分钟，所以我不确定它是否会阻止我服务器中的其他进程访问 CPU。那么我应该在这里使用 Schedulers.parallel() 还是 Schedulers.bondedElastic()。

Answer 1

如果您的方法 pollConfirmOrderStatus() 不阻塞并行调度程序的线程，那应该没问题。否则，您可能会阻塞并行调度程序中的所有可用线程，如果您的状态从未就绪，这可能会导致死锁。

在这里，它解释了并行调度程序是为 non-blocking 调用保留的，并且您可以使用 BlockHound 来发现来自 non-blocking 预期线程的阻塞调用。 https://spring.io/blog/2019/03/28/reactor-debugging-experience

应该并行 API 调用使用 Schedulers.parallel() 或 Schedulers.boundedElastic()

Should parallel API call use Schedulers.parallel() or Schedulers.boundedElastic()

reactor

kotlin