Rx (Reactive Extensions for Java) 带时间间隔的 Zip 运算符

Question

我对 RxJava 比较陌生，我已经和操作符玩了一段时间了。

我看到了这个在短时间间隔（1s）后发出项目的小例子：

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

在最后一句中它说：它只会发射与发射最少项目的源 Observable 发射的项目数量一样多的项目。

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

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

Answer 1

Observable.interval 在幕后使用Scheduler。它将从另一个线程发出。同时，主线程已经完成了所有的编写，将退出。大概您在 main 方法中有此代码，这就是您的程序退出的原因。

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

在一个示例程序中，您可以通过从标准输入读取一个字节来导致主线程阻塞。像这样的：

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();

Answer 2

基本上你的主程序在可观察对象有机会发出任何东西之前就退出了，这就是你看不到任何输出的原因。修复它的方法是以某种方式阻止，直到Observable 发出所有项目，这是使用CountDownLatch 的一种方法：

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

latch.await(10, TimeUnit.SECONDS);