Backpressure Example 코드 질문드립니다

Question

기존 예제 (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);
    }
}

Kevin · Answer

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

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 퍼센트 다 이해하고 있다고 한다면 솔직히 거짓말일텐데 아무튼 쉽게 생각해서 다음 데이터를 처리하기 위해 넉넉한 시간을 가질 수 있다면 여유를 가지고 처리를 할 수 있는거라고 생각해주시면 좋을 것 같아요.

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

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

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

감사합니다.

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

Backpressure Example 코드 질문드립니다