이번 시간에는 Reactor의 백프레셔가 생산자와 구독자 사이에서 실제로 무엇을 해 주는지, request(n)을 직접 찍어 보고 onBackpressure* 연산자를 하나씩 붙여 측정하며 배워가는 시간입니다.
예를 들어서 생산자 스레드 하나가 1KB짜리 값을 초당 1만 개씩 만들고, 구독자는 값 하나를 처리하는 데 10ms 남짓 걸린다고 해볼게요.
이 생산자와 구독자 사이에 onBackpressureBuffer, onBackpressureDrop, onBackpressureLatest를 하나씩 붙여 메모리와 잃은 값의 개수를 비교해 볼게요.
백프레셔와 request(n)
백프레셔는 값을 받는 쪽이 받을 수 있는 개수를 먼저 알리고, 보내는 쪽은 그 개수까지만 보내게 해서 속도를 받는 쪽에 맞추는 장치예요. Reactive Streams 명세에서는 구독자가 Subscription.request(n)으로 개수를 알려요.
range처럼 요청받을 때 값을 만드는 소스는 구독자 속도에 맞출 수 있어요. 하지만 타이머나 외부 이벤트는 구독자가 요청하지 않았다고 해서 기다려 주지 않죠.
요청을 지키지 않는 소스와 느린 구독자 사이에서 남는 값을 어떻게 할지 정하는 연산자가 onBackpressure*예요.
실습 준비
이 글의 명령은 모두 실습 코드의 reactor-backpressure/ 디렉터리에서 실행합니다.
compose.yaml: gradle 이미지로 앱을 실행하는 서비스env.sh: 셸 함수build,runapp/: Gradle 프로젝트(build.gradle,settings.gradle, 시나리오를 모두 담은Main.java)
Spring 없이 Reactor Core만 쓰는 콘솔 프로그램 하나로 실습할게요. 시나리오는 실행 인자로 골라요. 호스트에 JDK나 Gradle을 설치하지 않고 gradle:9.8.0-jdk21-noble 이미지 하나로 빌드와 실행을 모두 합니다.
build.gradle은 application 플러그인과 reactor-core 하나만 써요.
아래는 compose.yaml이에요. 같은 이미지로 java를 실행해요.
CPU 2개, 메모리 768MB로 제한하고, 힙은 JAVA_OPTS로 정해요. 기본은 -Xmx256m이고 메모리를 다 쓰는 회차에서만 128MB로 줄여요. 컨테이너 상한이 힙보다 넉넉해야 컨테이너가 먼저 강제 종료되지 않고 JVM이 OutOfMemoryError를 내거든요.
services:
app:
image: gradle:9.8.0-jdk21-noble
working_dir: /app
volumes:
- ./app:/app
environment:
JAVA_OPTS: ${JAVA_OPTS:--Xmx256m}
entrypoint: ["sh", "-c", "exec java $$JAVA_OPTS -cp 'build/install/app/lib/*' Main \"$$@\"", "java"]
cpus: 2
mem_limit: 768m
셸 함수는 env.sh에 두고 source ./env.sh로 불러 써요. COMPOSE_PROJECT_NAME을 rbp로 정해 두었어요.
build: gradle 이미지로installDist를 실행해app/build/install/app/lib에 jar를 모아요. 받은 의존성은 이름 있는 볼륨rbp-gradle에 남아요.run: 시나리오 이름을 인자로 받아 앱을 한 번 실행해요.
app/src/main/java/Main.java 하나에 공용 코드와 시나리오 메서드가 모두 들어 있어요.
monitor: 별도 데몬 스레드가 500ms마다 사용 힙(totalMemory - freeMemory)과 생산·소비·버린 개수를 찍어요. GC 로그는 켜지 않았어요.Slow: 느린 구독자예요. 구독하면request(1)만 하고, 값 하나마다 10ms를 자고 나서 다음 하나를 요청해요. 받은 seq의 앞 10개와 마지막 값을 기억합니다.producer: 요청과 상관없이 값을 만드는 생산자예요.Flux.create안에서producer라는 스레드를 새로 띄우고, 그 스레드가n개를 정해진 간격으로sink.next해요. 간격은 기본 100µs, 초당 1만 개예요.ReqLog: 위로 올라가는request를 처음 몇 번만 찍고, 나머지는 크기별로 세었다가 끝에 한 줄로 요약해요.
이제 빌드해 볼게요.
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
세 실습은 같은 생산자(1KB 값, 초당 1만 개)와 같은 구독자(값 하나에 10ms)를 두고 연산자만 바꿔요. 생산자는 OverflowStrategy.IGNORE로 만들어서 create의 내부 버퍼를 끄고, 전략 연산자를 생산자 바로 뒤에 붙여요. 구독자는 publishOn(Schedulers.single(), 16) 뒤의 single-1 스레드에서 돌아요.
로그에는 두 군데의 요청이 찍혀요.
생산자 <-: 전략 연산자가 생산자에게 보낸 요청이에요. 모든 회차에서 처음 한 번9223372036854775807이었어요. 전략 연산자는 위로 무제한을 요청해 들어오는 값을 모두 받고, 넘치는 값을 자기가 처리해요.전략 <-:publishOn이 전략 연산자에게 보낸 요청이에요. 처음 16개, 이후 12개씩이에요.
실습 1: onBackpressureBuffer(), 상한 없이 쌓기
인자 없는 onBackpressureBuffer()는 구독자가 요청하지 않은 값을 상한 없는 버퍼에 모두 쌓아요. 생산 개수를 100만 개(약 1GB)로 두고 힙을 128MB로 줄여 볼게요.
-XX:+ExitOnOutOfMemoryError도 줘요. OutOfMemoryError는 값을 만들고 버퍼에 넣는 생산자 스레드에서 날 가능성이 커요. 이 옵션이 없으면 그 스레드만 죽고 main과 구독자 스레드는 살아 있을 수 있어서, 첫 OutOfMemoryError에서 JVM을 바로 끝내게 했어요.
JAVA_OPTS="-Xmx128m -XX:+ExitOnOutOfMemoryError" run s5-unbounded; echo "exit=$?"
0.04s [main] 시나리오 s5-unbounded maxHeap=123MB
0.21s [main] 전략 <- request(16) produced=0 consumed=0
0.21s [main] 생산자 <- request(9223372036854775807) produced=0 consumed=0
0.21s [producer] 생산 시작
0.21s [single-1] 첫 소비 seq=0
0.35s [single-1] 전략 <- request(12) produced=1351 consumed=12
0.48s [single-1] 전략 <- request(12) produced=2695 consumed=24
0.66s [monitor] heap=6MB produced=4452 consumed=39 dropped=0
1.16s [monitor] heap=11MB produced=9461 consumed=84 dropped=0
1.66s [monitor] heap=16MB produced=14471 consumed=129 dropped=0
2.16s [monitor] heap=21MB produced=19484 consumed=174 dropped=0
...
5.18s [monitor] heap=52MB produced=49704 consumed=419 dropped=0
...
10.24s [monitor] heap=104MB produced=100261 consumed=805 dropped=0
10.74s [monitor] heap=109MB produced=105281 consumed=839 dropped=0
11.25s [monitor] heap=114MB produced=110391 consumed=875 dropped=0
11.75s [monitor] heap=118MB produced=115430 consumed=906 dropped=0
12.26s [monitor] heap=123MB produced=120451 consumed=941 dropped=0
Terminating due to java.lang.OutOfMemoryError: Java heap space
exit status 3
exit=3
출력을 보면 사용 힙이 0.5초마다 5MB씩, 한 번도 줄지 않고 늘었어요. 구독자가 받지 못한 값이 모두 버퍼에 남아 있으니 GC가 치울 게 없거든요.
12.26초에 123MB(이 JVM이 쓸 수 있는 최대 힙)에 닿은 뒤 JVM이 종료 코드 3으로 끝났어요. 3은 -XX:+ExitOnOutOfMemoryError가 JVM을 끝낼 때 쓰는 코드이고, 컨테이너 메모리 상한을 넘어 커널이 프로세스를 죽였다면 137이 나왔을 거예요.
세 번 모두 같은 모양이었어요. 마지막 모니터 줄은 11.67~12.26초, 사용 힙 118~123MB, 생산 115,394~120,451개였고 구독자가 받은 값은 809~941개였어요.
생산이 소비보다 계속 빠르면 onBackpressureBuffer()는 문제가 드러나는 시점을 힙이 찰 때까지 미룰 뿐이에요.
실습 2: onBackpressureDrop, 못 받은 값은 버리기
onBackpressureDrop(콜백)은 아래에서 요청이 없을 때 들어온 값을 그 자리에서 버리고 콜백으로 넘겨요. 버퍼가 없어요. 여기서부터는 생산 개수를 10만 개로 두고 힙은 기본 256MB예요.
run s6-drop
0.01s [main] 시나리오 s6-drop maxHeap=247MB
0.13s [main] 전략 <- request(16) produced=0 consumed=0
0.13s [main] 생산자 <- request(9223372036854775807) produced=0 consumed=0
0.13s [producer] 생산 시작
0.13s [single-1] 첫 소비 seq=0
0.28s [single-1] 전략 <- request(12) produced=1473 consumed=12
0.43s [single-1] 전략 <- request(12) produced=2990 consumed=24
0.59s [monitor] heap=4MB produced=4627 consumed=38 dropped=4575
...
5.15s [monitor] heap=3MB produced=50259 consumed=357 dropped=49895
...
10.13s [producer] 생산 끝 produced=100000
10.23s [monitor] heap=2MB produced=100000 consumed=715 dropped=99276
10.34s [single-1] onComplete consumed=724
10.36s [main] 생산자 <- request 횟수=1 크기별 횟수={9223372036854775807=1}
10.36s [main] 전략 <- request 횟수=61 크기별 횟수={12=60, 16=1}
10.36s [main] 결과 produced=100000 consumed=724 dropped=99276 maxHeap=5MB
10.36s [main] 받은 seq 앞 10개=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 마지막=99853 종료=onComplete
받은 724개와 버린 99,276개를 더하면 100,000개예요. 생산이 끝나고 0.2초 만에 구독도 끝났어요. 남아 있는 값이 publishOn 큐의 몇 개뿐이기 때문이에요.
여기서 눈여겨볼 건 마지막으로 받은 seq가 99853이라는 점이에요. 생산자가 마지막으로 만든 99999가 아니죠. 그 값이 들어올 때 구독자가 요청하지 않고 있었다면 그 값도 버려져요.
세 번 실행에서 받은 값은 712~724개, 마지막 seq는 98670~99853으로 회차마다 달랐어요. 사용 힙은 2~5MB였어요.
실습 3: onBackpressureLatest, 마지막 하나만 남기기
onBackpressureLatest()는 요청이 없는 동안 들어온 값 중 가장 최근 값 하나만 들고 있다가, 다음 요청이 오면 그 값을 보내요. 그사이 덮어쓴 값은 버려지지만 콜백이 없어서, 이 실습에서는 버린 개수를 세지 않았어요.
run s6-latest
0.01s [main] 시나리오 s6-latest maxHeap=247MB
0.12s [main] 전략 <- request(16) produced=0 consumed=0
0.12s [main] 생산자 <- request(9223372036854775807) produced=0 consumed=0
0.12s [producer] 생산 시작
0.13s [single-1] 첫 소비 seq=0
0.28s [single-1] 전략 <- request(12) produced=1596 consumed=12
0.45s [single-1] 전략 <- request(12) produced=3263 consumed=24
0.60s [monitor] heap=4MB produced=4723 consumed=34 dropped=0
...
5.15s [monitor] heap=3MB produced=50216 consumed=364 dropped=0
...
10.12s [producer] 생산 끝 produced=100000
10.17s [monitor] heap=2MB produced=100000 consumed=721 dropped=0
10.22s [single-1] onComplete consumed=725
10.23s [main] 생산자 <- request 횟수=1 크기별 횟수={9223372036854775807=1}
10.23s [main] 전략 <- request 횟수=61 크기별 횟수={12=60, 16=1}
10.23s [main] 결과 produced=100000 consumed=725 dropped=0 maxHeap=5MB
10.23s [main] 받은 seq 앞 10개=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 마지막=99999 종료=onComplete
받은 개수는 Drop과 비슷한 725개지만, 마지막으로 받은 seq가 생산자가 마지막으로 만든 99999예요. 세 번 모두 725개를 받았고 마지막 seq는 99999였어요. 사용 힙은 1~5MB였어요.
Drop은 요청이 없을 때 들어온 값을 버리니까 마지막 값도 사라질 수 있어요. 반면 Latest는 가장 최근 값을 남겨 두니까 끝에 가장 새로운 값이 반드시 도착합니다.
정리
범위: Reactor Core 3.8.7, JDK 21(gradle:9.8.0-jdk21-noble 이미지), 컨테이너 CPU 2개·메모리 768MB, 생산자 초당 1만 개(1KB 값), 구독자 값 하나에 10ms, 시나리오마다 3번 실행.
- 상한 없는
onBackpressureBuffer()는 힙 128MB에서 11.67~12.26초 뒤OutOfMemoryError로 끝났어요. 생산이 소비보다 계속 빠른 소스에 상한 없는 버퍼를 쓰지 마세요. - 이 글에 싣지 않은 추가 측정에서는 상한을 준
onBackpressureBuffer(1000)이 seq 1016에서 넘쳐 생산을 멈췄고 사용 힙은 4MB였지만, 구독자는 13.80~14.05초에야 오류를 받았어요. 요청을 넘는 값이 오면 바로 오류를 내는onBackpressureError는 16개를 받은 뒤 0.32~0.37초에 오류를 받았어요. 상한 버퍼의 오류는 버퍼를 다 비운 뒤에 오니 넘친 순간은 콜백으로 잡아 주세요. onBackpressureDrop은 마지막으로 받은 seq가 98670~99853으로 회차마다 달랐고,onBackpressureLatest는 세 번 모두 마지막 seq 99999를 받았으며 둘 다 사용 힙이 5MB 이하였어요. 가장 새 값이 반드시 도착해야 하면 Drop이 아니라 Latest를 쓰세요.