코드스테이츠 백엔드 부트캠프 70일차 - [Spring WebFlux] Project Reactor

wish17·2023년 3월 26일

Project Reactor Spring section4 spring webflux 코드스테이츠

코드스테이츠 백엔드 부트캠프

목록 보기

58/76

Reactor 공부 개꿀팁...(인텔리제이 사용)

메소드 위에 마우스 올려서 마블다이어그램 확인할 때 창 분리해서 볼 수 있다.

Project Reactor

Reactor

리액티브 스트림즈(Reactive Streams)를 구현한 구현체 중 하나

리액티브한 애플리케이션으로 동작하는데 있어 핵심적인 역할을 담당하는 리액티브 프로그래밍을 위한 라이브러리

Reactor 특징

Non-Blocking

요청 쓰레드가 차단이 되지 않는다.
Non-Blocking 통신을 완벽하게 지원하는 Reactor는 MSA 구조에 적합한 라이브러리다.

Publisher 타입

Mono[0|1]
- 0과 1의 의미는 0건 또는 1건의 데이터를 emit 할 수 있음을 의미
Flux[N]
- N의 의미는 여러 건의 데이터를 emit할 수 있음을 의미

Backpressure

Subscriber의 처리 속도가 Publihser의 emit 속도를 따라가지 못할 때 적절하게 제어하는 전략을 의미
- Publihser가 이미 받은 데이터들을 아직 처리 못하고 있는데 Subscriber가 계속 데이터를 보내주면 처리 못한 데이터가 쌓이다가 오버플로우가 발생할 수 있다.

Reactor 구성 요소

public class HelloReactorExample {
    public static void main(String[] args) throws InterruptedException {
        Flux    //  Reactor Sequence의 시작점, Reactor Sequence가 여러 건의 데이터를 처리함을 의미
                .just("Hello", "Reactor") // 원본 데이터 소스로부터(Original Data Source) 데이터를 emit하는 Publisher의 역할
                .map(message -> message.toUpperCase()) // 발행자로부터 받은 데이터를 처리
                // 여기서는 just() 연산자가 방출하는 영어 문자열을 대문자로 변환한다.
                .publishOn(Schedulers.parallel()) // Reactor Sequence에서 쓰레드 관리자 역할을 하는 Scheduler를 지정
                // publishOn()에서 스케줄러를 지정하면, 다운스트림 스레드가 publishOn() 기준으로 Downstream의 쓰레드가 Scheduler에서 지정한 유형의 쓰레드로 변경된다.
                // 즉, 이 코드에서는 리액터 시퀀스상에서 두 개의 스레드가 실행된다.
                .subscribe(System.out::println, // subscribe()는 총 세 개의 람다 표현식을 매개변수로 가지며, 첫 번째 매개변수는 Publisher가 emit한 데이터를 전달 받아서 처리하는 역할을 한다.
                        error -> System.out.println(error.getMessage()), // 두 번째 매개변수는 Reqctor 시퀀스에서 오류가 발생했을 때 오류를 수신하고 처리하는 역할을 한다.
                        () -> System.out.println("# onComplete")); // 세 번째 매개변수는 리액터 시퀀스가 종료된 후 일부 후처리를 수행하는 역할을 한다.

        Thread.sleep(100L); // Reactor에서 Scheduler로 지정한 쓰레드는 모두 데몬 쓰레드이기 때문에 주 쓰레드인 main 쓰레드가 종료되면 동시에 종료된다.
        // 따라서 데몬 쓰레드 작업이 다 끝나고 main 쓰레드가 끝나도록 0.1초 동작 지연을 넣어준 것이다.
    }
}

Reactor에서 Scheduler로 지정한 쓰레드는 모두 데몬 쓰레드이기 때문에 주 쓰레드인 main 쓰레드가 종료되면 동시에 종료된다.

마블 다이어그램(Marble Diagram)

구슬 모양의 동그라미는 하나의 데이터를 의미하며, 다이어그램 상에서 시간의 흐름에 따라 변화하는 데이터의 흐름을 표현한다.

