Backpressure Example 코드 질문드립니다

Kevin의 알기 쉬운 Spring Reactive Web Applications: Reactor 1부

Backpressure Example 코드 설명

작성

205

기존 예제 (sleep 시간이 5L인 경우)에선 Exception이 발생하는 것을 확인했습니다. 하지만 sleep 시간을 더 늘리니까 Erorr가 발생하지않았습니다.

제 예상대로라면 버퍼가 더 빨리 차기 때문에 에러가 발생해야하는데, 동작이 이해가 되지 않습니다.

    public class BackpressureStrategyErrorExample {
    public static void main(String[] args) {
        Flux
                .interval(Duration.ofMillis(1L))
                .onBackpressureError()
                .doOnNext(Logger::doOnNext)
                .publishOn(Schedulers.parallel())
                .subscribe(data -> {
//                    왜 50L, 500L로 하면 에러가 발생하지 않을까?
                        TimeUtils.sleep(500L);
                        Logger.onNext(data);
                    },
                    error -> Logger.onError(error));

        TimeUtils.sleep(5000L);
    }
}

java spring project-reactor webflux

답변 2

Kevin

지식공유자

안녕하세요? 답변이 조금 늦었습니다. 육아를 병행하고 있어서 답변이 늦어진 점 양해 부탁드릴게요.

Backpressure Error 전략은 Downstream 쪽에서 Upstream 쪽의 속도에 대응하지 못할 경우 에러를 발생시키는 전략인데요. 에러가 발생할 때 아래와 같은 에러 로그가 출력됩니다.

reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)

이 문장에서 bounded queue가 일종의 버퍼를 의미하는데요. Downstream 쪽에서 내부적으로 데이터를 처리할 때 사용하는 버퍼라고 생각하시면 될 것 같습니다.

그런데 여기서의 버퍼는 Backpressure Buffer 전략에서의 Buffer와 조금 다른 동작을 합니다.

Downstream 쪽에서 지연 시간을 짧게 주면 버퍼에 그만큼 더 빠르게 데이터가 쌓이게 되고, 지연 시간을 길게 주었을 때 데이터가 그만큼 천천히 쌓인다고 생각하시면 될 것 같습니다.(내부적으로 버퍼에 쌓이는 동작을 조절한다고 보시면 될 것 같습니다.)

Backpressure의 내부 메커니즘을 100 퍼센트 다 이해하고 있다고 한다면 솔직히 거짓말일텐데 아무튼 쉽게 생각해서 다음 데이터를 처리하기 위해 넉넉한 시간을 가질 수 있다면 여유를 가지고 처리를 할 수 있는거라고 생각해주시면 좋을 것 같아요.

데이터를 처리할 준비가 안되어 있는데 무작정 데이터가 들어오면 과부하가 걸릴테니까요.

다만, 이 경우에는 여유는 생기겠지만 처리할 수 있는 데이터의 건 수는 그만큼 줄어들테니 적절한 조건을 찾아야 될테구요.

적절한 답변이 되셨으면 좋겠습니다.

감사합니다.

Choi hwan

저도 이거에 대해서 많은 고민이 있었는데 제가 찾아낸 해답은 아래와 같습니다.

먼저 질문 주신 부분의 코드에서는 에러가 발생한 상황이 맞습니다. 그런데 왜 로그에서 에러가 보이지 않는가하면 아직 에러가 로깅이 출력되지 못한 상황으로 보여집니다.

아래의 TImeUitls.sleep(5000L) 을 늘리게되면 제가 첨부한 사진처럼 OnNext() 가 전부 출력된 이후에 error 로그가 찍히게 됩니다.

스크린샷 2025-05-26 오후 9.04.23.png.webp

다른 쓰레드에서 동작하니 중간에 발생하고 로그를 적어야하는거아니냐? 라는 의문이 드실수도 있는데 GPT를 통한 답변입니다.

예외는 parallel-2 (emit 쪽) 스레드에서 발생했지만,
subscribe()에 등록된 onError 콜백은 parallel-1 (consume 쪽) 워커에서 실행되는 것.

즉:

onBackpressureError() → queue.offer() 실패 → 예외 발생 ← 이건 emit 쪽 (parallel-2)
이 예외는 downstream으로 전파됨
리액터는 전파된 예외를 subscribe(onNext, onError)에서 처리함
subscribe에서 실행되는 onError는 publishOn() 이후이므로 consume 워커 (parallel-1)
에서 실행

라고 하네요. 저도 공부하면서 궁금했던 내용이였는데 비슷한 질문인것 같아서 답변남깁니다.

인프런 커뮤니티 질문&답변

Backpressure Example 코드 질문드립니다