如何使用RxJava Interval运算符

HuangDong.Li 发表于 Dev

黄东立

我正在学习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， interval

创建一个Observable，它发出以特定时间间隔隔开的整数序列

我是犯错了还是忘记了什么？

罗伊特·沙拉巴尼（Reut Sharabani）

您必须阻塞直到观察到的东西被消耗：

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