복잡한 데이터의 흐름을 마블 다이어그램을 통해 좀 더 쉽게 이해할 수 있다.

Mono의 마블 다이어그램

Reactor의 데이터 타입 중 하나인 Mono를 마블 다이어그램으로 표현한 것이다.

Mono는 0건 또는 1건의 데이터만 emit하는 Reactor 타입이다.
| 는 정상 종료, X는 에러로 인한 비정상 종료를 의미한다.

Flux의 마블 다이어그램

Flux는 여러 개(0 ~ N)의 데이터를 emit하는 Reactor 타입이다.

Operator의 마블 다이어그램 예시

map() Operator는 입력으로 들어오는 데이터를 개발자가 구현하는 동작대로 변환해서 Downstream으로 전달하는 역할을 한다.

import reactor.core.publisher.Flux;

public class MarbleDiagramExample {
    public static void main(String[] args) {
        Flux
            .just("Green-Circle", "Orange-Circle", "Blue-Circle")   // 세 개의 문자열을 emit
            .map(figure -> figure.replace("Circle", "Diamond"))   // Circle -> Diamond
            .subscribe(System.out::println);   // map() Operator 내부에서 변환된 문자열을 출력
    }
}

//출력
Green-Diamond
Orange-Diamond
Blue-Diamond

스케줄러(Scheduler)

쓰레드를 관리하는 관리자의 역할을 한다.
- Reactor Sequence 상에서 처리되는 동작들을 하나 이상의 쓰레드에서 동작하도록 별도의 쓰레드를 제공해 준다.

Reactor의 Scheduler는 복잡한 멀티쓰레딩 프로세스를 단순하게 해준다.

스케줄러 주요 기능

작업 분배

리액터 시퀀스에서 처리되는 작업들을 스레드 풀 내의 여러 스레드에 분배한다.

백프레셔 처리

발행자와 구독자 사이의 속도 차이를 조율하고, 이로 인한 과부하를 방지한다.

작업 실행 시점 조절

작업들을 지연시키거나, 반복 실행 등의 스케줄링 전략을 제공한다.

Scheduler 전용 Operator

subscribeOn()

구독 직 후 실행되는 Operator 체인의 실행 쓰레드를 Scheduler에서 지정한 쓰레드로 변경한다.
- doOnSubscribe() Operator는 구독 발생 직 후에 트리거 되는 Operator로써 구독 직 후에 실행되는 쓰레드와 동일한 쓰레드에서 실행된다. 만약 구독 직 후에 어떤 동작을 수행하고 싶다면 doOnSubscribe()에 로직을 작성하면 된다.

@Slf4j
public class SchedulersExample02 {
    public static void main(String[] args) throws InterruptedException {
        Flux
                .range(1, 10)
---------------------------------------------------------------------------------------------------------------------------------------------------
                .doOnSubscribe(subscription -> log.info("# doOnSubscribe")) // 구독 직후에 어떤 동작을 수행하고 싶다면 doOnSubscribe()에 로직을 작성
                .subscribeOn(Schedulers.boundedElastic())     // Operator 체인의 실행 쓰레드를 Scheduler에서 지정한 쓰레드(boundedElastic)로 변경
---------------------------------------------------------------------------------------------------------------------------------------------------                
                .filter(n -> n % 2 == 0)
                .map(n -> n * 2)
                .subscribe(data -> log.info("# onNext: {}", data));

        Thread.sleep(100L);
    }
}

//실행결과
14:07:17.781 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
14:07:17.821 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # doOnSubscribe
14:07:17.822 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # onNext: 4
14:07:17.825 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # onNext: 8
14:07:17.825 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # onNext: 12
14:07:17.825 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # onNext: 16
14:07:17.825 [boundedElastic-1] INFO com.codestates.example.schedulers.SchedulersExample02 - # onNext: 20

Reactor Sequence에서 발생하는 signal의 전파 흐름

publishOn()

