如何使用RxJava区间运算符

Question

我正在学习RxJava运算符，我发现下面的这些代码没有打印任何东西：

public static void main(String[] args) {

    Observable
    .interval(1, TimeUnit.SECONDS)
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError -> " + e.getMessage());
        }

        @Override
        public void onNext(Long l) {
            System.out.println("onNext -> " + l);
        }
    });
}

作为ReactiveX，间隔

创建一个可观察对象，该对象发出一系列以特定时间间隔隔开的整数

我是不是搞错了或者忘了什么？

Answer 1

正如他们告诉您的那样，间隔工作是异步的，因此您必须等待所有事件完成。

您可以在订阅后获得订阅，然后使用作为reactiveX平台一部分的TestSubcriber，它将为您提供等待所有事件终止的功能。

       @Test
public void testObservableInterval() throws InterruptedException {
    Subscription subscription = Observable.interval(1, TimeUnit.SECONDS)
              .map(time-> "item emitted")
              .subscribe(System.out::print,
                         item -> System.out.print("final:" + item));
    new TestSubscriber((Observer) subscription)
            .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);
}

如果您需要，我的github中有更多示例https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java

Answer 2

放置<代码>线程。在订阅之后睡眠（1000000），您将看到它正在工作<代码>可观察。默认情况下，间隔在调度程序上运行。computation（）因此流是在主线程以外的线程上运行的。

Answer 3

您必须阻止直到可观察对象被消耗：

public static void main(String[] args) throws Exception {

    CountDownLatch latch = new CountDownLatch(1);

    Observable
    .interval(1, TimeUnit.SECONDS)
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
            // make sure to complete only when observable is done
            latch.countDown();
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError -> " + e.getMessage());
        }

        @Override
        public void onNext(Long l) {
            System.out.println("onNext -> " + l);
        }
    });

    // wait for observable to complete (never in this case...)
    latch.await();
}

例如，您可以添加. get（10）以查看可观察的完整。