[Spring Webflux] 컨텍스트 (Context)

각각의 스레드에서 상태 값을 저장하여 사용할 수 있는 스레드로컬과 유사하게 리액터에서는 컨텍스트를 사용해서 리액터 시퀀스상에 상태 값을 저장하고, 저장된 상태 값을 사용할 수 있다.

먼저 리액터에서의 컨텍스트가 무엇인지 살펴보고, 컨텍스트의 특징과 자주 사용되는 API를 살펴보자

컨텍스트란

• Reactor Sequence상에서 상태를 저장할 수 있고, 저장된 상태 값을 Operator 체인에서 공유해서 사용할 수 있는 인터페이스이다.
• Context에 값을 저장하기 위해서는 contextWrite()를 사용하고, 저장된 상태 값은 key, value 형태로 저장된다.
• Context에 저장된 값을 읽어오기 위해서는 읽기 전용 뷰인 ContextView를 사용한다.
• ContextView는 Reactor Sequence에서 deferContextual() 또는 transformDeferredContextual()을 통해서 제공된다.

예제코드

public class ContextIntroduceExample01 {
    public static void main(String[] args) throws InterruptedException {
        String key = "message";
        Mono<String> mono = Mono.deferContextual(ctx ->
                        Mono.just("Hello" + " " + ctx.get(key)).doOnNext(Logger::doOnNext)
                )
                .subscribeOn(Schedulers.boundedElastic())
                .publishOn(Schedulers.parallel())
                .transformDeferredContextual((mono2, ctx) -> mono2.map(data -> data + " " + ctx.get(key)))
                .contextWrite(context -> context.put(key, "Reactor"));

        mono.subscribe(data -> Logger.onNext(data));
        Thread.sleep(100L);
    }
}

• 실행 결과를 보면 Mono.deferContextual에서 contextWrite(context -> context.put(key, "Reactor"))를 통해 저장된 값을 한번 읽어오고, transformDeferredContextual((mono2, ctx) -> mono2.map(data -> data + " " + ctx.get(key)))에서 key에 저장된 값을 한번 더 추가하기 때문에 Hello Reactor Reactor결과가 출력된 것을 확인할 수 있다.
• 두 쓰레드가 서로 다르지만, 리액터에서의 context를 이용하면 context 내부에 저장된 상태 값을 공유하기 때문에 위와 같은 결과가 나올 수 있었다.

자주 사용되는 Context API

리액터에서 사용할 수 있는 Context API는 크게 두 가지로 나뉘는데, Context에 대한 API와 ContextView에 대한 API로 분류할 수 있다.

Context에 대한 API (쓰기)

• put(key, value): key/value 형태로 Context에 값을 쓴다.
• ContextOf(key1, value1, key2, value2, ...): key/value 형태로 Context에 여러 개의 값을 쓴다.
• putAll(ContextView): 파라미터로 입력된 ContextView를 merge한다.
• delete(key): Context에서 key에 해당하는 value를 삭제한다.

ContextView API (읽기)

• get(key): ContextView에서 key에 해당하는 value를 반환한다.
• getOrEmpty(key): ContextView에서 key에 해당하는 value를 Optional로 래핑해서 반환한다.
• getOrDefault(key, default value): ContextView에서 key에 해당하는 value를 가져온다. 만약 key에 해당하는 value가 없으면 dafault value를 가져온다.
• hasKey(key): ContextView에서 특정 key가 존재하는지 확인한다.
• isEmpty(): Context가 비어있는지 확인한다.
• size(): Context내에 있는 key/value의 개수를 반환한다.

Context의 특징

리액터 시퀀스에서 사용하는 컨텍스트는 몇가지 알아두어야할 특징이 있다.

• Context는 각각의 subscriber를 통해 Reactor Sequence에 연결되며, 체인에서 각각의 Operator들이 실행 쓰레드가 달라도 연결된 Context에 접근할 수 있다.
• Context는 체인의 맨 아래에서부터 위로 전파된다.
  • Context는 Downstream에서 Upstream으로 전파된다.
  • Operator 체인에서 Context read 메서드가 Context write 메서드 밑에 있을 경우에는 write된 값을 read할 수 없다.
  • 따라서 일반적으로 Context에 write 할때에는 Operator 체인의 마지막에 둔다.
• 동일한 키에 대해서 write할 경우, 값을 덮어쓴다.
• 메인 Operator 내부에서 Sequence를 생성하는 flatMap()같은 Operator내에서 write된 Context의 값은 Inner Sequence 내부에서만 유효하고, 외부 Operator 체인에서는 보이지 않는다. (즉, Inner Sequence에서 저장된 데이터는 Inner Sequence에서만 사용할 수 있다.)

예제코드

public class ContextFetureExample01 {
    public static void main(String[] args) throws InterruptedException {
        String key1 = "name";

        Mono<String> mono = Mono.deferContextual(ctx ->
                        Mono.just("name: " + ctx.get(key1))
                )
                .publishOn(Schedulers.parallel());

        mono.contextWrite(context -> context.put(key1, "hyuuny"))
                .subscribe(data -> Logger.onNext("subscriber 1", data));

        mono.contextWrite(context -> context.put(key1, "world"))
                .subscribe(data -> Logger.onNext("subscriber 2", data));

        Thread.sleep(100L);
    }

• 실행 결과를 보면 첫 번째 구독에서는 hyuuny라는 값이 출력되었고, 두 번째 구독에서는 world라는 값이 출력되었다. 즉, 컨텍스트는 구독이 발생할 때마다 하나씩 연결이 된다는 사실을 알 수 있다.

public class ContextFetureExample02 {
    public static void main(String[] args) throws InterruptedException {
        String key1 = "name";

        Mono.deferContextual(ctx ->
                        Mono.just("name: " + ctx.get(key1))
                )
                .publishOn(Schedulers.parallel())
                .contextWrite(context -> context.put(key1, "hyuuny"))
                .contextWrite(context -> context.put(key1, "world"))
                .subscribe(Logger::onNext);

        Thread.sleep(100L);
    }
}

• Context는 Downstream -> Upstream으로 전파되기 때문에, 아래쪽에 위치한 world가 위에서 저장한 hyuuny라는 값으로 덮어씌워진 것을 확인할 수 있다.