publishOn()을 추가할 때 마다 추가한 publishOn()을 기준으로 Downstream쪽 쓰레드가 publishOn()에서 Scheduler로 지정한 쓰레드로 변경된다.
전달 받은 데이터를 가공 처리하는 Operator 앞에 추가해서 실행 쓰레드를 별도로 추가한다.
- publishOn()은 Operator 앞에 여러번 추가할 경우 별도의 쓰레드가 추가로 생성되지만 subscribeOn()은 여러 번 추가해도 하나의 쓰레드만 추가로 생성한다.

@Slf4j
public class SchedulersExample03 {
    public static void main(String[] args) throws InterruptedException {
        Flux
                .range(1, 10)
                .subscribeOn(Schedulers.boundedElastic())
                .doOnSubscribe(subscription -> log.info("# doOnSubscribe"))

                .publishOn(Schedulers.parallel())  // 필터 기능 쓰레드 변경
                .filter(n -> n % 2 == 0)
                .doOnNext(data -> log.info("# filter doOnNext")) // doOnNext()는 doOnNext() 바로 앞에 위치한 Operator가 실행될 때, 트리거 되는 Operator

                .publishOn(Schedulers.parallel())    // map Operator 쓰레드 변경
                .map(n -> n * 2)
                .doOnNext(data -> log.info("# map doOnNext"))

                .subscribe(data -> log.info("# onNext: {}", data));

        Thread.sleep(100L);
    }
}

//출력
14:30:44.118 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
14:30:44.159 [main] INFO com.codestates.example.schedulers.SchedulersExample03 - # doOnSubscribe
14:30:44.170 [parallel-2] INFO com.codestates.example.schedulers.SchedulersExample03 - # filter doOnNext
14:30:44.170 [parallel-2] INFO com.codestates.example.schedulers.SchedulersExample03 - # filter doOnNext
14:30:44.170 [parallel-2] INFO com.codestates.example.schedulers.SchedulersExample03 - # filter doOnNext
14:30:44.170 [parallel-2] INFO com.codestates.example.schedulers.SchedulersExample03 - # filter doOnNext
14:30:44.170 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # map doOnNext
14:30:44.170 [parallel-2] INFO com.codestates.example.schedulers.SchedulersExample03 - # filter doOnNext
14:30:44.170 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # onNext: 4
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # map doOnNext
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # onNext: 8
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # map doOnNext
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # onNext: 12
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # map doOnNext
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # onNext: 16
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # map doOnNext
14:30:44.173 [parallel-1] INFO com.codestates.example.schedulers.SchedulersExample03 - # onNext: 20

subscribeOn()예시 코드와 다르게 # doOnSubscribe 출력문이 main쓰레드로 되어 있는 것을 확인할 수 있었다. 왜 그런것일까?

A: subscribeOn() 오퍼레이터와 doOnSubscribe() 오퍼레이터의 순서가 영향이 있다. 아래 그림과 같이 doOnSubscribe() 오퍼레이터는 다운스트림에서 업스트림으로 가는 사이에 동작한다.

이와 같이 다양한 오퍼레이트의 마블 다이어그램을 참고하면 동작 과정을 이해하는데 도움이 될 것 같다.
(인텔리제이에서 오퍼레이트 위에 마우스 올리면 확인 가능)

핵심 포인트

Reactor에서의 Scheduler는 쓰레드를 관리하는 관리자 역할을 한다.
Scheduler는 한마디로 Reactor Sequence 상에서 처리되는 동작들을 하나 이상의 쓰레드에서 동작하도록 별도의 쓰레드를 제공해 준다.
subscribeOn()은 주로 데이터 소스에서 데이터를 emit하는 원본 Publisher의 실행 쓰레드를 지정하는 역할을 한다.
publishOn()은 전달 받은 데이터를 가공 처리하는 Operator 앞에 추가해서 실행 쓰레드를 별도로 추가하는 역할을 한다.

Operators

상황별로 분류된 Operator 목록

새로운 Sequence를 생성(Creating)하고자 할 경우

just(): 단일 항목을 가진 시퀀스를 생성한다.

fromStream(⭐): 스트림을 사용하여 시퀀스를 생성한다.

fromIterable(⭐): 반복 가능한 객체(예: 리스트, 세트)를 사용하여 시퀀스를 생성한다.

