继续 zip()，如果一个源已完成答案

【问题标题】：Continue zip(), if one source is completed继续 zip()，如果一个源已完成
【发布时间】：2021-07-01 12:35:15
【问题描述】：

我在使用 .zip() 运算符时遇到了一些问题。

让我用一个小例子来简化我的问题。

    Flux<Integer> flux1 = Flux.just(9, 8, 3, -2);
    Flux<Integer> flux2 = Flux.just(7);
    Flux<Integer> flux3 = Flux.just(6, 5, 4, -4);

    List<Flux<Integer>> list1 = Arrays.asList(flux1, flux2, flux3);

    TreeSet<Integer> set = new TreeSet<>(Comparator.reverseOrder());

    Set<Integer> list = Flux.zip(list1, objects -> {
        boolean setChanged = false;
        for (Object o : objects) {
            Integer i = (Integer) o;

            if (set.size() < 5 || i > set.last()) {
                setChanged = true;
                set.add(i);
                if (set.size() > 5) {
                    set.pollLast();
                }
            }
        }

        return setChanged;
    }).takeWhile(val -> val)
            .then(Mono.just(set))
            .block();

    System.out.println(list);

这里我有 3 个不同的来源（默认情况下它们按降序排序，它们的数量可能要大得多），我想从它们中获取 5 个按降序排序的元素的集合。不幸的是，我不能只使用 concat() 或 merge() 运算符，因为现实生活中的源可能非常大，但我只需要少量元素。

我在这里期待 [9, 8, 7, 6, 5]，但其中一个来源在第一次压缩后完成。

您能否建议我如何解决这个问题？

【问题讨论】：

为什么不能使用合并？这似乎是一个典型的用例——我只是合并所有这些通量然后做mergedFlux.take(5).sort(Comparator.reverseOrder()).collectList() 或类似的。在非常大或无限的通量上使用它没有问题 - 除了预取，它只会获取它需要的结果然后取消。这是首先使用反应式流的一大优势，您可以处理任意大的流（如果需要，可以处理无限大），而无需将其全部保存在内存中。
@MichaelBerry 好吧，如果使用 merge() 你建议我如何得到 [9, 8, 7, 3, -2] 结果。所以，看起来我们正在迭代整个第一个 Flux，然后开始迭代第二个。这对我来说不合适，因为在我的情况下，所有来源都使用 DB，如果 Flux1 有 > 1,000,000 个元素，即使我只需要 50 个元素，它也会花费很多时间。可能我对反应流的概念有误解，还有另一种做法？
不，merge() 将从任何通量中获取第一个可用结果（请参阅大理石图 here。您看到的行为是使用 Flux.just() 的副作用 - 所有元素可以同时立即可用，所以这里的结果有点不确定。或者我错过了什么，你不只是想要从任何通量中发射的前 5 个元素？
谢谢你的图表，现在我明白了。但是有什么办法可以避免这种影响并保证我的最终结果数据是一致的？在我的例子中，用户可以请求 100 个元素，但只能请求 5 个。
不确定我明白你的意思 - 在返回通量的“真实”场景中几乎永远无法保证一致性，因为它完全取决于元素何时发出（这反过来取决于哪个网络调用或其他非阻塞操作恰好首先完成。）或者您不希望在所有通量中发出第一个值？

标签： spring-webflux project-reactor flux reactor

【解决方案1】：

你可以试试reduce操作

    @Test
    void test() {
        Flux<Integer> flux1 = Flux.just(9, 8, 3, -2);
        Flux<Integer> flux2 = Flux.just(7, 0, -2, 4,3,2,2,1);
        Flux<Integer> flux3 = Flux.just(6, 5, 4, -4);

        var k = 5;
        List<Flux<Integer>> publishers = List.of(flux1, flux2, flux3);
        var flux = Flux.merge(publishers)
                .log()
                .limitRate(2)
                .buffer(2)
                .reduce((l1, l2) -> {
                    System.out.println(l1);
                    System.out.println(l2);

                    return Stream.concat(
                            l1.stream(),
                            l2.stream()
                    ).sorted(Comparator.reverseOrder())
                    .limit(k)
                    .collect(Collectors.toList());
                })
                .log();

        StepVerifier.create(flux)
                .expectNext(List.of(9,8,7,6,5))
                .expectComplete()
                .verify();
    }

您可以分块获取数据并比较它们以找到前 K 个元素。

在顺序情况下，它将获取一个新批次，将其与当前的前 k 个结果进行比较，并返回一个新的顶部，如上例所示（如果 k 很大，PriorityQueue 可能更适合排序）。如果您使用并行调度程序并且批处理是并行获取的，那么它可以独立地将它们相互比较，这应该会更快一些。您还可以通过 rateLimit、缓冲区、delayElements 等完全控制获取的数据

【讨论】：