RxJS - Scheduler

Scheduler(스케줄러)는 무엇일까요? Scheduler는 구독의 시작과 알림을 전달하는 시기를 제어합니다. Scheduler는 세개의 구성요소로 구성되어 있습니다.

Scheduler는 자료구조입니다. 우선순위나 다른 기준에 따라서 작업을 저장하고 대기열에 넣는 방법을 알고 있습니다.
Scheduler는 실행 컨텍스트 입니다. 작업이 실행되는 장소와 시간을 나타냅니다.(예를 들어 즉시 혹은 setTimeout, process.nextTick, the animation frame 등과 같은 다른 콜백 메커니즘)
스케줄러에는 (가상의)시계가 있습니다. 스케줄러의 getter 메소드 now()에 의해 "시간"개념을 제공합니다. 특정 스케줄러에서 예약된 작업은 해당 스케줄러의 시계가 나타내는 시간만 준수합니다.
스케줄러를 사용하면 Observable 어떤 실행 콘텍스트에서 Observer에게 알림을 전달할지 정의할 수 있습니다.

아래의 예에서 값 1, 2, 3을 동기적으로 내보내는(emits) 보통의 단순한 Observable에 연산자 observeOn 사용해서 값을 전달하는데 사용할 비동기 스케줄러를 지정합니다.

import { Observable, asyncScheduler } from 'rxjs';
import { observeOn } from 'rxjs/operators';

const observable = new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
}).pipe(
  observeOn(asyncScheduler)
);

console.log('just before subscribe');
observable.subscribe({
  next(x) {
    console.log('got value ' + x)
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
     console.log('done');
  }
});
console.log('just after subscribe');

출력과 같이 실행됩니다.

just before subscribe
just after subscribe
got value 1
got value 2
got value 3
done

지금까지 본 기본적인 동작과 다른 구독 바로 직후에 값이 전달된 것에 주목하십시오. 이는 observeOn(asyncScheduler)가 새로운 Observable과 최종 Observer 사이에 프록시 Observer를 도입하기 때문입니다. 예제 코드에서 구분을 명확히 하기 위해서 몇개의 식별자의 이름을 변경하겠습니다.

import { Observable, asyncScheduler } from 'rxjs';
import { observeOn } from 'rxjs/operators';

var observable = new Observable((proxyObserver) => {
  proxyObserver.next(1);
  proxyObserver.next(2);
  proxyObserver.next(3);
  proxyObserver.complete();
}).pipe(
  observeOn(asyncScheduler)
);

var finalObserver = {
  next(x) {
    console.log('got value ' + x)
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
     console.log('done');
  }
};

console.log('just before subscribe');
observable.subscribe(finalObserver);
console.log('just after subscribe');

observeOn(asyncScheduler)에서 작성된 proxyObserver의 함수 next(val)는 대략 아래와 같습니다.

const proxyObserver = {
  next(val) {
    asyncScheduler.schedule(
      (x) => finalObserver.next(x),
      0 /* delay */,
      val /* will be the x for the function above */
    );
  },

  // ...
}

비동기 스케줄러는 설령 지정된 지연 시간이 0 이더라도 setTimeout 혹은 setTimeout과 함께 동작합니다. JavaScript에서 setTimeout(fn, 0)는 보통 다음 이벤트 루프에서 함수 fn을 가장 빠르게 실행하는 것으로 알려져 있습니다. 이는 왜 구독 바로 직후에 값 1이 finalObserver에게 전달되었는지를 설명합니다.

스케줄러의 schedule() 메소드는 스케줄러 내부 시계와 관련된 시간량으로 참조되는 지연시간을 인자로 받습니다. 스케줄러의 시계는 실제의 시간과 관련이 있을 필요가 없습니다. 지연 연산과 같은 시간과 관계된 연산자는 실제 시간이 아니라 스케줄러의 시계에 의해서 동작합니다. 이는 실제로 스케줄된 작업을 동기적으로 실행하는 동안 가상시간 스케줄러를 이용하여 시간을 위조해야하는 테스트에 매우 유용합니다.

Scheduler Types

비동기 스케줄러는 RxJS가 제공하는 빌트인 스케줄러의 하나입니다. 그것은 각각 스케줄러 오브젝트의 정적 프로퍼티에 의해서 생성되어 반환될 수 있습니다.

