反应堆项目中的条件平行性

Question

在某些流中，必须基于特定字段并平行解析消息。

使用 GroupBy 不是一个选项，因为它会收集大量结果然后对它们进行分组（它引入了延迟）。 另外，如果使用 subscribeon（Schedulers.Single（）），则结果还可以，但它引入了饥饿问题。

作为示例订单，应顺序执行具有相同 batchid 的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));
    }

Answer 1

您应该能够创建2个单独的 flux 实例。一个处理每个发射的项目顺序的，另一个处理它们并并行处理它们。然后，您可以合并将它们纳入单个 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 将并行处理。

Note

您尚未清楚需要应用哪种条件逻辑，因此我现在只应用了一些虚拟逻辑。

另外， collectList（）仅应与有限流一起使用。

- 此操作员等待一个内部完成，然后再生成下一个并订阅它。

merge docs - Unlike concat, sources are subscribed to eagerly< /代码>

Answer 2

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

flux.bufferUntilChanged(Order::getBatchId)

这将返回A flux＆lt; list＆lt; order＆gt;＆gt; 。您可以并行处理列表。

WindownuntilChanged 也可以是一个选项。它将返回 flux＆lt; lux＆lt; order＆gt;＆gt; 。