RxJava2 - interwał i harmonogramy

Question

Powiedzmy, że mam interwał i podałem mu obliczenie harmonogramu. Lubię to:

Observable
    .interval(0, 1, TimeUnit.SECONDS, computationScheduler)
    .flatMap { ... }

Czy w takim razie wszystko, co dzieje się w flatmap {...}, będzie również zaplanowane w wątku obliczeniowym?

W źródłach dla Observable.interval (long initialDelay, long period, TimeUnit unit, Scheduler Scheduler) jest napisane:

 * @param scheduler
 * the Scheduler on which the waiting happens and items are emitted

Jako początkujący w RxJava, ciężko mi zrozumieć ten komentarz. Rozumiem, że w wątku obliczeniowym występuje licznik czasu / logika oczekiwania. Ale czy ostatnia część, dotycząca emitowanych przedmiotów, oznacza również, że emitowane przedmioty zostaną zużyte w tym samym wątku? A może do tego jest wymagane followOn? Lubię to:

Observable
    .interval(0, 1, TimeUnit.SECONDS, computationScheduler)
    .observeOn(computationScheduler)
    .flatMap { ... }

Czy byłoby to konieczne, jeśli chcę, aby emisje były przetwarzane w wątku obliczeniowym?

Answer 1

Jest to proste do zweryfikowania: po prostu wydrukuj bieżący wątek, aby zobaczyć, w którym wątku jest wykonywany operator:

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
    .flatMap(e -> {
        System.out.println("on flatmap: " + Thread.currentThread().getName());
        return Observable.just(e).map(x -> "--> " + x);
    })
    .subscribe(s -> {
        System.out.println("on subscribe: " + Thread.currentThread().getName());
        System.out.println(s);
    });

To zawsze wydrukuje:

on subscribe: main
--> 1
on flatmap: main
on subscribe: main
--> 2
on flatmap: main
on subscribe: main
--> 3
on flatmap: main
on subscribe: main
--> 4
on flatmap: main
on subscribe: main
--> 5
on flatmap: main
on subscribe: main
--> 6
on flatmap: main
on subscribe: main
--> 7
on flatmap: main
on subscribe: main
--> 8
on flatmap: main
on subscribe: main
--> 9

Przetwarzane sekwencyjnie, ponieważ wszystko dzieje się w jednym wątku -> main.

observeOn zmieni wątek wykonawczy podrzędny:

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
    .observeOn(Schedulers.computation())
    .flatMap(e -> {
         System.out.println("on flatmap: " + Thread.currentThread().getName());
         return Observable.just(e).map(x -> "--> " + x);
     })
     .observeOn(Schedulers.io())
     .subscribe(s -> {
         System.out.println("on subscribe: " + Thread.currentThread().getName());
         System.out.println(s);
      });

Wynik tym razem będzie inny dla każdego wykonania, ale flatmapi subscribebędzie przetwarzany w różnych wątkach:

on flatmap: RxComputationThreadPool-1
on subscribe: RxCachedThreadScheduler-1

intervalbędzie działał jako observeOni zmienił podrzędny wątek wykonawczy (harmonogram):

Observable.interval(0, 1, TimeUnit.SECONDS, Schedulers.computation())
    .flatMap(e -> {
        System.out.println("on flatmap: " + Thread.currentThread().getName());
        return Observable.just(e).map(x -> "--> " + x);
    })
    .subscribe(s -> {
        System.out.println("on subscribe: " + Thread.currentThread().getName());
        System.out.println(s);
    });

Tym razem wykonanie jest sekwencyjne w jednym wątku harmonogramu obliczeń:

on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 0
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 1
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 2
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 3
...

intervaldomyślnie użyje harmonogramu obliczeń, nie musisz przekazywać go jako argumentu i observeOnnie jest potrzebny