[WebFlux] Reactor 연산자 정리

flatMapMany

flatMapMany

Mono에서 여러 개 요소를 생성할 때 사용 
👉 Mono에서 반환된 결과를 여러 개의 요소로 변환하여 Flux로 처리할 때 유용

 

 

 

fromIterable

Iterable 타입의 데이터(list, set, map 등의 컬렉션)를 Flux로 변환 
👉 컬렉션 내의 요소들을 리액티브 스트림으로 변환하여 비동기적으로 처리

 

 

map, mapNotNull

map은 onNext 이벤트를 받아서 값을 변경하고 아래로 전달 

Flux.range(1, 5)
    .map(value -> value * 2)
    .doOnNext(value -> {
    	log.info("doOnNext: " + value);
    })
	.subscribe();​

mapNotNull은 변경된 값이 null인 경우 필터 
👉 Flux에서 null 값을 onNext로 전달하면 에러 발생

Flux.range(1, 5)
    .mapNotNull(value -> {
        if (value % 2 == 0) {
        	return value;
    	}
    	return null;
    })
    .doOnNext(value -> {
    	log.info("doOnNext: " + value);
    })
	.subscribe();​

 

 

 

doOnXX

doOnSubscribe, doOnNext, doOnComplete, doOnError 등

각각의 이벤트를 흐름에 영향을 주지 않고 위에서 내려오는 이벤트에 대해서 로깅이나 추가 작업 가능 

 

Flux.range(1, 5)
    .map(value -> value * 2)
    .doOnNext(value -> {
    	log.info("doOnNext: " + value);
    })
    .doOnComplete(() -> {
    	log.info("doOnComplete");
    })
    .doOnSubscribe(subscription -> {
    	log.info("doOnSubscribe");
    })
    .doOnRequest(value -> {
    	log.info("doOnRequest: " + value);
    })
    .map(value -> value / 2)
.subscribe();

 

✴️ 이벤트 발생 순서 

1. doOnSubscribe : Flux가 구독될 때 발생

2. doOnRequest : 구독자가 데이터를 요청할 때 발생 👉 구독이 시작될 때 구독자가 publisher에게 최대한 많은 데이터를 받고자 함 (Long.MAX_VALUE)

3. 데이터 발행 및 처리

- Flux.range(1, 5)에서 1~5까지의 값이 순차적으로 발행되며 map 연산자에서 각 값이 2배로 변환
- doOnNext 연산자에서 변환된 각 값이 로그에 기록됨 

 

4. doOnComplete : 모든 값이 성공적으로 발행되고 완료되었을 때 발생 

---

✳️ 결과값 

doOnSubscribe
doOnRequest: 9223372036854775807
doOnNext: 2
doOnNext: 4
doOnNext: 6
doOnNext: 8
doOnNext: 10
doOnComplete

 

 

flatMap

onNext 이벤트를 받아서 publisher를 반환하고 publisher의 이벤트를 아래로 전달
여러 publisher를 조합해야하는 경우 유용 

 

Flux.range(1, 5)
    .flatMap(value -> {
    	return Flux.range(1, 2)
    		.map(value2 -> value + ", " + value2)
    		.publishOn(Schedulers.parallel());
    })
    .doOnNext(value -> {
    	log.info("doOnNext: " + value);
    })
	.subscribe();

 

• flatMap은 각 요소를 처리할 때마다 내부적으로 새로운 Flux 생성 
• publisher가 비동기로 동작할 때 순서 보장 X
• publishOn(Schedulers.parallel())에 의해 내부 Flux의 작업이 병렬로 처리됨

 

✳️ 결과값

doOnNext: 1, 1
doOnNext: 1, 2
doOnNext: 2, 1
doOnNext: 2, 2
doOnNext: 3, 1
doOnNext: 3, 2
doOnNext: 4, 1
doOnNext: 4, 2
doOnNext: 5, 1
doOnNext: 5, 2

---

🚨 flatMap과 map의 차이?

1. 변환 범위 
🔹 map : 각 요소를 변환하여 새로운 값 생성 👉단순한 매핑 작업
🔹 flatMap : 각 요소를 변환하고 그 결과로부터 여러 개의 Flux나 Mono 생성하여 하나의 스트림으로 병합 👉 병렬처리나 다양한 비동기 작업에 유리 

2. 출력 스트림 형태
🔹 map : 각 요소를 변환하고 원래의 요소와 동일한 타입의 스트림 생성
🔹 flatMap : 변환 결과로 생성된 여러 개의 스트림을 하나로 병합하여 하나의 스트림 생성 

 

 

 

concatMap

각 입력 요소에 대해 비동기 작업을 순차적으로 실행
내부적으로 순서를 보장하며, 한 작업이 완료된 후 다음 작업을 시작

 

 

 

flatMapSequential

비동기 작업을 순차적으로 수행할 때 사용
flatMap과 다르게 순서 보장

 

 

🌟 flatMapSequential과 concatMap의 차이 

concatMap은 인자로 지정된 함수에서 리턴하는 Publisher의 스트림이 다 끝난 후에 그다음 넘어오는 값의Publisher 스트림을 처리
👉 결과 다소 느리게 반환하므로 작업 간의 의존성이 강한 경우, 순서가 보장되어야 할 때 사용됨

flatMapSequential은 일단 오는 대로 구독하고 결과는 순서에 맞게 리턴하는 역할
👉 비동기 작업이 병렬로 실행되더라도 순서가 중요한 경우 사용됨

 

 

filter

onNext 이벤트를 받아서 boolean 반환해서 true라면 onNext 이벤트 전파하고 false면 필터

👉 데이터 스트림에서 조건을 만족하는 요소만 통과시키며 조건에 맞지 않는 요소는 필터링하여 제외해 필요한 부분만 선택적으로 처리  

 

Flux.range(1, 10)
    .filter(value -> value % 2 == 0) // 짝수만 필터링
    .subscribe(System.out::println);

 

collectList

next 이벤트가 전달되면 내부에 item을 저장하고 complete 이벤트가 전달되면 저장했던 item들을 list 형태로 만들어 아래에 onNext 발행

👉 Flux를 Mono로 바꿀 때 유용

 

Flux.range(1, 3)
    .map(value -> value * 10)
    .collectList()
    .subscribe(list -> System.out.println("Collected list: " + list));
    
// 출력: Collected list: [10, 20, 30]

 

 

reduce

데이터를 집계(aggregate)할 때 사용되는 연산자로, 주로 스트림의 모든 요소를 하나의 값으로 결합하는 데 사용

 

 

 

 

참고자료

https://tech.kakao.com/posts/350