Reactor publishOn과 subscribeOn, 어느 스레드에서 돌까?

이번 시간에는 Reactor의 publishOn과 subscribeOn이 체인의 어느 부분을 다른 스레드로 옮기는지, Spring 없이 Reactor만 쓰는 콘솔 프로그램에서 단계마다 스레드 이름을 찍어 확인하며 배워가는 시간입니다.

예를 들어서 우리가 Flux.range(1, 3) 뒤에 map을 세 개 이은 체인을 만들었다고 해볼게요.

첫 번째 map 뒤에 publishOn(Schedulers.parallel())을 넣으면 소스와 첫 번째 map은 main 스레드에서 실행돼요. 나머지 두 map과 구독 콜백은 parallel-1에서 실행됩니다. parallel-1은 작업에 스레드를 내주는 Reactor 스케줄러 parallel의 1번 스레드라고 보시면 됩니다.

같은 스케줄러를 subscribeOn으로 바꿔 체인 맨 끝에 붙이면 어떻게 될까요? 이번에는 소스부터 구독 콜백까지 전부 parallel-1에서 실행됩니다. 두 연산자를 여러 번 쓰면 어느 쪽이 적용되는지도 함께 봅니다.

실습 준비

이 글의 명령은 모두 실습 코드의 reactor-publishon-subscribeon/ 디렉터리에서 실행합니다.

• app/: Gradle 프로젝트(build.gradle, settings.gradle, 시나리오 0~8을 담은 src/main/java/Main.java)
• compose.yaml: 빌드한 jar를 실행하는 컨테이너
• env.sh: 빌드·실행 셸 함수

호스트에 JDK나 Gradle을 설치할 필요는 없어요. gradle:9.8.0-jdk21-noble 이미지 하나로 빌드와 실행을 모두 하거든요.

라이브러리는 io.projectreactor:reactor-core:3.8.7 하나뿐입니다. Main은 시나리오 번호를 인자로 받아 해당 실험만 돌려요.

공통으로 쓰는 도우미는 다음과 같습니다.

• log: 단계 이름과 값, 그 코드를 실행한 스레드 이름을 기록해요. -v를 주면 호출될 때마다 한 줄씩 찍고, 주지 않으면 summary가 단계별 스레드 이름만 모아서 찍습니다.
• source(): Flux.range(1, 3) 바로 뒤에 doOnNext를 붙여 소스가 값을 내보낸 스레드를 찍어요. sourceSub()는 소스가 구독된 스레드(onSub(src))도 함께 찍습니다.
• run: 체인을 main에서 구독하고 CountDownLatch로 끝날 때까지 기다려요.

main은 시작할 때 JVM이 보는 CPU 수와 parallel, boundedElastic 스케줄러의 기본 스레드 수를 찍습니다.

이번엔 compose.yaml을 볼게요. 빌드한 jar를 같은 이미지에서 java로 실행합니다.

services:
  app:
    image: gradle:9.8.0-jdk21-noble
    working_dir: /app
    volumes:
      - ./app:/app
    entrypoint: ["sh", "-c", "exec java -Xmx256m $$JAVA_OPTS -cp 'build/install/app/lib/*' Main \"$$@\"", "--"]
    environment:
      JAVA_OPTS: ${JAVA_OPTS:-}
    cpus: 2
    mem_limit: 768m

셸 함수는 env.sh에 들어 있어요. compose 프로젝트 이름은 rsch입니다.

• build: gradle 이미지로 installDist를 실행해요. Gradle이 받은 의존성은 이름 있는 볼륨 rsch-gradle에 남아 재빌드할 때 다시 받지 않습니다.
• run: 시나리오 번호와 옵션을 넘겨 앱을 한 번 실행해요.

export COMPOSE_PROJECT_NAME=rsch
build() { docker run --rm -v "$PWD/app":/app -v rsch-gradle:/home/gradle/.gradle -w /app gradle:9.8.0-jdk21-noble gradle --no-daemon -q installDist; }

run() { docker compose run --rm -T app "$@"; }

그럼 먼저 빌드해 볼게요.

source ./env.sh
build
ls app/build/install/app/lib

app.jar
jspecify-1.0.0.jar
reactive-streams-1.0.4.jar
reactor-core-3.8.7.jar

실습 1: publishOn과 subscribeOn을 붙이는 위치

