[RxJava] Scheduler

Scheduler

RxJava는 Scheduler를 통해 멀티 스레드와 같은 비동기 작업을 돕는다. Schedulers 클래스에서 제공하는 정적 팩토리 메서드로 설정할 수 있다.

 

 

종류

IO

네트워킹, DB, 파일 시스템 등의 블로킹 이슈가 발생하는 곳에서 비동기적인 작업을 위해 사용한다.

 

newThread

매번 배로운 스레드를 생성하여 작업한다.

 

Computation

단순 반복적인 작업, 콜백 처리 그리고 다른 계산적인 작업에 사용된다. 블로킹 이슈가 존재하는 곳에서는 사용하지 않는다.

 

Trampoline

현재 스레드에 무한한 크기의 큐를 생성하여 모든 작업을 순차적으로 처리하는 것을 보쟝한다.

 

mainThread

RxAndroid에서 제공하는 스케줄러로 안드로이드의 메인 스레드에서 작동한다.

 

---

subscribeOn과 observeOn 연산자

한 줄 요약

• subscribeOn: 어떤 스레드(스케줄러)를 통해 아이템을 발행할 지 결정
• observeOn: 어떤 스레드(스케줄러)를 통해 아이템을 관찰 후 작업을 수행할 지 결정

 

RxJava에서 스케줄러를 이용하는 방법은 subscribeOn과 observeOn 연산자를 통해 가능하다. 이 연산자들로 멀티 스레딩을 제공할 수 있다.

먼저, subscribeOn연사자는 Observable 소스를 어떤 스레드(스케줄러)를 통해 아이템을 발행할 지 알려준다. 만약 subscribeOn연산자만 사용하고 observeOn이 없다면 해당 스케줄러는 아이템 발행 및 구독을 모두 한 스레드에서 수행한다.

Observable<Integer> src = Observable.create(emitter ->{
    for(int i=0; i<3; i++){
    	String threadName = Thread.currentThread().getName();
        System.out.println("Subs on "+threadName +": "+i);
        emitter.onNext(i);
        Thread.sleep(100);
     }
     emitter.onComplete();
});

src.subscribeOn(Scheduler.io())
    .subscribes(s->{
    	String threadNmae = Thread.currentThread().getName();
        System.out.println("Obsv on "+threadName+": "+s);
    });
Thread.sleep(500);


//실행 결과
Subs on RxCachedThreadScheduler-1:0
Obsv on RxCachedThreadScheduler-1:0
Subs on RxCachedThreadScheduler-1:1
Obsv on RxCachedThreadScheduler-1:1
Subs on RxCachedThreadScheduler-1:2
Obsv on RxCachedThreadScheduler-1:2

 

위의 결과에서 observeOn을 사용하면 어떻게 되는지 다음 코드를 통해 확인할 수 있다.

src.subscribeOn(Schedulers.io())
    .observeOn(schedulers.computation())
    .subscribes(s->{
    	String threadNmae = Thread.currentThread().getName();
        System.out.println("Obsv on "+threadName+": "+s);
    });
Thread.sleep(500);


//실행 결과
Subs on RxCachedThreadScheduler-1:0
Obsv on RxComputationThreadScheduler-1:0
Subs on RxCachedThreadScheduler-1:1
Obsv on RxComputationThreadScheduler-1:1
Subs on RxCachedThreadScheduler-1:2
Obsv on RxComputationThreadScheduler-1:2

observeOn 연산자를 사용하여 스케줄러를 지정하면 Observable에서 발행된 아이템을 가로채어 해당 스케줄러로 아이템을 구독한다. 

 

 

일반적인 안드로이드 개발 환경에서 네트워크나 데이터베이스 접근 과정에서는 메인 스레드가 블로킹 되는 것을 방지하도록 IO 스케줄러를 사용한다. 그리고, 요청한 데이터를 구독한 결과로 메인 스레드에서 IO를 갱신해야 한다.

repository.getUsers()
	.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(users -> {
    	//UI 갱신
    });