RxJava2-간격 및 스케줄러

Question

간격이 있고 여기에 computeScheduler를 부여했다고 가정 해 보겠습니다. 이렇게 :

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

그러면 플랫 맵 {...}에서 일어나는 모든 일이 계산 스레드에서도 예약됩니까?

Observable.interval (long initialDelay, long period, TimeUnit unit, Scheduler scheduler)의 소스에서 다음과 같이 말합니다.

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

RxJava의 초심자로서 저는이 코멘트를 이해하는 데 어려움을 겪고 있습니다. 인터벌 타이머 / 대기 논리가 계산 스레드에서 발생한다는 것을 이해합니다. 그러나 방출되는 항목에 대한 마지막 부분도 방출 된 항목이 동일한 스레드에서 소비 된다는 것을 의미 합니까? 아니면 ObservOn이 필요합니까? 이렇게 :

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

계산 스레드에서 방출을 처리하려면 observeOn이 필요합니까?

Answer 1

확인하는 것은 간단합니다. 현재 스레드를 인쇄하여 연산자가 실행되는 스레드를 확인합니다.

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

이것은 항상 다음을 인쇄합니다.

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

모든 것이 단일 스레드->에서 발생하므로 순차적으로 처리 main됩니다.

observeOn 다운 스트림 실행 스레드를 변경합니다.

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

이번에는 각 실행에 대해 다른되지만 결과 flatmap와 subscribediffrent 스레드로 처리한다 :

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

intervalobserveOn다운 스트림 실행 스레드 (스케줄러) 로 작동 하고 변경합니다.

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

이번에는 계산 스케줄러의 한 스레드 내에서 순차적으로 실행됩니다.

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
...

interval기본적으로 계산 스케줄러를 사용하므로 인수로 전달할 observeOn필요가 없으며 필요하지 않습니다.