[Webflux] Reactor 시작하기

공부 목적으로 작성된 글입니다 왜곡된 내용이 포함되어 있을 수 있습니다.

 

Reactor

reactor는 Spring Framework 주도하에 개발된 리액티브 스트림즈의 구현체로 Spring WebFlux 기발의 리액티브 애플리케이션을 제작하기 위한 핵심 역활을 담당한다.

https://projectreactor.io/

 

Reactor는 다음과 같은 특징을 가지고 있다.

1. Reactice Streams: 리액티브 스트림즈을 구현한 라이브러리이다.
2. Non-Blocking: 비동기 방식을 지원한다. 가장 큰 특징중 하나이다. (공부 목적이기도 하다)
3. Java's functional API: Publister와 Subscriber의 상호작을 JAVA의 함수형 프로그래밍을 통해 이루어진다.
4. Flux[$N$]: Reactor에서 지원하는 Publisher type이다.$N$개의 데이터를 emit한다는 의미이다
5. Mono[$0|1$]: Flux 와 마찬가지로 Publisher type이다. 단발성 데이터 emit에 특화되어 있다.
6. Well-suited for microservices: 마이크로 서비스에 적합하다. 비동기 통신의 특징이다.
7. Backpressure-ready network: Publisher로부터 전달받은 데이터를 처리할때 과부하가 발생하지 않도록 Backpressure을 지원한다(스케줄링과 비슷해 보인다)

Reactor를 사용하여 Hello Reactor를 출력해보자

public static void main(String[] args) {
		Flux<String> sequence=Flux.just("Hello","Reactor","Let's","study","Webflux");
		sequence.map(String::toLowerCase)
				.subscribe(System.out::println);
}

Flux가 Publisher의 역활을 하고 Publister가 최초로 제공하는데 가공되지 않는 데이터를 데이터 소스라고한다(각각의 문자열)

subscribe으로 system.out.println이 역활을 하고 있다.

또한 String::toLowerCase 처럼 메소드 레퍼런스의 형태로 구현이 되어 있음을 확인 할 수 있다.

 

이렇게 reactor에서는 데이터를 생성해서 제공하고(step 1), 데이터를 가공한 후에 (step 2) 전달받은 데이터를 처리하는(step 3) 3개의 프로세스로 진행된다.

 

마블 다이어그램

Reactor에서 지원하는 Operator을 통해 진행되는 일련의 프로세스 과정, 다양한 Operator을 이해할 수 있다.

https://gngsn.tistory.com/226

Reactor 제대로 이해하기, Marble Diagram

블로그에 정리가 너무 잘되있어서... 대체

 

Cold Sequence Hot Sequence

Cold sequence: Subcriber가 구독할 때마다 데이터 흐름이 처음부터 다시 시작되는 Sequence(별도의 타임 라인)

    @Test
    @DisplayName("Cold sequence 예제1")
    public void example1() throws InterruptedException {
        Flux<String> coldFlux=
                Flux
                        .fromIterable(Arrays.asList("KOREA","JAPAN","CHINESE"))
                        .map(String::toLowerCase);
        coldFlux.subscribe(country->log.info("# Subscriber1: {}",country));
        System.out.println("-------------------------------------------------------------------");
        Thread.sleep(2000L);
        coldFlux.subscribe(country->log.info("# Subscriber2: {}",country));
    }

 

Hot Sequence: 구독이 발생한 시점 이전에 Pushlisher로부터 emit된 데이터는 Subscriber가 전달 받지 못하고 구독이 발생한 시점 이후에 emit 데이터만 전달 받을 수 있는 sequence

    @Test
    @DisplayName("Hot sequence 예제2")
    public void example2() throws InterruptedException {
        String[] singers={"Singer A","Singer B","Singer C","Singer D","Singer E"};
        log.info("'# Begin concert: ");

        Flux<String>concertFlux=
                Flux
                        .fromArray(singers)
                        .delayElements(Duration.ofSeconds(1))
                        .share();

        concertFlux.subscribe(
                singer->log.info("# subscribe1 is wathcing {}'s song",singer)
        );

        Thread.sleep(2500L);

        concertFlux.subscribe(
                singer->log.info("# subscribe2 is wathcing {}'s song",singer)
        );
        Thread.sleep(3000L);
    }

Singer C 부터 subscribe가 구독한 것을 확인 할 수 있다.

share() Operator에 의해 Cold Sequence을 Hot Sequence로 등록한다.

 

메소드 체이닝을 통해 반환되는 publisher는 모두 다른 참조값을 가지게 되는데 share을 통해 원본 Flux(share을 호출하는 Flux)를 공유하여 동일한 참고값을 보고 있는 것이다.

 

아직 리액티브 프로그래밍에 대해서 정확한 형체는 파악이 되지 않는다..

이후에도 Hot, Cold는 계속해서 사용될 예정이니 잘 알아두자