Scheduler
쓰레드를 관리하는 관리자의 역할
Reactor Sequence 상에서 처리되는 동작들을 하나 이상의 쓰레드에서 동작하도록 별도의 쓰레드를 제공해 준다.
Reactor의 Scheduler는 복잡한 멀티쓰레딩 프로세스를 단순하게 해준다.
전용 Operator
- 적절한 상황에 맞는 쓰레드를 추가로 생성하는 Operator
range() Operator를 이용해 1부터 10개의 숫자를 emit한 후, emit된 숫자 데이터 중에서 filter() Opertor를 이용해 짝수만 필터링 한 뒤, 필터링 된 데이터를 map() Operator에서 2를 곱해서 Subscriber에게 전달하는 간단한 예제다.
Scheduler를 지정하지 않았기 때문에 실행 결과를 보면 main 쓰레드에서 실행이 되는 것을 확인할 수 있다.
subscribeOn()Operator
(1)과 같이 subscribeOn() Operator 내부에 Schedulers.boundedElastic() 같은 Scheduler를 지정하면 구독 직후에 실행되는 쓰레드가 main 쓰레드에서 Scheduler로 지정한 쓰레드로 바뀌게 된다.
실행 결과를 보면 doOnSubscribe() Operator에서 출력되는 로그는 main 쓰레드에서 실행이 되고, Subscriber에서 출력되는 데이터는 boundedElastic-1 이라는 이름을 가진 쓰레드에서 실행이 되는 것을 확인할 수 있다.
doOnSubscribe() Operator는 구독 발생 직후에 트리거 되는 Operator로써 구독 직후에 어떤 동작을 수행하고 싶다면 doOnSubscribe() 에 로직을 작성하면 된다.
그런데 doOnSubscribe() Operator는 main 쓰레드에서 실행이 된다. 그리고 이후에 range() Operator에서 데이터를 생성하고 emit하는 단계부터는 boundedElastic-1 에서 실행된다.
이를 통해서 알 수 있는 사실은 subscribeOn() 의 경우, 구독 직후 실행되는 Operator 체인의 실행 쓰레드를 Scheduler로 지정한 쓰레드로 변경한다는 것이다.
publishOn()Operator
(2)와 (4)의 doOnNext() 는 doOnNext() 바로 앞에 위치한 Operator가 실행될 때, 트리거 되는 Operator다. 여기서는 filter() 와 map() Operator가 어느 쓰레드에서 실행이 되는지 확인하기 위한 용도로 사용하고 있다.
실행 결과를 보면, publishOn() Operator가 fiter()와 map() Operator 앞에 추가되면서 publishOn() 이 추가될 때 마다 실행 쓰레드가 바뀌는 것을 확인할 수 있다.
subscribeOn() 은 구독 시점 직후의 실행 흐름을 다른 쓰레드로 바꾸는데 사용한다.
range() Operator처럼 원본 데이터를 생성하고, 생성한 데이터를 emit하는 작업이 구독 직후에 실행이 된다.
즉, subscribeOn() 은 주로 데이터 소스에서 데이터를 emit하는 원본 Publisher의 실행 쓰레드를 지정하는 역할을 한다.
publishOn() 은 전달 받은 데이터를 가공 처리하는 Operator 앞에 추가해서 실행 쓰레드를 별도로 추가하는 역할을 한다.
주의 할 것은
publishOn()은 Operator 앞에 여러번 추가할 경우 별도의 쓰레드가 추가로 생성되지만subscribeOn()은 여러 번 추가해도 하나의 쓰레드만 추가로 생성된다.
subscribeOn() 에서는 주로 Schedulers.boundedElastic() 을 사용하고, publishOn() 에서는 주로 Schedulers.parallel() 를 사용한다.