시나리오 2는 같은 체인에 publishOn과 subscribeOn을 위치만 바꿔 네 가지로 붙여 봅니다. 기본 체인은 source().map(a).map(b).map(c)예요.

• (a): map(a) 뒤에 publishOn(parallel)
• (b): map(a) 앞(소스 바로 뒤)에 subscribeOn(parallel)
• (c): 맨 끝에 subscribeOn(parallel)
• (d): map(a) 뒤에 publishOn(parallel), 맨 끝에 subscribeOn(boundedElastic)

run 2

availableProcessors=2 parallel=2 boundedElastic=20
== (a) map(a) 뒤에 publishOn(parallel)
source      main
map(a)      main
map(b)      parallel-1
map(c)      parallel-1
subscribe   parallel-1
== (b) map(a) 앞에 subscribeOn(parallel)
source      parallel-2
map(a)      parallel-2
map(b)      parallel-2
map(c)      parallel-2
subscribe   parallel-2
== (c) 맨 끝에 subscribeOn(parallel)
source      parallel-1
map(a)      parallel-1
map(b)      parallel-1
map(c)      parallel-1
subscribe   parallel-1
== (d) map(a) 뒤 publishOn(parallel), 맨 끝 subscribeOn(boundedElastic)
source      boundedElastic-1
map(a)      boundedElastic-1
map(b)      parallel-2
map(c)      parallel-2
subscribe   parallel-2

두 번 실행했는데 스레드 번호까지 똑같았어요.

먼저 (a)부터 볼게요. publishOn 위에 있는 source와 map(a)는 그대로 main이고, 아래에 있는 map(b), map(c), 구독 콜백은 parallel-1입니다.

왜 이렇게 나뉠까요? publishOn은 위에서 받은 값을 자기 큐에 넣고, 스케줄러에서 얻은 워커로 큐의 값을 꺼내 아래로 넘기기 때문이에요.

워커는 스케줄러가 구독 하나에 내주는 실행 단위로, parallel에서는 스레드 하나에 묶여 있어요. 위쪽은 이미 main에서 실행되고 있으니 publishOn이 바꿀 수가 없습니다.

나머지 세 구성은 이렇습니다.

• (b): map(a) 앞, 소스 바로 뒤에 subscribeOn(parallel)을 붙였어요. 소스부터 구독 콜백까지 전부 parallel-2입니다.
• (c): 같은 subscribeOn(parallel)을 체인 맨 끝에 붙였어요. 결과는 (b)와 같습니다. 번호가 1과 2로 다른 건 구독할 때마다 parallel의 워커를 돌아가며 고르기 때문이고, 위치와는 상관없어요.
• (d): 맨 끝의 subscribeOn(boundedElastic)이 source와 map(a)를 boundedElastic-1로 옮겼고, publishOn 아래는 그대로 parallel-2입니다.

subscribeOn이 옮기는 것은 구독이에요. subscribe()를 부르면 구독 신호는 체인 아래에서 위로 올라갑니다. 맨 끝에 있는 subscribeOn을 지나면서 이 신호가 다른 스레드로 넘어가고, 소스는 그 스레드에서 구독과 요청을 받아요.

range는 구독과 요청을 받은 그 호출 안에서 바로 값을 내보내요. 그러니 소스부터 아래가 모두 그 스레드에서 실행됩니다. (d)에서는 그 흐름이 중간의 publishOn에서 끊긴 거예요.

javadoc은 두 연산자의 용도도 나눠 적고 있어요. subscribeOn은 느린 발행자, 예를 들어 블로킹 I/O에 쓰고, publishOn은 발행은 빠르고 소비가 느릴 때 씁니다.

실습 2: 여러 번 쓰면

시나리오 3은 같은 연산자를 두 번 이상 써 봅니다. 소스가 구독된 스레드를 보려고 sourceSub()을 쓰고, (b)에서는 두 subscribeOn 사이에도 doOnSubscribe를 넣었어요. 그 지점을 지나는 구독 신호의 스레드(onSub(mid))를 찍기 위해서입니다.

