java - 项目 Reactor 中的 Parallel Flux 与 Flux

Question

所以我从文档中了解到，并行 Flux 本质上是将通量元素划分为单独的轨道。(本质上类似于分组)。就线程而言，这将是调度程序的工作。因此，让我们考虑这样的情况。所有这些都将在通过 runOn() 方法提供的同一个调度程序实例上运行。
让我们考虑如下情况:

Mono<Response> = webClientCallAPi(..) //function returning Mono from webclient call

现在假设我们打了大约 100 个电话

Flux.range(0,100).subscribeOn(Schedulers.boundedElastic()).flatMap(i -> webClientCallApi(i)).collecttoList() // or subscribe somehow

如果我们使用paralleFlux:

Flux.range(0,100).parallel().runOn(Schedulers.boundedElastic()).flatMap(i -> webClientCallApi(i)).sequential().collecttoList();

所以如果我的理解是正确的，它看起来几乎是相似的。那么 ParallelFlux 与 Flux 相比有哪些优势，什么时候应该使用parallelFlux 而不是flux？

Answer 1

在实践中，您可能很少需要使用并行通量，包括在本例中。
在您的示例中，您将触发 100 个 Web 服务调用。请记住，执行此操作所需的实际工作非常少——您生成并触发一个异步请求，然后一段时间后您会收到一个响应。在请求和响应之间，您根本没有做任何工作，在发送每个请求时只需要少量 CPU 资源，而在收到每个响应时只需要少量 CPU 资源。 (这是使用异步框架发出 Web 请求的核心优势之一，在请求进行中时您不会占用任何线程。)
如果您拆分此通量并并行运行它，则表示您希望拆分这些微量的 CPU 资源，以便它们可以在不同的 CPU 内核上同时运行。这绝对没有意义 - 拆分通量，并行运行它然后稍后组合它的开销将比仅仅让它在正常的顺序调度程序上执行要大得多。
另一方面，假设我有一个 Flux<Integer>例如，我想检查这些整数中的每一个是否都是素数 - 或者可能是 Flux<String>我想根据 BCrypt 哈希检查的密码。这些类型的操作确实是 CPU 密集型的，因此在这种情况下，用于跨内核拆分执行的并行通量可能很有意义。但实际上，这些情况在正常的 react 堆用例中很少发生。
(此外，作为结束语，您几乎总是想使用 Schedulers.parallel() 与平行通量，而不是 Schedulers.boundedElastic() 。)