反应堆项目中的条件并行答案

【问题标题】：Conditional parallelism in reactor project反应堆项目中的条件并行
【发布时间】：2022-06-11 18:29:25
【问题描述】：

有一个流必须基于特定字段并行解析消息。

使用groupBy 不是一种选择，因为它会收集大量结果，然后对它们进行分组（它会引入延迟）。另外，如果使用subscribeOn(Schedulers.single())，结果还可以，但是会引入饥饿问题。

例如，具有相同batchId 的订单应按顺序执行。因此，订单 1 和订单 2 应该顺序处理，订单 3 可以并行执行。

    record Order(Integer id,Integer batchId){}
    void testParallel() {
        Flux.just(new Order(1,1),new Order(2,1),new Order(3,2));
    }

【问题讨论】：

标签： parallel-processing spring-webflux project-reactor

【解决方案1】：

您应该能够创建 2 个单独的 Flux 实例。一个顺序处理每个发出的项目，另一个并行处理它们。然后，您可以将 merge 合并为一个 Flux。

类似的东西

        Flux.just(1, 2, 3, 4, 5, 6)
                .collectList()
                .flatMapMany(list -> {
                    Stream<Integer> sequential = list.stream().filter(i -> i < 4);
                    Stream<Integer> parallel = list.stream().filter(i -> i > 3);
                    Flux<Integer> sequentialFlux = Flux.fromStream(sequential).concatMap(i -> /** do your work **/);
                    Flux<Integer> parallelFlux = Flux.fromStream(parallel).flatMap(i -> /** do your work **/);

                    return Flux.merge(sequentialFlux, parallelFlux);
                }).log().subscribe();

在上面的示例中，元素 1, 2, 3 将被顺序处理，而元素 3, 4, 5 将被并行处理。

注意

你没有说清楚需要应用什么条件逻辑，所以我现在只应用一些虚拟逻辑。

另外，collectList() 只能用于有限流。

concatMap docs - this operator waits for one inner to complete before generating the next one and subscribing to it.

merge docs - Unlike concat, sources are subscribed to eagerly

【讨论】：

它没有回答这个问题。需要一个在运行时动态并行化的解决方案，而不是硬编码的解决方案。
@MohsenKashi 更新以使其更具活力

【解决方案2】：

要求有点模糊，但似乎bufferUntilChanged 可以解决问题。在您的示例中，它将收集一批的连续元素，然后发出它们。

flux.bufferUntilChanged(Order::getBatchId)

这会返回一个Flux<List<Order>>。您可以并行处理这些列表。

windowUntilChanged 也可以是一个选项。它会返回一个Flux<Flux<Order>>。

【讨论】：