Project Reactor 的 Flux 能否一一处理消息答案

【问题标题】：Can Flux of Project Reactor process messages one by oneProject Reactor 的 Flux 能否一一处理消息
【发布时间】：2020-10-31 08:36:44
【问题描述】：

我正在尝试使用 Reactor Flux 一个一个地处理一个数字列表，例如 1 到 10，并且有一个 API /double 可以简单地将传入的 加倍Integer (1 -> 2, 4 -> 8...) ，但是这个API有性能问题，总是需要2秒才能响应结果。当使用limitRate(1) 时，我期望的是 Reactor 会一个接一个地处理请求，如下所示：

2020-01-01 00:00:02 - 2
2020-01-01 00:00:04 - 4
2020-01-01 00:00:06 - 6
2020-01-01 00:00:08 - 8
2020-01-01 00:00:10 - 10
...

但实际上 Reactor 会同时触发所有请求：

2020-01-01 00:00:02 - 6
2020-01-01 00:00:02 - 10
2020-01-01 00:00:02 - 2
2020-01-01 00:00:02 - 4
2020-01-01 00:00:02 - 8
...

这里是代码

Flux.range(1, 10).limitRate(1)
                .flatMap(i -> webClient.get().uri("http://localhost:10001/double?integer={int}", i).exchange()
                        .flatMap(resp -> resp.bodyToMono(Integer.class)))
                .subscribe(System.out::println);
Thread.sleep(10000);

似乎limitRate 没有像我预期的那样工作，出了什么问题？有没有办法使用 Reactor 一个接一个地处理请求？提前致谢。

【问题讨论】：

标签： project-reactor reactor

【解决方案1】：

.flatMap 在这里不起作用，因为它急切地订阅内部流 - 也就是说，它不会在订阅下一个流之前等待内部流发出 onComplete。这就是为什么您的所有呼叫都是同时进行的。它在receive->dispatch->receive->dispatch 模式下工作。

Reactor 提供了flatMap 的重载版本，您可以在其中将并发因子指定为.flatMap(innerstream, concurrency)。这个因素限制了 flatMap 将订阅的流的数量。如果是 5，flatMap 最多可以订阅 5 个内部流。一旦达到此限制，它必须等待内部流发出onComplete，然后再订阅下一个。

在您的情况下，您可以将其设置为 1 或使用 .concatMap()。 concatMap() 与 concurrency = 1 完全一致。它基本上可以在receive->dispatch->wait->receive->dispatch->wait 模式下工作。

前段时间我写了一篇文章，解释了flatMap 的工作原理，因为我认为很多人在使用它时并不了解它的内部结构。可以参考文章here

【讨论】：

【解决方案2】：

考虑改用concatMap：

/**
 * Transform the elements emitted by this {@link Flux} asynchronously into Publishers,
 * then flatten these inner publishers into a single {@link Flux}, sequentially and
 * preserving order using concatenation.
 * <p>
 * There are three dimensions to this operator that can be compared with
 * {@link #flatMap(Function) flatMap} and {@link #flatMapSequential(Function) flatMapSequential}:
 * <ul>
 *     <li><b>Generation of inners and subscription</b>: this operator waits for one
 *     inner to complete before generating the next one and subscribing to it.</li>
 *     <li><b>Ordering of the flattened values</b>: this operator naturally preserves
 *     the same order as the source elements, concatenating the inners from each source
 *     element sequentially.</li>
 *     <li><b>Interleaving</b>: this operator does not let values from different inners
 *     interleave (concatenation).</li>
 * </ul>
 *
 * <p>
 * Errors will immediately short circuit current concat backlog.
 *
 * <p>
 * <img class="marble" src="doc-files/marbles/concatMap.svg" alt="">
 *
 * @reactor.discard This operator discards elements it internally queued for backpressure upon cancellation.
 *
 * @param mapper the function to transform this sequence of T into concatenated sequences of V
 * @param <V> the produced concatenated type
 *
 * @return a concatenated {@link Flux}
 */
public final <V> Flux<V> concatMap(Function<? super T, ? extends Publisher<? extends V>>
        mapper) {

注意sequentially and preserving order using concatenation. 短语。对我来说似乎是你在寻找什么。

【讨论】：

【解决方案3】：

受 Artem Bilan 的回答启发，我发现 flatMapSequential 更适合我的情况，因为 flatMapSequential 接受第二个参数为 maxConcurrency，因此可以不逐个处理消息，而是一次处理两次等等。感谢 Artem Bilan 和 Prashant Pandey 的回答，真的很有帮助。

【讨论】：

flatMap 也接受第二个参数，大多数情况下不需要顺序