[📕 Reactive] 6. Sinks와 Scheduler

Sinks 란 ?

Reactor에서 제공하는 Sinks는 비동기 데이터를 안전하게 발행할 수 있는 유틸리티로, Publisher 및 Subscriber 역할을 지원한다.

 

일반적으로 generate() operator나 create() operator는 싱글스레드 기반에서 signal을 전송하는 데 사용하는 반면, 

Sinks는 멀티스레드 방식으로 signal을 전송해도 스레드 안전성을 보장하기 때문에 예기치 않은 동작으로 이어지는 것을 방지해준다.

 

💡스레드 안전성이란 ? 

함수나 변수 같은 공유 자원에 동시 접근할 경우에도 프로그램의 실행에 문제가 없음을 의미한다.
동시 접근을 감지하고, 동시 접근하는 스레드 중 하나가 빠르게 실패함으로써 스레드 안전성을 보장한다. 

 

 

 

Sinks 종류 및 특징

 

✳️ Sinks.One

• Sinks.one() 메서드를 사용해서 한 건의 데이터를 전송
• Sinks.One으로 아무리 많은 수의 데이터를 emit한다 하더라도 처음 emit한 데이터는 정상적으로 emit되지만 나머지 데이터들은 drop됨

 

✳️ Sinks.Many

• Sinks.many() 메서드를 사용해서 여러 건의 데이터를 여러 가지 방식으로 전송
• ManySpec이라는 인터페이스를 리턴 👉ManySpec은 데이터 emit을 위한 여러 가지 기능 정의
• UnicastSpec은 단 하나의 Subscriber에게만 데이터를 emit
• MulticastSpec은 하나 이상의 Subscriber에게 데이터를 emit
• MulticastReplaySpec은 emit된 데이터 중에서 특정 시점으로 되돌린 데이터부터 emit

 

 

Scheduler 란 ?

Reactor에서 사용되는 Scheduler는 Reactor Sequence에서 사용되는(비동기 프로그래밍을 위해 사용되는) 스레드를 관리해주는 관리자 역할을 한다.

즉, 어떤 스레드에서 무엇을 처리할지 제어함으로써 개발자가 직접 스레드를 제어해야하는 부담을 덜어준다.

 

✳️ 물리적 스레드

• 하드웨어와 관련된 스레드 (CPU 코어)
• 병렬성과 관련 있음
• 물리적인 스레드가 실제로 동시에 실행되기 때문에 여러 작업을 동시에 처리

 

✳️ 논리적 스레드 

• 프로세스 내에서 실행되는 세부 작업의 단위
• 동시성과 관련 있음

 

 

 

 

Scheduler를 위한 전용 Operator

✳️ subscribeOn()

• 구독이 발생한 직후 실행될 스레드를 지정
• 원본 Publisher의 동작을 수행하기 위한 스레드

@Slf4j
public class Example{
	public static void main(String[] args) throws InterruptedException {
    	Flux.fromArray(new Integer[] {1, 3, 5, 7})
        	.subscribeOn(Schedulers.boundedElastic())
            .doOnNext(data -> log.info("# doOnNext: {}", data))
            .doOnSubscribe(subscription -> log.info("# doOnSubscribe"))
            .subscribe(data -> log.info("# onNext: {}", data));
            
        Thread.sleep(500L);
    }
}

• boundedElastic이라는 유형의 Scheduler 지정 
• doOnNext를 사용해 원본 Flux에서 emit되는 데이터를 로그로 출력
• doOnSubscribe를 사용해 구독이 발생한 시점에 추가적인 어떤 처리가 필요할 경우 해당 처리 동작 추가 
  • 현재 여기서는 구독이 발생한 시점에 실행되는 스레드가 무엇인지 확인
• 구독이 발생한 직후부터는 원본 Flux의 동작을 처리하는 스레드가 변경

 

 

✳️ publishOn()

• downstream으로 signal을 전송할 때 실행되는 스레드를 제어하는 역할 
• publishOn()을 기준으로 아래쪽인 Downstream의 실행 스레드를 변경
• Scheduler를 지정함으로서 해당 Scheduler의 특성을 가진 스레드로 변경 

