[Spring WebFlux] Project Reactor 용어를 예제와 함께 보기

 

 

Reactor, RxJava, WebFlux에서 사용되는 용어들

 

Spring WebFlux, Project Reactor 등 Reactive Streams가 사용되는 프로젝트를 개발하면서 구글링을 하면 아래의 용어들을 만나볼 수 있다.

앞으로 구글링할 때 읽다가 단어의 뜻을 몰라 브레이크가 걸리지 않도록 용어의 뜻을 예제와 함께 알아보자

 

Publisher : 발행자, 생산자, Emitter 데이터를 생산하는 주체

Emit : Publisher가 데이터를 Downstream(아래 참고)으로 방출하는 것

Subscriber : 구독자, 소비자 데이터를 소비하는 주체

Sequence : Publisher가 emit하는 데이터의 연속적인 흐름 stream, pipeline과도 의미가 같다.

Subscribe : Subscriber가 Upstream Sequence를 구독하는 것

 

Publisher가 Emit하는 요소들이 Reactive Streams Operator들을 지나서 가공되는 흐름을 Sequence 라고 한다.

Subscriber가 Cold Publisher를 Subscribe 하게될 때 Publisher가 요소들을 Emit하기 시작한다.

Subscriber가 Cold Publisher를 Subscribe 하게될 때 Reactive Stream에서 데이터가 흐르기 시작한다.

Hot Publisher는 Subscriber의 유무와 관계없이 데이터를 생산하는 상태이다.

Subscriber가 Hot Publisher를 Subscribe 하는 경우 구독 이후 생산된 요소들만 소비할 수 있다.

 

 

예제로 알아보기

 

@Test
void pubsublearn() {
	//@1
	Flux<Integer> publisher = Flux.just(1, 2, 3, 4)
		.filter(i -> i % 2 == 0);

	//@2
	BaseSubscriber<Integer> subscriber = new BaseSubscriber<>() {
		@Override
		protected void hookOnNext(Integer value) {
			System.out.println("Received: " + value);
		}
	};

	// @3 구독관계 생성
	publisher.subscribe(subscriber);

	// @4 주로 사용되는 subscribe 패턴
	publisher.subscribe(i -> System.out.println("Received: " + i));

	// @5 하나의 연결된 파이프라인으로 표현
	Flux.just(1, 2, 3, 4)
		.filter(i -> i % 2 == 0)
		.subscribe(i -> System.out.println("Received: " + i));
}

 

@1 : 0~n개의 Integer를 {발행, 생산, emit} 하는 publisher가 정의되어있다.

Flux.just(1, 2, 3, 4) 에서는 1,2,3,4를 방출할 것이고, .filter(i -> i % 2 == 0) 에서는 2,4를 방출하게 될 것이다.

 

@2 : 어떤 Publisher가 {발행, 생산, emit} 하는 {데이터, 요소} 들을 구독할 subscriber가 정의되어있다.

OnNext로 방출되는 각 요소를 hook 하여 정의한 데이터 소비행위를 한다.

 

@3 : 1,2 번 과정만 수행한경우 publisher, subscriber가 서로를 모르기때문에 publisher는 데이터를 발행하지 않고, subscriber는 소비할 데이터가 없는 상태이다.

하지만 구독관계를 생성하게되면 publisher가 emit한 2,4 가 소비되는것을 확인할 수 있다.

 

 

@4 : 모든 Subscriber를 2번과 같이 정의해야한다면 편리하지도 않고 stream을 흘러가는대로 이해할 수 없게된다. 하지만 subscribe에 전달되는 매개변수인 CoreSubscriber 대신 함수형 인터페이스인 Consumer 를 활용할 수 있다.

BaseSubscriber 에서 hookOnNext소비되는 각각의 요소에 취할 functional method인 void accept(Object) 를 람다식으로 전달하면 됨

 

@5 : sequence가 눈에 잘 띄도록 publisher → subscriber 를 연결해놓은 코드, {sequence, stream, pipeline} 등 세세하게는 뜻이 다를 수 있어도 비슷한 의미로 표현된다.

 

Downstream : 현재 Operator 를 기준으로 하위 Operator 및 method 체인

Upstream : 현재 Operator 를 기준으로 상위 Operator 및 method 체인

 

 

어떤 operator를 기준으로 publisher방향에 위치한 operator들을 말할 땐 upstream

어떤 operator를 기준으로 consumer(subscriber) 방향에 위치한 operator들을 말할 땐 downstream이라고 부른다.

 	                   upstream           downstream
publisher/past operators <--------- operator -----------> consumer/further operators

 

아래 주석메세지를 보면 Downstream 과 Upstream 의 사용 용례를 알 수 있다.

Flux.just(1, 2, 3, 4, 5, 100, 1000)
	// 위의 just는 아래 filter의 Upstream이다.
	.filter(i -> i >= 100)
	// 아래 map, subscribe는 위 filter의 Downstream이다.
	.map(String::valueOf)
	.subscribe(System.out::println);

 

 

Dispose : Suscriber가 Sequence의 구독을 해지 하는 것

 

@Test
void dispose() throws InterruptedException {

	// @1 도매상의 펜 재고, 초록색 펜이 5번째에 있음에 유의
	List<Pen> penList = new ArrayList<>();
	penList.add(new Pen("검정"));
	penList.add(new Pen("불량"));
	penList.add(new Pen("빨강"));
	penList.add(new Pen("불량"));
	penList.add(new Pen("초록"));

	// @2 penList에서 매 초 Pen 1개 씩 emit되어 불량인지 검사하는 도매상의 판매 과정
	Flux<Pen> penWholesaler = Flux.fromIterable(penList).delayElements(Duration.ofSeconds(1))
		.filter(Pen::checkQuality);
	
	// @3 소매상의 도매상(Publisher)으로부터 생성되는 펜들(Sequence)을 구입(Subscribe)하여 판매(소비)하는 계약 성립
	Disposable contract = penWholesaler.subscribe(
		pen -> System.out.printf("도매상에게서 %s을(를) 100원에 샀다.%n손님에게 %s을(를) 500원에 팔았다.%n", pen, pen));

	// @4 4초 후 계약을 취소(subscribe 후 dispose)했으므로 4초간 pub-sub stream에서 데이터가 흐를 것,
	// 따라서 5번째 펜 재고인 초록색 펜은 결과로 출력되지 않는다.
	Thread.sleep(4000);
	contract.dispose(); // @5
}

 

테스트 실행 결과

 

@3 에서 subscribe하여 생긴 pub-sub관계를 Disposable 형식의 객체로 받아올 수 있다.

언제든지 @5 처럼 dispose()를 호출하여 pub-sub 관계를 해제할 수 있다는 의미로 이해하면 좋을 것 같다.

 

[참고] Pen class 구현

public static class Pen {
	String color;

	public Pen(String color) {
		this.color = color;
	}

	public boolean checkQuality() {
		return !Objects.equals("불량", this.color);
	}

	@Override
	public String toString() {
		return this.color + "색 펜";
	}
}