延迟和间隔运算符无法正常工作 [英] delay and interval operators do not work properly

问题描述

如下图所示，我正在创建Observables。我想在几秒钟内等待特定的时间，如代码所示。因此我使用了延迟或间隔
运算符。我希望代码等待5秒，然后从观察者那里打印System.out.println。

As shown below, I am creating Observables. I would like to wait specific amount of time in seconds as shown in the code. thereforeI used either delay or interval operator. I expected the code to wait i.e 5 seconds then System.out.println from the observer to be printed.

但是会发生什么，执行doOnNext并且代码永远不会去进一步......我的意思是即使经过5秒后执行也会在doOnNext停止。

but what happens is, doOnNext is executed and the code never goes further..i mean the execution stops at doOnNext even after the 5 seconds elapsed.

代码：

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

推荐答案

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

You have to wait in the main method. Put Thread.sleep(10000) at the very end of the main() method so the Observable has chance to run. RxJava threads are daemon threads that stop when the application thread falls out of the main() method.

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

}

这篇关于延迟和间隔运算符无法正常工作的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！