• (a): map(a) 뒤에 publishOn(parallel), map(b) 뒤에 publishOn(boundedElastic), 그다음 map(c)
• (b): 소스 바로 뒤에 subscribeOn(parallel), 맨 끝에 subscribeOn(boundedElastic)
• (b'): (b)에서 두 스케줄러를 맞바꿈
• (c): map(a) 뒤에 publishOn(parallel), map(b) 뒤에 subscribeOn(boundedElastic), 그 아래 publishOn(single)과 map(c)

run 3

availableProcessors=2 parallel=2 boundedElastic=20
== (a) publishOn(parallel) 다음 publishOn(boundedElastic)
onSub(src)  main
source      main
map(a)      main
map(b)      parallel-1
map(c)      boundedElastic-1
subscribe   boundedElastic-1
== (b) 소스 쪽 subscribeOn(parallel), 맨 끝 subscribeOn(boundedElastic)
onSub(mid)  boundedElastic-1
onSub(src)  parallel-2
source      parallel-2
map(a)      parallel-2
map(b)      parallel-2
map(c)      parallel-2
subscribe   parallel-2
== (b') 소스 쪽 subscribeOn(boundedElastic), 맨 끝 subscribeOn(parallel)
onSub(mid)  parallel-1
onSub(src)  boundedElastic-1
source      boundedElastic-1
map(a)      boundedElastic-1
map(b)      boundedElastic-1
map(c)      boundedElastic-1
subscribe   boundedElastic-1
== (c) publishOn(parallel) 아래 subscribeOn(boundedElastic), 그 아래 publishOn(single)
onSub(src)  boundedElastic-1
source      boundedElastic-1
map(a)      boundedElastic-1
map(b)      parallel-2
map(c)      single-1
subscribe   single-1

이번에도 두 번 실행한 결과가 번호까지 같았어요.

• (a) publishOn 두 번: 구간이 세 개로 나뉩니다. map(a)까지 main, map(b)는 parallel-1, map(c)와 구독 콜백은 boundedElastic-1이에요. publishOn은 쓸 때마다 그 아래를 새로 바꿉니다.
• (b) subscribeOn 두 번: 맨 끝의 boundedElastic은 구독 신호가 지나간 onSub(mid)에만 나타나요. 소스는 소스 쪽에 붙인 parallel-2에서 구독되고 값을 내보냈고, 그 아래 map과 구독 콜백도 모두 parallel-2입니다.
• (b') 순서를 바꾸면 결과도 뒤집혀요. 이번에는 소스 쪽에 붙인 boundedElastic-1이 적용됐습니다. 마지막에 붙인 쪽이 아니라 소스에 가까운 쪽이 적용되는 거예요.
• (c) 섞어 쓰기: subscribeOn을 publishOn 아래에 붙였는데도 소스와 map(a)를 boundedElastic-1로 옮겼어요. subscribeOn이 체인 어디에 있든 영향을 주는 범위는 소스부터 첫 publishOn까지예요.

실무에서는 라이브러리 메서드가 안에서 이미 subscribeOn을 붙인 Flux를 돌려줄 때 이 규칙을 만나게 됩니다. 호출하는 쪽에서 subscribeOn을 다시 붙여도 소스 스레드는 바뀌지 않아요. 값을 받은 뒤의 처리를 다른 스레드로 옮기고 싶다면 publishOn을 쓰셔야 합니다.

정리

범위: Reactor 3.8.7(reactor-core), JDK 21(gradle:9.8.0-jdk21-noble 이미지), CPU 2개와 메모리 768MB로 제한한 컨테이너, Spring 없는 콘솔 프로그램.

• map(a) 뒤에 둔 publishOn(parallel)은 그 아래 map(b), map(c), 구독 콜백만 parallel-1로 옮겼고, 위쪽은 main 그대로였습니다. 블로킹 호출이 아래쪽 연산자에 있으면 그 연산자 위에 publishOn을 두세요.
• subscribeOn(parallel)은 소스 바로 뒤에 붙여도 체인 맨 끝에 붙여도 소스부터 구독 콜백까지 모두 같은 parallel 스레드로 옮겼어요. 소스 안의 블로킹 호출은 subscribeOn으로 옮기세요.
• subscribeOn을 두 번 쓰면 소스에 가까운 쪽이 값의 스레드를 정했고, 맨 끝의 subscribeOn은 구독 신호만 한 번 더 옮겼습니다. 이미 subscribeOn이 붙은 Flux를 받아 뒤 처리를 옮기려면 publishOn을 쓰세요.