<div id="readability-page-1" class="page"><article>
 

컴파일도 잘 되었고, 테스트도 잘 통과 했는데, 실제 서비스를 실행하면 에러가 발생한다. 에러 메시지는 낯설고 뭐가 문제인지 모르겠다.

<div><pre><code>Rejecting additional inbound receiver. State=[terminated=false, cancelled=false, pending=0, error=false]
</code></pre></div>

<h2 id="원인">원인</h2>

결국 외부 서비스의 응답을 사용하는 과정에서 “또” 조회를 하는 바람에 발생한 에러였다.

WebFlux 로 개발 할 때, 테스트 및 자체적으로 생성하는 Mono(혹은 Flux) 는 계속 생성되기 때문에 고민없이 구현하다보면 잘 된다고 생각한 부분에서 오류를 만나게 된다.

예를 들어 아래와 같은 코드는 그냥 잘 된다. 코드에 정의된 Flux 는 계속 생성될 수 있기 때문이다.

<div><pre><code>public void someMethod() {
 List&lt;User&gt; users = List.of( ... some random generated users ... );
 Flux&lt;User&gt; userFlux = Flux.fromIterable(users);

 userFlux.map( ... )
 .subscribe(System.out::println);
 userFlux.map( ... )
 .subscribe(System.out::println);
}
</code></pre></div>

하지만 다음과 같은 코드는 에러가 발생한다. Upstream 을 다시 생성시킬수 없기 때문이다.

<div><pre><code>public void someMethod(Flux userFlux) {
 userFlux.map( ... )
 .subscribe(System.out::println);
 userFlux.map( ... )
 .subscribe(System.out::println);
}
</code></pre></div>

실제 서비스에서는 source 는 단 한 번 만 생성(publish)되는 경우가 많을텐데, 이를 습관적으로 보통의 Java 객체처럼 생각하고 여러번 구독하면 에러가 발생한다.

<h2 id="결론">결론</h2>

Mono(Flux) 는 한 번 구독하면 소모되어버린다는 점을 잊지 말자.

<h2 id="cache-로-해결">Cache 로 해결</h2>

How to use “cache” method of Mono

<ul>
 <li><a href="https://stackoverflow.com/a/67487325/8350542">https://stackoverflow.com/a/67487325/8350542</a></li>
 <li><a href="https://stackoverflow.com/a/52518786/8350542">https://stackoverflow.com/a/52518786/8350542</a></li>
</ul>

<div><pre><code>var cached = mono.cache();
</code></pre></div>

<h2 id="cache-에-대한-설명">Cache 에 대한 설명</h2>

요즘은 ChatGPT 에서 잘 설명해 준다.

<blockquote>
 <code>Mono</code>를 두 번 구독할 수 없는 제약은 리액티브 스트림의 원칙에 따른 것입니다. 리액티브 프로그래밍에서는 데이터의 비동기적인 흐름을 제어하기 위해 이러한 제약이 적용됩니다. 만약 동일한 데이터를 두 번 이상 사용해야 한다면 다음과 같은 방법을 고려할 수 있습니다:

 <ol>
 <li><code>cache()</code> 사용: <code>Mono</code>의 <code>cache()</code> 연산자를 사용하여 데이터를 캐시하고 여러 번 구독할 수 있도록 할 수 있습니다. <code>cache()</code> 연산자는 <code>Mono</code>를 구독할 때 데이터를 캐시하여 나중에 다시 사용할 수 있도록 합니다. 단, 데이터가 크거나 지속적으로 변경되는 경우 캐시 사용에 주의해야 합니다.</li>
 </ol>

</blockquote>

참고로 cache() 도 구독이다. cache() 하면서 source 를 읽어왔기 때문이다.

 </article></div>

Personal Tech Note

WebFlux 를 실행하며 만난 에러
컴파일도 잘 되었고, 테스트도 잘 통과 했는데, 실제 서비스를 실행하면 에러가 발생한다. 에러 메시지는 낯설고 뭐가 문제인지 모르겠다.
Rejecting additional inbound receiver. State=[terminated=false, cancelled=false, pending=0, error=false]
원인
결국 외부 서비스의 응답을 사용하는 과정에서 “또” 조회를 하는 바람에 발생한 에러였다.
WebFlux 로 개발 할 때, 테스트 및 자체적으로 생성하는 Mono(혹은 Flux) 는 계속 생성되기 때문에 고민없이 구현하다보면 잘 된다고 생각한 부분에서 오류를 만나게 된다.
예를 들어 아래와 같은 코드는 그냥 잘 된다. 코드에 정의된 Flux 는 계속 생성될 수 있기 때문이다.
public void someMethod() {
 List<User> users = List.of( ... some random generated users ... );
 Flux<User> userFlux = Flux.fromIterable(users);
 userFlux.map( ... )
 .subscribe(System.out::println);
 userFlux.map( ... )
 .subscribe(System.out::println);
}
하지만 다음과 같은 코드는 에러가 발생한다. Upstream 을 다시 생성시킬수 없기 때문이다.
public void someMethod(Flux userFlux) {
 userFlux.map( ... )
 .subscribe(System.out::println);
 userFlux.map( ... )
 .subscribe(System.out::println);
}
실제 서비스에서는 source 는 단 한 번 만 생성(publish)되는 경우가 많을텐데, 이를 습관적으로 보통의 Java 객체처럼 생각하고 여러번 구독하면 에러가 발생한다.
결론
Mono(Flux) 는 한 번 구독하면 소모되어버린다는 점을 잊지 말자.
Cache 로 해결
How to use “cache” method of Mono
https://stackoverflow.com/a/67487325/8350542
https://stackoverflow.com/a/52518786/8350542
var cached = mono.cache();
Cache 에 대한 설명
요즘은 ChatGPT 에서 잘 설명해 준다.
Mono를 두 번 구독할 수 없는 제약은 리액티브 스트림의 원칙에 따른 것입니다. 리액티브 프로그래밍에서는 데이터의 비동기적인 흐름을 제어하기 위해 이러한 제약이 적용됩니다. 만약 동일한 데이터를 두 번 이상 사용해야 한다면 다음과 같은 방법을 고려할 수 있습니다:
cache() 사용: Mono의 cache() 연산자를 사용하여 데이터를 캐시하고 여러 번 구독할 수 있도록 할 수 있습니다. cache() 연산자는 Mono를 구독할 때 데이터를 캐시하여 나중에 다시 사용할 수 있도록 합니다. 단, 데이터가 크거나 지속적으로 변경되는 경우 캐시 사용에 주의해야 합니다.
참고로 cache() 도 구독이다. cache() 하면서 source 를 읽어왔기 때문이다.

오늘의 실패사례 - WebFlux 여러번 구독 문제

댓글