flux.zip方法不发射所有元素

Question

我正在与反应流和出版商（Mono和Flux）合作，并使用zip和zipWith Flux方法将这两个出版商结合起来，如下所示：

Flux<String> flux1 = Flux.just(" {1} ","{2} ","{3} ","{4} " );
Flux<String> flux2 = Flux.just(" |A|"," |B| "," |C| ");
Flux.zip(flux1, flux2,
                    (itemflux1, itemflux2) -> "[ "+itemflux1 + ":"+ itemflux2 + " ] " )
            .subscribe(System.out::print);

以下是输出：

[  {1} : |A| ] [ {2} : |B|  ] [ {3} : |C|  ] 

由于flux1有四个元素，flux2有三个元素，flux1中的第四个元素丢失。当我试图打印通量的日志时，没有关于第四元素发生了什么的信息。

以下是打印日志的语句：

Flux.zip(flux1, flux2,
                (itemflux1, itemflux2) -> "[ "+itemflux1 + ":"+ itemflux2 + " ] " ).log()
        .subscribe(System.out::print);

下面是使用log方法的控制台日志：

[info] onSubscribe(FluxZip.ZipCoordinator)
[info] request(unbounded)
[info] onNext([  {1} : |A| ] )
[  {1} : |A| ] [info] onNext([ {2} : |B|  ] )
[ {2} : |B|  ] [info] onNext([ {3} : |C|  ] )
[ {3} : |C|  ] [info] onComplete()

从zip方法的文档中，我得到了

操作员将继续这样做，直到任何源完成。错误将立即转发。这种“分步合并”处理在分散-聚集场景中特别有用。

但在我的例子中，它没有记录任何错误，也没有记录任何关于丢失元素的消息。

我怎样才能得到丢失元素的信息？

请建议。

Answer 1

解释留档所说的
错误将立即转发。--

没有办法得到丢失的元素。因为当其中一个流结束时，不会进一步读取该流。希望这是清楚的。如果您真的想获取流的最后一个元素，请尝试其他操作符。

Answer 2

这是该运算符的预期行为。使用Flux。zip，提供的流量之一可能是无限的；一个常见的例子是用流量压缩流量。间隔（持续时间）实例（无限）。

如果你被困在这种情况下，这可能意味着你需要使用不同的操作员。

Answer 3

zip/zipWith将输出最短流量中元素的数量。当最小的通量终止时，它会取消较长的通量，如果将log（）放在源通量上，而不是压缩的通量上，则应该可以看到。

这可以通过这个代码段来演示（它被调优以显示1×1请求并作为单元测试运行，因此隐藏（）/zipBy（...，1）和block last（））：

@Test
public void test() {
    Flux<Integer> flux1 = Flux.range(1, 4).hide().log("\tFLUX 1");
    Flux<Integer> flux2 = Flux.range(10, 2).hide().log("\tFlux 2");

    flux1.zipWith(flux2, 1)
        .log("zipped")
        .blockLast();
}

哪些输出：

11:57:21.072 [main] INFO  zipped - onSubscribe(FluxZip.ZipCoordinator)
11:57:21.077 [main] INFO  zipped - request(unbounded)
11:57:21.079 [main] INFO    FLUX 1 - onSubscribe(FluxHide.HideSubscriber)
11:57:21.079 [main] INFO    FLUX 1 - request(1)
11:57:21.079 [main] INFO    FLUX 1 - onNext(1)
11:57:21.079 [main] INFO    Flux 2 - onSubscribe(FluxHide.HideSubscriber)
11:57:21.080 [main] INFO    Flux 2 - request(1)
11:57:21.080 [main] INFO    Flux 2 - onNext(10)
11:57:21.080 [main] INFO  zipped - onNext([1,10])
11:57:21.080 [main] INFO    FLUX 1 - request(1)
11:57:21.080 [main] INFO    FLUX 1 - onNext(2)
11:57:21.080 [main] INFO    Flux 2 - request(1)
11:57:21.080 [main] INFO    Flux 2 - onNext(11)
11:57:21.080 [main] INFO  zipped - onNext([2,11])
11:57:21.080 [main] INFO    FLUX 1 - request(1)
11:57:21.080 [main] INFO    FLUX 1 - onNext(3)
11:57:21.080 [main] INFO    Flux 2 - request(1)
11:57:21.080 [main] INFO    Flux 2 - onComplete()
11:57:21.081 [main] INFO    FLUX 1 - cancel() <----- HERE
11:57:21.081 [main] INFO    Flux 2 - cancel()
11:57:21.081 [main] INFO  zipped - onComplete()

这是“直到任何源完成”部分。