Reactor 예외 발생에 대해

리액터에서 예외를 발생시키는 방법은 다양하다

그 중에 2가지 throw와 error 에 대해 알아보려고 한다.

 

throw는 가장 일반적으로 예외를 발생시키는 키워드라 생략하고, error에 대해 알아보겠다.

public static <T> Mono<T> error(Throwable error)

 

공식 페이지를 살펴보면 위와 같다.

대략적으로 마블 다이어그램을 보면, consumer로 부터 subscribe가 발생하면 그 때, error 안에 값인 Throwable 값이 emit된다.

Mono.error(new IllegalArgumentException())

예제를 보면 위와 같이 사용할 수 있다.

 

그런데!! throw와 error를 사용하는 시점은 좀 다르다. (아래 예제 참조)

fluxOrMono.flatMap(next -> Mono.error(new IllegalArgumentException())) 
fluxOrMono.map(next -> { throw new IllegalArgumentException(); }) 
fluxOrMono.doOnNext(next -> { throw new IllegalArgumentException(); })

그 이유는 Reactor에 대해 생각해보면 알 수 있다. Reactor는 Stream(Upstream/Downstream)의 연속이다. 때문에 단계별로 넘어갈 때 시그널을 주게된다. 그 중에 하나가 error인데 downstream으로 내려갈 때, error를 emit하면서 Stream에서 에러를 판단하는 것이다.

반면, throw는 에러 생성 방식이 조금 다르다. throw 예외가 발생하게 되면, Reactor에서 catch해서 onError로 Steam에 전파하기 때문에 사실상 error와 throw는 비슷하다.

다만, throw를 사용하는 것은 Reactor Rule 상 맞지 않는다. Stream Chaining 구조인 Reactor에서 에러 시그널을 받아 핸들링을 해야하는데 throw는 그 과정을 코어 내부에 위임하면서 가독성을 떨어뜨릴 수 있다고 한다. (가독성 부분은 잘 이해가...)

그래서 문서에서도 Mono.error를 통해 에러를 downstream으로 내리는 걸 추천한다고 한다.

 

그러나 위 코드상에서 map, doOnNext는 subscribe 신호에 의해 진행되는 비동기가 아닌 내부 동기이기 때문에 throw를 사용 할 수 밖에 없다... 그럼 어떻게?!!!

stackoverflow.com/questions/53595420/correct-way-of-throwing-exceptions-with-reactor/53596358#53596358

Correct way of throwing exceptions with Reactor

스택오버플로우에 괜찮은 방법이 설명되어 있는데, 바로 concatMap을 사용하는 것이다.

해당 메소드는 여러개의 비동기 Stream를 처리하기 위한 것인데, (아래 설명 참조)

In its essence, flatMap is designed to merge elements from the multiple substreams that is executing at a time. It means that flatMap should have asynchronous streams underneath so, they could potentially process data on the multiple threads or that could be a several network calls. Subsequently, such expectations impact implementation a lot so flatMap should be able to handle data from the multiple streams (Threads) (means usage of concurrent data structures), enqueue elements if there is a draining from another stream (means additional memory allocation for Queues for each substream) and do not violate Reactive Streams specification rules (means really complex implementation). Counting all these facts and the fact that we replace a plain map operation (which is synchronous) onto the more convenient way of throwing an exception using Flux/Mono.error (which does not change synchronicity of execution) leads to the fact that we do not need such a complex operator and we can use much simpler concatMap which is designed for asynchronous handling of a single stream at a time and has a couple of optimization in order to handle scalar, cold stream.

요약하면, 여러 Stream을 결과를 합치기 때문에 단순 flatmap을 사용하는 것보다 성능이 좋다. 또한 내부적으로 Stream 하나씩 하나씩 처리하기 때문에 복잡하게 map을 써서 변환하기 보다는 concatMap과 error를 사용하는 편이 더 편리하다고 한다.