@Slf4j
public class Example{
	public static void main(String[] args) throws InterruptedException {
    	Flux.fromArray(new Integer[] {1, 3, 5, 7})
            .doOnNext(data -> log.info("# doOnNext: {}", data))
            .doOnSubscribe(subscription -> log.info("# doOnSubscribe"))
            .publishOn(Schedulers.parallel())
            .subscribe(data -> log.info("# onNext: {}", data));
            
        Thread.sleep(500L);
    }
}

 

 

✳️ parallel()

• subscribeOn()과 publishOn()은 동시성을 가지는 논리적인 스레드에 해당
• parallel()은 병렬성을 가지는 물리적인 스레드에 해당
• 라운드 로빈 방식으로 CPU 코어 개수만큼의 스레드를 병렬로 실행

@Slf4j
public class Example{
	public static void main(String[] args) throws InterruptedException {
    	Flux.fromArray(new Integer[] {1, 3, 5, 7, 9, 11, 13, 15, 17, 19})
        	.parallel()
            .runOn(Schedulers.parallel())
            .subscribe(data -> log.info("# onNext: {}", data));
            
        Thread.sleep(100L);
    }
}

• parallel()은 emit되는 데이터를 CPU의 논리적인 코어 수에 맞게 사전에 골고루 분배하는 역할 수행
• 실제로 병렬 작업을 수행할 스레드의 할당은 runOn()이 담당

 

 

 

publishOn()과 subscribeOn()의 동작 이해

• publishOn()과 subscribeOn()을 사용하지 않을 경우 모두 main 스레드에서 실행됨 
• 한 개 이상의 publishOn()을 사용하면 실행 스레드를 목적에 맞게 적절하게 분리 가능

publishOn()을 2번 사용할 경우

 

subscribeOn()과 publishOn()을 함께 사용할 경우

 

• subscribeOn()은 구독이 발생한 직후에 실행될 스레드를 지정하기 때문에 fromArray()는 A 스레드에서 실행됨 

 

 

 

 

Scheduler의 종류

✳️ Schedulers.immediate()

• 별도의 스레드를 추가적으로 생성하지 않고, 현재 스레드에서 작업을 처리하고자 할 때 사용

🙋‍♀️ 현재 스레드에서 작업을 처리하고 싶다면 publishOn()을 한 번만 사용하면 되지 왜 굳이 Schedulers.immediate()을 사용하나요 ??

👉 publishOn()이 여러 번 호출된 상황에서, 이후 작업들이 현재 스레드에서 계속 실행되기를 보장하려는 경우 사용하거나 스케줄링 전환 없이 현재 스레드에서 실행하고자 하는 의도 명확히 표현하고자 함 !

 

✳️ Schedulers.single()

• 스레드 하나만 생성해서 Scheduler가 제거되기 전까지 재사용
• 하나의 스레드로 다수의 작업을 처리해야 되므로 지연 시간이 짧은 작업을 처리하는 것이 효과적

 

✳️ Schedulers.newSingle()

• 호출할 때마다 새로운 스레드 하나를 생성
• Schedulers.newSingle("new-single", true)
  • 첫 번째 파라미터에는 생성할 스레드의 이름 지정
  • 두 번째 파라미터에는 이 스레드를 데몬 스레드로 동작하게 할지 여부 설정

 

✳️ Schedulers.boundedElastic()

• ExecutorService 기반의 스레드 풀을 생성한 후, 그 안에서 정해진 수만큼의 스레드를 사용하여 작업을 처리하고 작업이 종료된 스레드는 반납하여 재사용
• Blocking I/O 작업에 최적화되어 있음

 

✳️ Schedulers.parallel()

• Non-Blocking I/O에 최적화되어 있음
• CPU 코어 수만큼의 스레드 생성

 

✳️ Schedulers.fromExecutorService()

• 기존에 이미 사용하고 있는 ExecutorService가 있다면 이 ExecutorService로부터 Scheduler를 생성
• 이 방식 권장 X

 

✳️ Schedulers.newXXXX()

• 스레드 이름, 생성 가능한 디폴트 스레드 개수, 스레드의 유휴 시간, 데몬 스레드로의 동작 여부 등을 직접 지정해서 커스텀 스레드 풀을 새로 생성