fromArray(): 배열을 사용하여 시퀀스를 생성한다.

range(): 특정 범위의 숫자들로 시퀀스를 생성한다.

interval(): 일정 시간 간격으로 생성되는 시퀀스를 만든다.

empty(): 빈 시퀀스를 생성한다.

never(): 완료되거나 값을 방출하지 않는 시퀀스를 생성한다.

defer(): 구독 시점에 시퀀스를 생성하는 연산자를 반환한다.

using(): 자원을 사용하여 시퀀스를 생성하고, 구독이 종료되면 자원을 해제한다.

generate(): 초기 상태, 상태 변환 함수, 종료 조건 등을 사용하여 시퀀스를 생성한다.

create(⭐): 사용자 정의 방식으로 시퀀스를 생성한다. 이것도 중요한 Operator임을 나타낸다.

기존 Sequence에서 변환 작업(Transforming)이 필요한 경우

map(⭐): 시퀀스의 각 항목에 대해 주어진 함수를 적용한 결과를 방출하는 새로운 시퀀스를 생성한다.

flatMap(⭐): 시퀀스의 각 항목에 대해 주어진 함수를 적용한 결과로 생성되는 시퀀스들을 병합하여 새로운 시퀀스를 생성한다.

concat(⭐): 여러 시퀀스를 연결하여 하나의 시퀀스로 만든다.

collectList(): 시퀀스의 모든 항목을 리스트로 수집한다.

collectMap(): 시퀀스의 각 항목을 키-값 쌍으로 변환하여 맵으로 수집한다.

merge(): 여러 시퀀스를 병합하여 하나의 시퀀스로 만든다.

zip(⭐): 두 개 이상의 시퀀스에서 동시에 항목을 가져와 주어진 함수를 적용한 결과를 방출하는 새로운 시퀀스를 생성한다.

then(): 주어진 시퀀스가 완료되면 다른 시퀀스를 실행한다.

switchIfEmpty(): 시퀀스가 비어 있을 경우 대체할 시퀀스를 제공한다.

and(): 두 시퀀스를 동시에 실행하고 결과를 병합한다.

when(): 주어진 조건이 충족되면 시퀀스를 실행한다.

Sequence 내부의 동작을 확인(Peeking)하고자 할 경우

doOnSubscribe: 구독 시점에 호출할 액션을 정의한다.

doOnNext(⭐): 각 항목이 방출될 때 호출할 액션을 정의한다.

doOnError(): 에러가 발생했을 때 호출할 액션을 정의한다.

doOnCancel(): 구독이 취소되었을 때 호출할 액션을 정의한다.

doFirst(): 시퀀스가 처음 시작될 때 호출할 액션을 정의한다.

doOnRequest(): 요청이 발생했을 때 호출할 액션을 정의한다.

doOnTerminate(): 시퀀스가 종료되었을 때 호출할 액션을 정의한다.

doAfterTerminate(): 시퀀스가 종료된 후 호출할 액션을 정의한다.

doOnEach(): 각 이벤트(방출, 완료, 에러)에 대해 호출할 액션을 정의한다.

doFinally(): 시퀀스의 종료 후 호출할 액션을 정의한다.

log(⭐): 시퀀스의 동작을 로깅한다.

Sequence에서 데이터 필터링(Filtering)이 필요한 경우

filter(⭐): 주어진 조건에 맞는 항목만 통과시키는 새로운 시퀀스를 생성한다.

ignoreElements(): 시퀀스의 모든 항목을 무시하고 완료 이벤트만 전달한다.

distinct(): 중복되지 않은 항목만 통과시키는 새로운 시퀀스를 생성한다.

take(⭐): 시퀀스에서 처음부터 지정한 개수의 항목만 가져오는 새로운 시퀀스를 생성한다.

next(): 시퀀스의 첫 번째 항목을 반환한다.

skip(): 시퀀스에서 처음부터 지정한 개수의 항목을 건너뛰고 나머지 항목을 방출하는 새로운 시퀀스를 생성한다.

