[RxSwift] 12. Scheduler

 

* 사용 : 쓰레드와 병렬처리를 다룰 때

 

1. Scheduler 개념

1) 정의

 프로세스가 있는 곳의 context (여기서의 context = thread, dispatch queue)

 

2) 사용

한 Context에서 cache 연산자를 이용하여 전달

2. 연산자

1) subscribeOn(_:)

Observable의 프로세싱이 어디서 일어나게 할 것인지 결정하는 연산자

let fruit = Observable<String>.create { observer in
    observer.onNext("[apple]")
    sleep(2)
    observer.onNext("[pineapple]")
    sleep(2)
    observer.onNext("[strawberry]")
    return Disposables.create()
}

let globalScheduler = ConcurrentDispatchQueueScheduler(queue: DispatchQueue.global())
fruit
    .subscribeOn(globalScheduler)
    .subscribe(
        onNext: {print($0)}
)
    .disposed(by: db)

2) observeOn(_:)

Observer가 Observable을 어디서 oberve할 것인지

fruit
    .subscribeOn(globalScheduler)
    .dump()
    .observeOn(MainScheduler.instance)
    .dumpingSubscription()
    .disposed(by: bag)

observeOn

 

* 일반적으로 subscribeOn(_:)은 Background Thread에서, observeOn(_:)은 Main Thread에서 사용

단, 다음과 같이 사용 가능

(observeOn)

Observable.just("Some string")                  // UI
       .map(str -> str.length())               // UI
       .observeOn(Schedulers.computation())   // Changing the thread
       .map(length -> 2 * length)            // Computation
       .subscribe(---)

(subscribeOn)

Observable.just("Some String")              // Computation
  .map(str -> str.length())                // Computation
  .map(length -> 2 * length)              // Computation
  .subscribeOn(Schedulers.computation()) // -- changing the thread
  .subscribe(number -> Log.d("", "Number " + number));// Computation

3. Scheduler의 종류

serial scheduler를 사용한다면, Rx는 직렬적으로 진행

cuncurrent scheduler를 사용한다면, Rx는 동시에 진행

 

1) MainScheduler

- 메인 쓰레드에서 가장 위에 존재(UI와 high-priority tasks를 진행 ,,, heavy task는 피해야함(API request등)

- UI를 갱신하려면 이 스케줄러로 변경하여 사용

 

2) SerialDispatchQueueScheduler

- background에서 추출하는 일을 처리할 때 사용 (Firebase를 사용할 때 서버의 endpoint에 너무 많은 pressured을 줄일 수 있음)

 

3) ConcurrentDispatchQueueScheduler

- SerialDispatchQueueScheduler와 같이 추출하는 일을 처리할 때 사용, 단 병렬적

 (작업량이 많은 일에 사용)

 

4) TestScheduler

- 테스트를 위한 것이며, production code에는 사용하지 않음

- RxTeat 라이브러리에 존재

let scheduler = TestScheduler(initialClock: 0)
let xs = scheduler.createColdObservable([ next(50, 42),
	next(60, 43),
	completed(70)
])

let res = scheduler.start {
	xs.delaySubscription(30, scheduler: scheduler) 
}

// test
XCTAssertEqual(res.events, [ 
	next(280, 42),
	next(290, 43), 
    completed(300)
])