UniRx의 작동 원리(5) 스케쥴러

스케쥴러란?

Observable의 메시지의 처리를 언제 어디에서 실행할 것인가를 제어하기 위한 도구 입니다.

Observable에서 병렬 처리를 수행하는 경우 어떤 스레드에서 처리할지, 어떤 타이밍에 실행할 것인가를 제어하는 것은 중요합니다.

이러한 처리를 쉽게 해주는 것이 스케쥴러 입니다.

 

UniRx에서는 다음과 같은 스케쥴러를 제공합니다.

 

ImmediateScheduler	Scheduler.Immediate	현재 실행중인 스레드에서 즉시 처리
CurrentThreadScheduler	Scheduler.CurrentThreadScheduler	일단 큐에 처리를 담았다가 현재 처리하고 있는 스레드에서 순서대로 처리
ThreadPoolScheduler	Scheduler.ThreadPoolScheduler	스레드 풀에서 처리를 실행
MainThreadScheduler	Scheduler.MainThreadScheduler	유니티 메인스레드에서 처리를 실행
IgonoreTimeScaleMainThreadScheduler	Scheduler.IgonoreTimeScaleMainThreadScheduler	유니티 메인 스레드에서 처리를 실행하지만 Time.timescale의 영향을 받지 않음
FixedUpdateMainThreadScheduler	Scheduler.FixedUpdateMainThreadScheduler	역시 메인 스레드에서 처리를 하지만 FixedUpdate기준
EndOfFrameMainThreadScheduler	Scheduler.EndOfFrameMainThreadScheduler	똑같이 EndOfFrame기준으로 실행

실행중인 스레드란 직전에 메시지 처리가 실행되고 있던 Thread를 의미합니다.

메인 스레드에서 실행되는 처리라면 그대로 메인스레드에서 스레드풀에서 처리를 하고 있는 경우에는 그대로 스레드 풀에서 처리됩니다.

이 것을 의미하는 것은 Observable이 정의한 위치와 관련이 없다는 이야기 입니다.  

메인 스레드에서 Observable을 정의하고 거기에서 Current ThreadScheduler를 지정했기 떄문에 처리는 메인 스레드에서 실행되어야한다라고 이해하면 잘못된 이해이니 주의하시기 바랍니다.

 

//현재 Start 함수가 실행되는 스레드의 아이디 체크
Debug.Log("Unity Main Thread ID : " + Thread.CurrentThread.ManagedThreadId);

var subject = new Subject<Unit>();
subject.AddTo(this);

//이 주제에 관련하여 구독을 시작하고, 구독자가 실행되는 스레드의 아이디를 출력
//ObserveOn에서 Scheduler.Immediate를 했기 때문에 처리되는 스레드에서 출력한다.
subject
    .ObserveOn(Scheduler.Immediate)
    .Subscribe(_ =>
    {
        Debug.Log("Thread ID" + Thread.CurrentThread.ManagedThreadId);
    });

//현재 스레드에서 OnNext
subject.OnNext(Unit.Default);

//다른 스레드에서 OnNext
Task.Run(() =>
{
    subject.OnNext(Unit.Default);
});

subject.OnCompleted();

 

 

실행결과

같은 곳에서 subject.Next를 한 경우 같은 스레드 아이디를 띄지만 Task.Run으로 돌린 OnNext는 다른 스레드의 아이디가 뜹니다.

 

 

스케쥴러 지정 및 기본 스케쥴러

 

스케쥴러 오퍼레이터 및 팩토리 메소드를 사용할 때 어떤 스케쥴러를 사용할지 지정할 수 있습니다.

이 경우 스케쥴러를 지정하지 않으면 오퍼레이터 및 팩토리 메소드에서 기본 스케쥴러가 선택됩니다.

시간의 계측이 필요한 오퍼레이터나 팩토리 메소드에 대해서는 디폴트로 MainThreadScheduler가 사용됩니다.

그 외에 오퍼레이터나 팩토리 메소드에서는 ImmediateScheduler 또는 CurrentThreadScheduler가 디폴트로 사용됩니다.

 

ImmediateScheduler와 CurrentThreadScheduler의 차이

자 그렇다면 위에서 특정 오퍼레이터나 팩토리 메소드에서는 ImmediateScheduler나 CurrentThreadScheduler가 사용된다고 하였는데 어떤 것의 차이가 있을까요?

일단 양쪽 모두 현재 실행중인 스레드 상에서 처리를 실시하는 스케쥴러입니다. 차이가 있다면 처리의 실행 요구가 있을 때 즉시 처리를 하느냐 한번 큐에 담았다가 꺼내어서 사용하느냐의 차이 입니다.

이 차이가 크게 나타나는 상황으로는 Observable이 재귀를 할 경우 입니다.

 

• ImmediateScheduler의 경우

Observable.Return(1, Scheduler.Immediate).Repeat().Take(1).Subscribe();

위 코드는 무한 루프가 발생하여 멈추지 않습니다. 이유는 Repeat에 있습니다. Repeat은 OnCompleted 메시지를 받으면 다시 Subscribe()를 호출하는 오퍼레이터입니다. Observable.Return은 Subscribe되면 OnNext OnCompleted 메시지를 계속 발행하는 메서드입니다. 이 두가지 동작이 맞물리면 OnCompleted 발행과 Repeat이 무한 반복적으로 이루어집니다.

특히! ImmediateScheduler를 사용하는 경우 Repeat를 통한 재 Subscribe 호출이 우선시 되기 때문에 OnComplete의 해체보다

Repeat의 동작이 우선되어버려 무한 루프가 발생합니다.

• CurrentThreadScheduler의 경우

Observable.Return(1, Scheduler.CurrentThread).Repeat().Take(1).Subscribe();

 

이 경우에는 위의 예와는 달리 정지합니다.

이유는 Current.ThreadScheduler가 요청한 처리를 한 번 대기열에 넣은 다음 하나씩 순서대로 실행하는 방식이기 때문입니다.

즉, OnCompleted 메시지 발행과 관련된 Observable 해체 처리가 완료된 후 Repeat의 Subscribe가 실행되기 때문에

Observable의 해제 처리가 실행되면 Repeat도 취소되고 Observable 전체가 중지됩니다.

 

 

스케쥴러의 사용법

 

메시지 처리 타이밍 변경을 하고 싶으면 ObserveOn이라는 오퍼레이터를 사용합니다. 

 

아래는 예시입니다.

Observable.Timer(TimeSpan.FromSeconds(1), Scheduler.MainThread)
    .Subscribe(x => Debug.Log(x))
    .AddTo(this);

Observable.Timer(TimeSpan.FromSeconds(1), Scheduler.MainThreadEndOfFrame)
    .Subscribe(x => Debug.Log(x))
    .AddTo(this);

var subject = new Subject<Unit>();
subject.AddTo(this);

subject
    .ObserveOn(Scheduler.ThreadPool)
    .Subscribe(_ =>
    {
        Debug.Log("Thread ID" + Thread.CurrentThread.ManagedThreadId);
    });

시간을 제어하는 오퍼레이터는 메인스레드가 디폴트라고 했으니 따로 붙이나 마나 겠지만 두번째 문단처럼

다른 스케쥴러에서 돌릴 수도 있습니다.

 

MainThread와 CurrentThread 사용 주의점!

메인 스레드는 멈추면 안되기 때문에 내부에서 코루틴을 이용한 처리가 되어있습니다.

하지만 CurrentThread로 스케쥴링을 잡을시에는 Thread.Sleep()이 호출 되어지기 때문에 메인스레드 모든 로직이

프리징 걸리게 됩니다.