sample(): 일정 시간 간격으로 시퀀스의 항목을 샘플링하는 새로운 시퀀스를 생성한다.

single(): 시퀀스에서 단 하나의 항목만 통과시키는 새로운 시퀀스를 생성한다.

에러를 처리(Handling errors)하고자 할 경우

error(⭐): 에러를 처리하는 함수를 정의한다. 이 함수는 에러가 발생할 때 호출되며, 적절한 처리를 수행한다.

timeout(⭐): 시퀀스가 지정한 시간 내에 값을 방출하지 않으면 에러를 발생시킨다.

onErrorReturn(): 에러가 발생하면 지정한 값을 방출하는 대신에 반환하는 새로운 시퀀스를 생성한다.

onErrorResume(): 에러가 발생하면 대체할 시퀀스를 실행한다.

onErrorMap(): 에러가 발생하면 에러를 변환하는 함수를 적용한다.

doFinally(): 시퀀스의 종료 후 호출할 액션을 정의한다.

retry(⭐): 에러가 발생할 때 지정한 횟수만큼 시퀀스를 재시도한다.

중요 Operator

새로운 Sequence를 생성(Creating)하고자 할 경우

fromStream()

Java의 Stream을 입력으로 전달 받아 emit하는 Operator
- 전달 받은 Stream이 포함하고 있는 데이터를 차례대로 emit한다.

fromIterable()

Java의Iterable을 입력으로 전달 받아 emit하는 Operator
- List, Map, Set 등의 컬렉션을 fromIterable()의 파라미터로 전달할 수 있다.

create()

사용자가 정의한(프로그래밍) 방식으로 Signal 이벤트를 발생시키는 Operator
- FluxSink = 데이터 emit 등의 기능을 개발자가 직접 코드를 구현해서 처리함을 의미
한 번에 여러 건의 데이터를 비동기적으로 emit할 수 있다.(= multhreaded source)

@Slf4j
public class CreateExample {
    private static List<Integer> source= Arrays.asList(1, 3, 5, 7, 9, 11, 13, 15, 17, 19);

    public static void main(String[] args) {
        Flux.create((FluxSink<Integer> sink) -> {   // FluxSink = 데이터 emit 등의 기능을 개발자가 직접 코드를 구현해서 처리함을 의미
            sink.onRequest(n -> { // Subscriber에서 데이터를 요청하면 onRequest()의 파라미터인 람다 표현식이 실행된다.
                for (int i = 0; i < source.size(); i++) {
                    sink.next(source.get(i));   // next() 메서드로 List source의 원소를 emit
                }
                sink.complete();    // Sequence를 종료
            });

            sink.onDispose(() -> log.info("# clean up")); // onDispose() = Sequence가 완전히 종료되기 직전에 호출되며, sequence 종료 직전 후처리 작업을 할 수 있다.
        }).subscribe(data -> log.info("# onNext: {}", data));
    }
}

기존 Sequence에서 변환 작업(Transforming)이 필요한 경우

flatMap()

내부로 들어오는 데이터 한 건당 하나의 Sequence가 생성
- ex. Upstream에서 3개의 데이터를 emit하고, flatMap() 내부에서 2개의 데이터를 emit하는 Sequence가 있다면 Downstream으로 emit되는 데이터는 총 6개(3 x 2)가 된다.
flatMap() 내부에서 정의하는 Sequence를 Inner Sequece라고 부른다.

@Slf4j
public class FlatMapExample01 {
    public static void main(String[] args) throws InterruptedException {
        Flux
                .range(2, 6)         // 2부터 6개 호출 (즉, 2~7 호출)
                .flatMap(dan -> Flux
                        .range(1, 9)  // 1부터 9개 호출 (1~9 호출)
                        .publishOn(Schedulers.parallel())   // flatMap() 내부의 Inner Sequence를 처리할 쓰레드를 할당 // 여러 개의 쓰레드가 비동기적으로 동작하게 만드는 것
                        .map(num -> dan + " x " + num + " = " + dan * num)) // 구구단 형식으로 문자열 구성
                .subscribe(log::info);

        Thread.sleep(100L);
    }
}

