[Reactive Programming] 마블 다이어그램(Marble Diagram)

본 포스팅은 스프링으로 시작하는 리액티브 프로그래밍 책을 참고하여 작성하였습니다. ( _ _)

---

마블 다이어그램(Marble Diagram) 이란?

마블 다이어그램은 비동기적인 데이터 흐름을 시간의 흐름에 따라 시각적으로 표시한 다이어그램을 의미한다.

제미니가 그려준 그래프

• 그림에서 두 개의 타임라인이 존재하는데, 첫 번째가 Publisher가 데이터를 emit 하는 타임라인이다
• 두 번째는 Operator 함수에서 가공 처리되어 출력으로 내보내진 데이터의 타임라인이다
• Marble로 표시된 데이터는 Publisher가 emit하는 데이터를 의미한다. 타임라인은 왼쪽에서 오른쪽으로 시간이 흐르는 것을 의미하기 때문에 가장 왼쪽에 있는 1번 구슬이 시간상으로 가장 먼저 emit된 데이터이다
• 중간 박스는 Publisher로부터 전달받은 데이터를 처리하는 Operator 함수이다.
  • Reactor는 굉장히 많은 수의 Operator를 지원하며, 각각의 Operator마다 해당 Operator를 잘 설명하는 마블 다이어그램을 가진다.
• 가장 오른쪽의 수직 바가 데이터의 emit이 정상적으로 끝났음을 의미한다. (onComplete Signal)

• X표시는 에러가 발생해 데이터 처리가 종료되었음을 의미한다. (onError Signal)

 

마블 다이어그램으로 Publisher 이해하기

지피티가 열심히 그려준 그래프

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

• Mono는 단 하나의 데이터를 emit하는 Publisher이기 때문에 그림에서도 하나의 데이터 (3)만 표현한다
• 정확히는 0개 또는 1개의 데이터를 emit하는 Publisher이다

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

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

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

• 데이터를 한 건도 emit하지 않는 경우, empty()를 사용해 onComplete Signal을 전송한다
• 실행해보면 # emitted onComplete signal이 출력되는데, 이를 통해 emit할 데이터가 없는 경우 바로 onComplete Signal을 전송한다는 걸 알 수 있다

subscrite()의 람다 표현식 파라미터

• 첫 번째 람다 표현식은 Publisher가 onNext Signal을 전송하면 실행
  즉, Subscriber가 Publisher로부터 데이터를 전달받기 위해 사용
• 두 번째 람다 표현식은 Publisher가 onError Signal을 전송하면 실행
• 세 번째 람다 표현식은 Publisher가 onComplete Signal을 전송하면 실행

 

위 그림은 Reactor의 Publisher 타입 중 하나인 Flux를 마블 다이어그램으로 표현한 것이다 

• Mono와 차이점은 Flux는 emit되는 데이터가 여러 개라는 점이다.
• Flux는 0개 또는 1개 이상의 데이터를 emit할 수 있기 때문에 Mono의 데이터 emit 범위를 포함한다.

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

• just()에서 emit하는 세 개의 숫자들을 전달받은 후에 map()에서 2로 나눈 나머지를 Subscriber에 전달

public class Example6_5 {
    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);
    }
}

• 데이터 소스로 제공되는 배열 데이터를 처리하기 위해 fromArray()를 사용한다
• 전달받은 배열의 원소를 하나씩 차례대로 emit하면 filter()에서 6보다 큰 수만 필터링 한 후, 다시 map()으로 전달한다

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

• 두 개의 Mono를 연결해 Flux로 변환
• just()의 경우 파라미터의 값으로 null을 허용하지 않지만 justOrEmpty()는 null을 허용한다
  • justOrEmpty()의 파라미터로 null이 전달되면 내부적으로 empty()를 호출하도록 구현되어 있다
• concatWith()는 concatWith()를 호출하는 Publisher와 concatWith()의 파라미터로 전달되는 Publisher가 각각 emit하는 데이터들을 하나로 연결해서 새로운 Publisher의 데이터 소스로 만들어준다
  • String의 concat처럼 하나로 이어 붙이는게 아니라, emit할 데이터들을 일렬로 줄 세워서 하나의 소스로 만든 후 차례차례 emit한다고 보면 된다

 

public class Example6_7 {
    public static void main(String[] args) {
        Flux.concat(
                        Flux.just("Mercury", "Venus", "Earth"),
                        Flux.just("Mars", "Jupiter", "Saturn"),
                        Flux.just("Uranus", "Neptune", "Pluto"))
                .collectList()
                .subscribe(planets -> System.out.println(planets));
    }
}

• concatWith()의 경우 두 개의 데이터 소스만 연결할 수 있지만, concat()은 여러 개의 데이터 소스를 원하는 만큼 연결할 수 있다
• collectList()는 Upstream Publisher에서 emit하는 데이터를 모아서 List의 원소로 포함시킨 새로운 데이터 소스로 만들어준다
• 간단한 문제 ㅎㅎ 
  • Q. concat()에서 리턴하는 Publisher는 Mono일까 Flux일까?
    • A. Flux이다. concat()을 이용해 3개의 Flux가 가지고 있는 아홉 개의 데이터를 연결하기 때문에 concat()의 리턴 값은 Flux일 수밖에 없음!
  • Q. collectList()에서 리턴하는 Publisher는 Mono일까 Flux일까?
    • A. Mono이다. collectList()는 여러 개의 데이터를 하나의 List에 원소로 포함시킨다. List에 포함된 원소는 여러 개이지만 List 자체는 하나이기 때문에 한 개의 데이터만 emit할 수 있는 Mono를 리턴한 것.
  • Q. 최종적으로 출력되는 데이터의 형태는 무엇일까?
    • A. [Mercury, Venus, Earth, Mars, Jupiter, Saturn, Uranus, Neptune, Pluto]
      정답이 List이기 때문에 각각의 데이터를 반복적으로 출력하는 것이 아니라, List 객체를 그대로 출력한다.

 

마무리하며,,,

마블 다이어그램을 가장한 Flux, Mono에 대한 포스팅이었다. (ㅋㅋ)

비동기 데이터 흐름을 파악하는게 쉽지 않기 때문에 마블 다이어그램이 참 유용한 것 같다!