스케줄러	목적
null	스케줄러를 전달하지 않으면 동기 및 재귀적으로 전달됩니다. 이것은 일정시간 연산 혹은 꼬리재귀에 사용할 수 있습니다.
queueScheduler	현재의 이벤트 프레임(trampoline scheduler) 큐에 스케줄링합니다. 반복 연산자에 사용하십시오.
asapScheduler	``Promise`에 사용되는 것과 동일한 마이크로 작업 큐에 스케줄링 합니다. 기본적으로 현재 잡이후 다음 잡이 실행되지 전에 스케줄링 됩니다. 비동기 변환에 사용됩니다.
asyncScheduler	`setInterval`과 함께 동작합니다. 시간 기반 연산자에 사용됩니다.
animationFrameScheduler	다음 브라우저 재구획 직전에 스케줄링 됩니다. 부드러운 브라우저 애니메이션을 위해 사용할 수 있습니다.

Using Schedulers

당신이 RxJS 코드상에서 명시적으로 스케줄러를 지정하지 않았더라도 이미 사용되었을 수 있습니다. 왜냐하면 동시성을 처리하는 모든 Observable 연산자는 선택적인 스케줄러를 가지고 있기 때문입니다. 스케줄러를 지정하지 않았을 경우 RxJS는 최소 동시성 원칙을 사용하여 기본 스케줄러를 선택합니다. 이것은 선택된 연산자가 요구하는 최소한의 동시성을 만족시키는 스케줄러가 선택됨을 의미합니다. 예를 들어 작고 유한한 수의 메시지를 반환하는 Observable의 연산자는 스케줄러를 사용하지 않습니다. 즉 null 혹은 undefined 스케줄러 입니다. 잠재적으로 대량 혹은 무한대의 메시지가 반환되는 연산자의 경우 큐 스케줄러가 사용됩니다. 타이머를 사용하는 연산자의 경우에는 비동기 스케줄러가 사용됩니다.

RxJS는 최소한의 동시성 스케줄러를 사용하므로 성능을 목적으로 동시성을 도입하려는 경우 다른 스케줄러를 지정할 수 있습니다. 특정 스케줄러를 지정하기 위해 스케줄러를 사용하는 연산자 메소드, 예를 들어 from ([10, 20, 30], asyncScheduler)을 사용할 수 있습니다.

일반적으로 정적 생성 연산자는 스케줄러를 인자로 받습니다. 예를 들어 from (array, scheduler)를 사용하면 배열에서 변환된 각각의 알림을 전달할 때 사용할 스케줄러를 지정할 수 있습니다. 보통 연산자의 마지막 인자입니다. 다음의 정적 생성 연산자는 스케줄러를 인자로 받습니다.

bindCallback
bindNodeCallback
combineLatest
concat
empty
from
fromPromise
interval
merge
of
range
throw
timer

subscribeOn을 사용하여 subscribe()가 호출될 컨텍스트를 스케줄링 하십시오. 기본적으로 Observable의 subscribe() 호출은 동기적으로 즉시 발생합니다. 그러나 인스턴스의 연산자 subscribeOn(scheduler)를 사용하여 scheduler에 인수로 지정한 스케줄러에 의해 실제 구독을 지연시키거나 스케줄링 할 수 있습니다.

observeOn을 사용하여 알림이 전달 될 컨텍스트를 스케줄링합니다. 위의 예에서 알 수 있듯이 인스턴스 연산자 observeOn(scheduler)는 소스 Observable 대상 Observable 사이에 지정한 스케줄러를 사용해서 스케줄링 해주는 중재자 Observer를 도입합니다.

인스턴스 연산자는 인자로 스케줄러를 받을 수 있습니다.

bufferTime, debounceTime, delay, auditTime, sampleTime,throttleTime, timeInterval, timeout, timeoutWith, windowTime과 같은 시간 관련 연산자 모두 스케줄러를 마지막 인자 받으며 지정하지 않을 경우 asyncScheduler를 사용합니다.

이외에도 cache, combineLatest, concat, expand, merge, publishReplay, startWith가 스케줄러를 인자로 받습니다.

cache와 publishReplay는 ReplaySubject를 사용하므로 스케줄러를 받는 것에 주의하기 바랍니다. 스케줄러의 컨텍스트에서만 의미가 있긴하지만 ReplaySubject는 시간을 처리할 수 있으므로 ReplaySubjects의 생성자는 마지막 인자로 스케줄러를 선택적으로 받습니다. 기본적으로 ReplaySubject는 큐 스케줄러를 이용하여 시간을 제공합니다.

출처

이글은 RxJS 공식 사이트 Scheduler, CC BY 4.0 페이지의 번역물 입니다.