[📕 Reactive] 5. Reactor

Reactor란 ? 

리액티브 프로그래밍을 위한 라이브러리 

 

Spring Framework 팀에 의해 개발된 리액티브 스트림즈의 구현체로서 Spring WebFlux 기반의 리액티브 어플리케이션을 제작하기 위한 핵심 역할을 담당한다.

---

✳️ Reactor의 특징 

• Reactive Streams
• Non-Blocking
• Java's Functional API
• Flux[N]
• Mono[0|1]
• Microservices
• Backpressure

 

 

Mono 기본 예제

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")
        	);
    }
}

 

• empty() Operator를 사용하면 내부적으로 emit할 데이터가 없는 것으로 간주하여 곧바로 onComplete signal 전송
• empty() Operator는 어떤 특정 작업을 통해 데이터를 전달받을 필요는 없지만 작업이 끝났음을 알리고 이에 따른 후처리를 하고 싶을 때 사용
  
  
• subscribe() 메서드
  • 첫 번째 람다 표현식은 Publisher가 onNext signal을 전송하면 실행됨 👉 Subscriber가 Publisher로부터 데이터를 전달받기 위해 사용됨
  • 두 번째 람다 표현식은 Publisher가 onError signal을 전송하면 실행됨 
  • 세 번째 람다 표현식은 Publisher가 onComplete signal을 전송하면 실행됨 

 

 

Flux 기본 예제

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(Sytstem.out::println);
    }
}

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

• justOrEmpty()의 파라미터로 null이 전달되면 내부적으로 empty() Operator 호출
• 두 개의 Mono에서 emit하는 데이터를 하나의 데이터 소스로 연결하여 새로운 Flux 리턴
• 문자열처럼 데이터 자체를 이어 붙여서 하나의 데이터를 emit 하는 게 아니라 emit할 데이터를 일렬로 줄 세워서 하나의 데이터 소스를 만든 후에 차례차례 데이터 emit 
• concatWith() 는 두 개의 데이터 소스만 연결 가능 
• concat()은 여러 개의 데이터 소스를 원하는 만큼 연결 가능

 

 

 

Cold / Hot Sequence

 

✳️ Cold Sequence

• Subscriber가 구독할 때마다 데이터 흐름이 처음부터 다시 시작되는 sequence

 

✳️ Hot Sequence

• 구독이 발생한 시점 이전에 Publisher로부터 emit된 데이터는 Subscriber가 전달받지 못하고, 구독이 발생한 시점 이후에 emit된 데이터만 받을 수 있음
• share(), cache() 등의 operator를 사용해서 cold sequence를 hot sequence로 변환 가능 

 

 

 

 

Backpressure

Publisher로부터 전달받은 데이터를 안정적으로 처리하기 위한 수단

 

✳️ Backpressure 처리 방식 

 

🔷 데이터 개수 제어

• Subscriber가 request() 메서드를 통해 적절한 데이터 개수 요청 

public class Example {
	public static void main(String[] args) {
    	Flux.range(1, 5) 
        	.doOnRequest(data -> log.info("# doOnRequest : {}", data))
            .subscribe(new BaseSubscriber<Integer>() {
            	@Override
                protected void hookOnSubscribe(Subscription subscription) {
                	request(1);
                }
                
                @SneakyThrows
                @Override
                protected void hookOnNext(Integer value) {
                	// Publisher의 데이터 emit 속도보다 Subscriber의 데이터 처리 속도가 더 느린 것 시뮬레이션
                	Thread.sleep(2000L); // 2초 지연 시간
                    log.info("# hookOnNext: {}", value);
                    request(1);
                }
			});               
    }
}

 

 

 

🔷 Backpressure 전략 사용

• IGNORE 전략
  • Backpressure 적용 X
  • Downstream에서의 Backpressure 요청이 무시되기 때문에 IllegalStateException 발생
• ERROR 전략
  • Downstream의 데이터 처리 속도가 느려서 Upstream의 emit 속도를 따라가지 못할 경우 IllegalStateException 발생
  • Publisher는 Error signal을 subscriber에게 전송하고 삭제한 데이터는 폐기
• DROP 전략
  • Publisher가 Downstream으로 전달할 데이터가 버퍼에 가득 찰 경우, 버퍼 밖에서 대기 중인 데이터 중에서 먼저 emit된 데이터부터 Drop (폐기)
• LATEST 전략
  • Publisher가 Downstream으로 전달할 데이터가 버퍼에 가득 찰 경우, 버퍼 밖에서 대기 중인 데이터 중에서 가장 최근에 (나중에) emit된 데이터부터 버퍼에 채움
• BUFFER 전략
  • 버퍼의 데이터를 폐기하지 않고, 버퍼링(데이터를 쌓아 두었다가 전송)을 하는 전략, 버퍼가 가득차면 버퍼 내의 데이터를 폐기하는 전략, 버퍼가 가득 차면 에러를 발생시키는 전략 등으로 구분