带时间间隔的Rx（Reactive Extensions for Java）Zip运算符

Question

Observable<String> data = Observable.just("one", "two", "three", "four", "five");
    Observable.zip(data, Observable.interval(1, TimeUnit.SECONDS), (d, t) -> {
        return d + " " + t;
    }).toBlocking().forEach(System.out::println);

这是可行的，但当我删除将源代码转换为BlockingObservable的toBlocking()时，程序执行并结束时没有输出。

我通常查看大理石图来正确理解事情：http://reactivex.io/documentation/operators/zip.html

在最后一句中，它说：它只会发射出与可观察源发射出的项数一样多的项，而可观察源发射出的项数最少。

这是否意味着dataObservable在不到1秒的时间内发出所有项目，并在打印每个Observable的前两个项目之前结束？因为每一个可观察到的东西本身都是异步的？

我需要清楚地了解正在发生的事情，以及是否有其他方法来处理类似的案件。有人吗？

Answer 1

observable.interval在封面下使用调度程序。它将从另一个线程发出。同时，主线程已经完成了所有的组成并将退出。您可能在main方法中有这段代码，这就是您的程序退出的原因。

在实际系统中，这不应该是一个问题（除非您的实际系统是一个包含此代码的main方法）。

在示例程序中，您可以通过从stdin读取一个字节来导致主线程阻塞。类似这样的事情：

Observable<String> data = Observable.just("one", "two", "three", "four", "five");
Observable.zip(data, Observable.interval(1, TimeUnit.SECONDS), (d, t) -> d + " " + t)
        .subscribe(System.out::println);

System.in.read();