[Spring] 리액티브 프로그래밍 학습 6 - Reactor (feat. Scheduler와 publishOn& subscribeOn)

Sihwan Kim·2024년 3월 28일

Reactive Programming

목록 보기

1/9

Scheduler

특정 설정을 하지 않으면 publisher는 subscribe를 호출한 caller의 쓰레드에서 실행된다.

Scheduler는 Publish나 Subscribe에서 어떤 쓰레드에서 실행할지 지정해줄 수 있다.

Immediate Scheduler

Flux.just(1,2,3,4)
	.subscribeOn(
    	Schedulers.immediate()
    )
    .subscribe(
    	logger.info("value: " + value);
    )

subscribe를 호출한 쓰레드에서 즉시 실행한다.
즉, Schedulers를 사용하지 않은 것과 같다.

Single Scheduler

	Flux.just(1,2,3,4)
		.subscribeOn(
			Schedulers.single()
		).subscribe {
			logger.info{ "value : $it" }
		}
	Thread.sleep(1000)

캐싱된 1개의 쓰레드풀을 제공하며 하나의 쓰레드에서만 실행된다.

Parallel Scheduler

	for(i : Int in 0..100){
		val idx = i
		Flux.create{sink ->
			sink.next(idx)
		}
			.subscribeOn(
				Schedulers.parallel()
			).subscribe {
				logger.info{ "value : $it" }
			}
	}
    Thread.sleep(1000);

캐싱된 n개 크기의 쓰레드풀을 제공한다. 기본적으로 CPU 코어 수
내 PC는 코어수가 6개다.

BoundedElastic Scheduler

	for(i : Int in 0..30){
		val idx = i
		Flux.create{sink ->
			sink.next(idx)
		}
			.subscribeOn(
				Schedulers.boundedElastic()
			).subscribe {
				logger.info{ "value : $it" }
			}
	}

캐싱된 스레드풀이지만, 크기는 고정되지 않은다.
재사용할 수 있는 쓰레드가 있다면 사용, 없으면 새로생성
생성 가능한 스레드 수는 제한되어있다. (기본 cpu x10)

publishOn과 subscribeOn

publishOn과 subscribeOn을 이용해서 Scheduler를 어디에 배치하느냐를 정할 수 있다.

스트림이 publishOn을 거치면 그 후에 publishOn에 명시한 Scheduler가 적용된다.
subscribeOn은 지정하면 구독하는 시작부터 publsihOn을 만날 때까지 명시한 Scheduler가 적용된다.

예시코드

	for(i : Int in 0..10){
		val idx = i
		Flux.create{sink ->
			sink.next(idx)
			logger.info{"start : ${idx}"}
		}
			.publishOn(Schedulers.single())
			.doOnEach {
				logger.info{"middle : ${it.get()}"}
			}
			.subscribeOn(
				Schedulers.boundedElastic()
			).subscribe {
				logger.info{ "end : $it" }
			}
	}