延迟和间隔运算符无法正常工作

Question

如下图，我正在创建Observables。我想以秒为单位等待特定的时间，如代码所示。因此我使用了延迟或间隔 操作员。我希望代码等待，即 5 秒，然后从观察者那里打印System.out.println。

但发生的情况是，doOnNext 被执行并且代码永远不会更进一步。我的意思是即使在 5 秒过去之后，执行也会在 doOnNext 处停止。

代码：

public static void main(String[] args) {
    Observable<List<Person>> observables = Observable.create(e-> {
        for(List<Person> p : Main.getPersons()) {
            e.onNext(p);
        }
        e.onComplete();
    });
     observables
    //.subscribeOn(Schedulers.newThread())//newThread
    .flatMap(p->Main.toObservable(p.get(0).getName()))
    .doOnNext(p-> System.out.println(p.length()) )
    .map(p->p+"..STRING")
    //.delay(5, TimeUnit.SECONDS)
    //.interval(0, 5, TimeUnit.SECONDS)
    .observeOn(Schedulers.io())
    .subscribe(new Observer() {
        @Override
        public void onComplete() {
            // TODO Auto-generated method stub
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable arg0) {
            // TODO Auto-generated method stub

        }

        @Override
        public void onNext(Object arg0) {
            // TODO Auto-generated method stub
            System.out.println("onNextFromObserver: " + arg0);
        }

        @Override
        public void onSubscribe(Disposable arg0) {
            // TODO Auto-generated method stub
        }
    });
}

private static <T> Observable<T> toObservable(T s) {
    return Observable.just(s);
}

private static List<List<Person>> getPersons() {
    return Arrays.asList(
            Arrays.asList(new Person("Sanna1", 59, "EGY"), new Person("Sanna2", 59, "EGY"), new Person("Sanna3", 59, "EGY")),
            Arrays.asList(new Person("Mohamed1", 59, "EGY"), new Person("Mohamed2", 59, "EGY")),
            Arrays.asList(new Person("Ahmed1", 44, "QTR"), new Person("Ahmed2", 44, "QTR"), new Person("Ahmed3", 44, "QTR")),
                    Arrays.asList(new Person("Fatma", 29, "KSA")),
                    Arrays.asList(new Person("Lobna", 24, "EGY"))
                    );
}

Answer 1

你必须在 main 方法中等待。将Thread.sleep(10000) 放在main() 方法的最后，这样Observable 就有机会运行。 RxJava 线程是守护线程，当应用程序线程退出main() 方法时停止。

public static void main(String[] args) {

    Observable.just("Hello World!", "Keep printing values!")
    .zipWith(Observable.interval(0, 5, TimeUnit.SECONDS), (a, b) -> a)
    .subscribe(v -> 
        System.out.println(Thread.currentThread() + ": " + v)
    );

    Thread.sleep(10000);  // <-----------------------------------

}