Seltsames Verhalten von Observable # repeat in reaktiven Erweiterungen - java, rx-java, reaktive Programmierung, Wiederholung, Sampling

Ich spiele mit den RX - Betreibern herum und bin gespannt warum just(null).repeat() funktioniert nicht als Parameter für einen der eingebauten Operatoren:

Observable.interval(1, TimeUnit.SECONDS)
.sample(Observable.just(null).repeat())
.subscribe(System.out::println);

Ich hätte erwartet, dass dies gedruckt wird 0 1 2 3 ... aber es hängt einfach. Ich stelle mir vor, es liegt an der repeat ist die Standardeinstellung Schedulerjedoch, wenn Sie die Rollen von tauschen interval und das just-repeat dann funktioniert es wie erwartet und druckt null einmal pro Sekunde:

Observable.just(null).repeat()
.sample(Observable.interval(1, TimeUnit.SECONDS))
.subscribe(System.out::println);

Was ist denn hier los?

Antworten:

Wenn Sie keinen Scheduler angeben (und kein Operator legt einen fest), erfolgt die gesamte Verarbeitung auf demselben Thread. just(null).repeat() wird 100% eines CPU-Kerns belasten, so dass nichts anderes die Chance bekommt, fortzufahren.

In Ihrem Fall, der interval wird produziert am Scedulers.computation() Scheduler, und weil es beim Start ist und danach keine Scheduler - Änderungen mehr stattfinden, ist Ihr repeat arbeitet auch am selben thread.

Im zweiten Fall wird alles im selben Thread abonniert, mit Ausnahme des Intervalls, das sich in einem eigenen Scheduler befindet. der rest ist abhängig von der internen umsetzung von sample.

Wenn Sie einen bestimmten Scheduler verwenden, sollte dieser funktionieren:

.sample(Observable.just(null).repeat().subscribeOn(Schedulers.computation()))

Beachten Sie, dass Sie nur Nullen anstelle der Zahlen verwenden möchten, die interval viel effizienter zu nutzen ist map Anstatt von sample:

.map(any -> (Object) null)