//출력
23:58:05.791 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
23:58:05.906 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 1 = 5
23:58:05.907 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 2 = 10
23:58:05.908 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 3 = 15
23:58:05.909 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 4 = 20
23:58:05.911 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 5 = 25
23:58:05.911 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 6 = 30
23:58:05.911 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 7 = 35
23:58:05.911 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 8 = 40
23:58:05.911 [parallel-4] INFO com.codestates.example.operators.FlatMapExample01 - 5 x 9 = 45
23:58:05.918 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 1 = 3
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 1 = 4
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 2 = 8
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 3 = 12
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 4 = 16
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 5 = 20
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 6 = 24
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 7 = 28
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 8 = 32
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 4 x 9 = 36
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 1 = 6
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 2 = 12
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 3 = 18
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 4 = 24
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 5 = 30
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 6 = 36
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 7 = 42
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 8 = 48
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 6 x 9 = 54
23:58:05.919 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 1 = 7
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 2 = 14
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 3 = 21
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 4 = 28
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 5 = 35
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 6 = 42
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 7 = 49
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 8 = 56
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 7 x 9 = 63
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 1 = 2
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 2 = 4
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 3 = 6
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 4 = 8
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 5 = 10
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 6 = 12
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 7 = 14
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 8 = 16
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 2 x 9 = 18
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 2 = 6
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 3 = 9
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 4 = 12
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 5 = 15
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 6 = 18
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 7 = 21
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 8 = 24
23:58:05.920 [parallel-2] INFO com.codestates.example.operators.FlatMapExample01 - 3 x 9 = 27

flatMap() Operator에서 추가 쓰레드를 할당할 경우, 작업의 처리 순서를 보장하지 않는다

concat()

입력으로 전달하는 Publisher의 Sequence를 연결해서 차례대로 데이터를 emit하는 Operator다.
- 하나의 Sequence로 이어 붙인다

zip()

입력으로 전달되는 여러 개의 Publisher Sequence에서 emit된 데이터를 결합하는 Operator다.
- 결합 = 각 Publisher가 emit하는 데이터를 하나씩 전달 받아서 새로운 데이터를 만든 후에 Downstream으로 전달한다는 의미

각각의 Sequence에서 emit되는 데이터 중에서 같은 차례(index)의 데이터들이 결합된다.

각 Sequence에서 emit되는 데이터의 시점이 다르기 때문에 결합되어야 하는 데이터(같은 index)가 emit이 될 때까지 기다렸다가 결합한다.

@Slf4j
public class ZipExample01 {
    public static void main(String[] args) throws InterruptedException {
        Flux<Long> source1 = Flux.interval(Duration.ofMillis(200L)).take(4); // 0.2초마다 한번씩 데이터를 emit (0,1,2,3)

        Flux<Long> source2 = Flux.interval(Duration.ofMillis(400L)).take(6); // 0.4초마다 한번씩 데이터를 emit (0,1,2,3,4,5)

        Flux
                .zip(source1, source2, (data1, data2) -> data1 + data2)   // 결합시키는데 source2의 남는 데이터(4,5)는 폐기 됨
                .subscribe(data -> log.info("# onNext: {}", data));

        Thread.sleep(3000L);
    }
}

Sequence 내부의 동작을 확인(Peeking)하고자 할 경우

doOnNext()

데이터 emit 시 트리거 되어 부수 효과(side-effect)를 추가할 수 있는 Operator
리턴값이 없음
주로 로깅(로그를 기록 또는 출력하는 작업)에 사용

부수 효과(side-effect)

어떤 동작을 실행하되 리턴 값이 없는 것을 의미

실제 emit된 데이터를 가지고 무언가 처리 작업을 하지만 Downstream으로 전달되는 것은 emit된 데이터이지 doOnNext() Operator에서 처리된 작업의 결과 값은 아니다.

@Slf4j
public class DoOnNextExample { // emit되는 데이터의 유효성 검증
    public static void main(String[] args) {
        Flux
                .fromIterable(SampleData.coffeeList)
                .doOnNext(coffee -> validateCoffee(coffee))
                .subscribe(data -> log.info("{} : {}", data.getKorname(), data.getPrice()));
    }

