[Spring] Project Reactor - Reactor와 마블 다이어그램

 

 

아래의 책을 정리했습니다. 

https://product.kyobobook.co.kr/detail/S000201399476

스프링으로 시작하는 리액티브 프로그래밍 | 황정식 - 교보문고

 

 

Reactor 개요

Reactor는 Spring Framework 팀 주도하에 개발된 리액티브 스트림즈의 구현체로 Spring Framework 5 버전부터 리액티브 스택에 포함되어 Spring WebFlux 기반의 리액티브 애플리케이션을 제작하기 위한 핵심 역할을 담당합니다. 

 

Reactor Core 라이브러리는 Spring WebFlux 프레임워크에 라이브러리로 포함되어 있기에 gradle dependency 설정에 spring-boot-starter-webflux를 추가하여 사용합니다. 

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
}

 

 

Reactor 특징

1. reactive streams : reactor는 리액티브 스트림즈 사양을 구현한 리액티브 라이브러리입니다. 

2. non-blocking : reactor는 JVM 위에서 실행되는 non-blocking 애플리케이션을 제작하기 위해 필요한 핵심 기술입니다. 

3. Java's functional API : Reactor에서 Publisher와 Subscriber 간의 상호작용은 java의 함수형 프로그래밍 API를 통해 이뤄집니다. 

4. Flux[n] : Reactor의 Publisher 타입은 두 가지인데, Flux는 N개의 데이터를 emit 한다는 것을 의미합니다. 

5. Mono[0|1] : Mono 역시 Reactor에서 제공하는 Publisher 타입 중 하나로, 데이터를 한 건도 emit 하지 않거나 한 건만 emit 하는 단발성 데이터에 특화되어 있습니다. 

6. Well-suited for microservices : 마이크로 서비스 기반 시스템이 non-blocking I/O에 적합한 시스템 중 하나입니다. 

7. Backpressure-ready network : Publisher로부터 전달받은 데이터를 처리하는 데 과부하가 걸리지 않도록 제어하는 backpressure을 지원하고 있습니다. 

 

 

코드로 보는 Reactor

코드를 설명해보면,

 

Flux는 Reactor에서 Publisher 역할을 합니다. 

just() 메서드의 파라미터로 전달된 데이터가 들어옵니다. > 데이터 소스

 

map() 은 Reactor에서 지원하는 Operator로 전달받은 데이터를 가공하는 역할을 합니다. 

 

subscribe() 메서드의 파라미터로 전달된 람다식이 Subscriber 역할을 하고 있습니다. 

public class Example {
  public static void main(String[] args) {
    Flux<String> sequence = Flux.just("Hello", "Reactor");
    sequence.map(data -> data.toLowerCase())
        .subscribe(data -> System.out.println(data));
  }
}

 

 

결과는 아래와 같습니다. 

 

 

마블 다이어그램

마블 다이어그램은 여러 가지 구슬 모양의 도형으로 구성된 도표로 Reactor에서 지원하는 Operator를 이해하는 데 중요한 역할을 하는 도표입니다. 

다이어그램에는 두 개의 타임라인이 존재합니다. Operator 함수를 지나기 전의 Upstream 데이터와 Downstream 데이터입니다. 

 

Publisher가 emit 한 데이터는 Operator 함수의 입력으로 전달되고 Operator 에서 데이터를 처리한 후, 내보냅니다. 정확하게 표현하면 Operator 함수에서 리턴하는 새로운 Publisher를 이용해 Downstream에 가공된 데이터를 전달합니다. 

 

정상적으로 데이터가 처리되지 않아 에러가 발생한 경우 X 표시를 하며 onError Signal을 보냅니다. 

 

 

마블 다이어그램 - Mono

 

Mono는 하나의 데이터를 emit 하는 Publisher이기 떄문에 단 하나의 데이터만 표현합니다. 

엄밀히 말하면, Mono는 하나의 데이터가 아닌 0~1 개의 데이터를 emit 하므로 1이라 적힌 구슬 데이터가 emit 되지 않고 onComplete signal만 전송될 수 있습니다. 

Hello Reactor를 출력하는 예제입니다. 

just() operator는 한 개 이상의 데이터를 emit 하기 위한 대표적인 operator로, 2개 이상의 데이터를 파라미터로 전달할 경우, 내부적으로 fromArray() operator를 이용해서 데이터를 emit 합니다. 

public class Example {
  public static void main(String[] args) {
    Mono.just("Hello Reactor")
        .subscribe(System.out::println);
  }
}

 

데이터를 0건 emit 하는 예제입니다. 

empty() operator를 사용하면 데이터를 emit 하지 않고 onComplete signal을 전송합니다. 

empty()는 주로 어떤 특정 작업을 통해 데이터를 전달받을 필요는 없지만 작업이 끝났음을 알리고 이에 따른 후처리를 하고 싶을 때 사용합니다. 

public class Example {
  public static void main(String[] args) {
    Mono
        .empty()
        .subscribe(
            none -> System.out.println("# emitted onNext Signal"),
            error -> {},
            () -> System.out.println("# emitted onComplete Signal")
        );
  }
}

	한 건만 emit	0건 또는 1건 emit
RxJava	Single	Maybe
Reactor	Mono

 

 

 

마블 다이어그램 - Flux

Flux는 여러 건의 데이터를 emit 할 수 있는 Publisher 입니다. 

정확히는 Flux는 0개 또는 1개 이상의 데이터를 emit 할 수 있기에 Mono의 데이터 emit 범위를 포함하고 있습니다. 

just() 에서 emit 하는 세 개의 숫자를 전달받은 후에 map() 에서 2로 나눈 나머지를 subscriber에게 전달하여 출력하는 예제입니다. 

public class Example {
  public static void main(String[] args) {
    Flux.just(6, 9, 13)
        .map(num -> num % 2)
        .subscribe(System.out::println);
  }
}

 

이 코드에서는 데이터 소스로 제공되는 배열 데이터를 처리하기 위해 fromArray()를 사용합니다. 

이렇게 전달받은 배열의 원소를 하나씩 emit 하면 filter에서 6보다 큰 숫자만 필터링하여 map()으로 전달합니다. map은 전달받은 숫자에 2를 곱하여 subscriber에게 전달하고 출력하는 예제입니다. 

public class Example {
  public static void main(String[] args) {
    Flux.fromArray(new Integer[]{3, 6, 7, 9})
        .filter(num -> num > 6)
        .map(num -> num * 2)
        .subscribe(System.out::println);
  }
}

 

justOrEmpty() 는 파라미터 값으로 null을 허용합니다. null이 전달되면 내부적으로 empty()를 호출하게 됩니다. 그리고 concatWith()은 각각 emit하는 데이터를 하나로 연결하여 새로운 Publisher의 데이터 소스로 만들어 주는 operator입니다. 

public class Example {
  public static void main(String[] args) {
    Flux<String> flux =
        Mono.justOrEmpty("Steve")
            .concatWith(Mono.justOrEmpty("jobs"));
    flux.subscribe(System.out::println);
  }
}