    private static void validateCoffee(Coffee coffee) {
        if (coffee == null) {
            throw new RuntimeException("Not found coffee");
        }
        // 유효성 검증에 필요한 로직을 필요한 만큼 추가할 수 있다.
    }
}

log()

Publisher에서 발생하는 Signal 이벤트를 로그로 출력해주는 역할

public class LogExample {
    public static void main(String[] args) {
        Flux
                .fromStream(Stream.of(200, 300, 400, 500, 600))
                .log()
                .reduce((a, b) -> a + b)
                .log()
                .subscribe(System.out::println);
    }
}

//출력
> Task :LogExample.main()
00:46:57.906 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
// onSubscribe Signal 이벤트 두번 발생
00:46:57.971 [main] INFO reactor.Flux.Stream.1 - | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)
00:46:57.973 [main] INFO reactor.Mono.Reduce.2 - | onSubscribe([Fuseable] MonoReduce.ReduceSubscriber)
// request Signal 이벤트 두번 발생
00:46:57.974 [main] INFO reactor.Mono.Reduce.2 - | request(unbounded)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | request(unbounded)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | onNext(200)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | onNext(300)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | onNext(400)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | onNext(500)
00:46:57.974 [main] INFO reactor.Flux.Stream.1 - | onNext(600)
00:46:57.975 [main] INFO reactor.Flux.Stream.1 - | onComplete()
00:46:57.975 [main] INFO reactor.Mono.Reduce.2 - | onNext(2000)
2000
00:46:57.975 [main] INFO reactor.Mono.Reduce.2 - | onComplete()

에러를 처리(Handling errors)하고자 할 경우

error()

onError Signal 이벤트를 발생 시키는 Operator

@Slf4j
public class ErrorExample {
    public static void main(String[] args) {
        Mono.justOrEmpty(findVerifiedCoffee())  // justOrEmpty() Operator는 파라미터로 전달되는 데이터소스가 null 이어도 에러가 발생하지 않는다.
                .switchIfEmpty(Mono.error(new RuntimeException("Not found coffee")))  // switchIfEmpty() => 만약 null이면 괄호안 수행
                .subscribe(
                        data -> log.info("{} : {}", data.getKorname(), data.getPrice()), // 만약 정상적인 coffee 객체를 받으면, 받은 데이터 출력
                        error -> log.error("# onError: {}", error.getMessage()));  // error 객체를 전달 받아서 에러 메시지를 출력
    }

    private static Coffee findVerifiedCoffee() {
        // 데이터베이스에서 Coffee 정보를 조회할 때 null값이 리턴됐다고 가정
        return null;
    }
}

//출력
> Task :ErrorExample.main()
01:14:54.256 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
01:14:54.269 [main] ERROR com.codestates.example.operators.ErrorExample - # onError: Not found coffee

timeout()

입력으로 주어진 시간동안 emit되는 데이터가 없으면 onError Signal 이벤트를 발생
timeout() Operator는 retry() Operator와 함께 사용하는 경우가 많다.

retry()

Sequence 상에서 에러가 발생할 경우, 입력으로 주어진 숫자만큼 재구독해서 Sequence를 다시 시작

@Slf4j
public class TimeoutRetryExample { // 일정 시간 내에 데이터가 emit 되지 않으면 다시 시도하는 예제
    public static void main(String[] args) throws InterruptedException {
        getCoffees()
                .collect(Collectors.toSet())   // emit된 데이터를 Set<Coffee>으로 변환 // timeout이 되기전에 이미 emit된 데이터가 있으므로 재구독 후, 다시 emit된 데이터에 동일한 데이터가 있으므로 중복을 제거하기 위함
                .subscribe(bookSet -> bookSet
                        .stream()
                        .forEach(data ->
                                log.info("{} : {}", data.getKorname(), data.getPrice())));

        Thread.sleep(12000);
    }

    private static Flux<Coffee> getCoffees() {
        final int[] count = {0};
        return Flux
                .fromIterable(SampleData.coffeeList)
                .delayElements(Duration.ofMillis(500)) // 입력으로 주어진 시간만큼 각각의 데이터 emit을 지연시키는 Operator (0.5초 지연)
                .map(coffee -> {
                    try {
                        count[0]++;
                        if (count[0] == 4) { // 4번째 emit일 때
                            Thread.sleep(2000); // 2초 지연
                        }
                    } catch (InterruptedException e) {
                    }

                    return coffee;
                })
                .timeout(Duration.ofSeconds(2))   // 2초안에 데이터가 emit되지 않으면 onError Signal 이벤트 발생
                .retry(1)     // 에러나면 1회 재구독을 해서 Sequence를 다시 시작 (처음부터 한번 더 다시하라는 명령)
                .doOnNext(coffee -> log.info("# getCoffees > doOnNext: {}, {}",
                        coffee.getKorname(), coffee.getPrice()));
    }
}

//출력
> Task :TimeoutRetryExample.main()
03:11:54.666 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
03:11:55.294 [parallel-2] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 아메리카노, 2500
03:11:55.802 [parallel-4] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 카페라떼, 3500
03:11:56.313 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 바닐라 라떼, 4500
03:11:58.826 [parallel-8] DEBUG reactor.core.publisher.Operators - onNextDropped: com.codestates.example.operators.Coffee@4d42eba7
03:11:58.841 [parallel-10] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 아메리카노, 2500
03:11:59.352 [parallel-12] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 카페라떼, 3500
03:11:59.864 [parallel-2] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 바닐라 라떼, 4500
03:12:00.377 [parallel-4] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 카라멜 마끼아또, 5500
03:12:00.890 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - # getCoffees > doOnNext: 에스프레소, 5000
03:12:00.891 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - 카라멜 마끼아또 : 5500
03:12:00.891 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - 아메리카노 : 2500
03:12:00.891 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - 바닐라 라떼 : 4500
03:12:00.891 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - 카페라떼 : 3500
03:12:00.891 [parallel-6] INFO com.codestates.example.operators.TimeoutRetryExample - 에스프레소 : 5000

핵심 포인트

Reactor의 Operator는 그 종류가 너무 많기 때문에 한번에 모든 Operator의 사용법을 익히는 것은 사실상 불가능하다. 상황별로 자주 사용되는 Operator의 사용법을 익히고, 그때 그때 필요한 상황에 해당 Operator를 찾아서 사용하는 것이 좋다.
flatMap() 처럼 내부에서 정의하는 Sequence를 Inner Sequece라고 한다.
flatMap() Operator에 추가 쓰레드를 할당할 경우, 작업의 처리 순서를 보장하지 않는다.
create() Operator는 프로그래밍 방식으로 Signal 이벤트를 발생시키는 Operator로써 한 번에 여러 건의 데이터를 비동기적으로 emit할 수 있다.
Reactor의 concat() Operator는 논리적으로 하나의 Sequence로 이어 붙인 후, 이어 붙인 Sequecne에서 시간 순서대로 데이터를 차례대로 emit한다.
zip() Operator는 입력으로 주어진 Sequence의 emit 시점이 매번 다르더라도 동일 index 상의 emit 시점이 늦은 데이터가 emit될 때까지 대기했다가 데이터가 모두 emit되면 해당 데이터를 모두 전달 받는다.
함수형 프로그래밍 세계에서 부수 효과(side-effect)는 어떤 동작을 실행하되 리턴 값이 없는 것을 의미한다.
timeout() Operator는 입력으로 주어진 시간동안 emit되는 데이터가 없으면 onError Signal 이벤트를 발생시킨다.
retry() Operator는 Sequence 상에서 에러가 발생할 경우, 입력으로 주어진 숫자만큼 재구독해서 Sequence를 다시 시작한다.

wish17

이전 포스트

코드스테이츠 백엔드 부트캠프 69일차 - [Spring WebFlux] 리액티브 프로그래밍

다음 포스트

코드스테이츠 백엔드 부트캠프 70일차 - [Spring WebFlux] Project Reactor