본문으로 건너뛰기

Webflux기본 코드로 달려

0개 또는 1개의 결과를 처리

Mono.just()

Mono<String> monoString = Mono.just("Hello world!");

Iterable객체를 처리

Flux.fromIterable()

List<Integer> numbers = Arrays.asList(1,2,3,4,5);
Flux<Integer> fluxNumbers = Flux.fromIterable(numbers);

비동기 처리후 결과를 Mono로

Mono.fromCallable()

Callable<String> callableTask = () -> "Result from callable task";
Mono<String> monoFromCallable = Mono.fromCallable(callableTask);

여러개의 Flux를 순차적으로 연결

Flux.concat()

Flux<Integer> flux1 = Flux.just(1,2,3);
Flux<Integer> flux2 = Flux.just(4,5,6);
Flux<Integer> concatenatedFlux = Flux.concat(flux1, flux2);

여거래의 Flux를 병합하여 병렬처리

Flux.merge()

Flux<Integer> flux1 = Flux.just(1,3,5);
Flux<Integer> flux2 = Flux.just(2,4,6);
Flux<Integer> mergeFlux = Flux.merge(flux1, flux2);

Mono결합 -> 결과기다림 -> 튜플반환

Mono.zip()

Mono<String> mono1 = Mono.just("Hello");
Mono<String> mono2 = Mono.just("Webflux");
Mono<Tuple2<String, String>> zippedMono = Mono.zip(mono1, mono2);

Flux각 요소에 함수를 적용하여 변환

.map()

Flux<Integer> numbers = Flux.just(1,2,3);
Flux<String> mappedNumbers = numbers.map(num -> "Number: " + num);

Mono의 값을 조건에 따라 필터링함

.filter()

Mono<Integer> numberMono = Mono.just(5);
Mono<Integer> filteredMono = numberMono.filter(num -> num > 3);

Flux의 각 요소에 비동기 함수를 적용후 결과Flux를 병합

.flatMap()

Flux<Integer> numbers = Flux.just(1,2,3);
Flux<Integer> doubledNumbers = numbers.flatMap(num -> Mono.just(num*2));

Flux가 비어있을 때 대체 Flux를 사용

.switchIfEmpty()

Flux<Integer> numbers = Flux.empty();
Flux<Integer> defaultNumbers = Flux.just(1,2,3);
Flux<Integer> result = numbers.switchIfEmpty(defaultNumbers); // result: 1,2,3

Mono에서 에러가 발생 했을때 대체 Mono를 사용

.onErrorResume()

Mono<Integer> numberMono = Mono.just(10)
.map(num -> num / 0)
.onErrorResume(ex -> Mono.just(0));

Flux에서 발생시 조건에 따라 에러무시하고 진행

.onErrorContinue()

Flux<Integer> numbers = Flux.just(1,2,3,4);
Flux<Integer> result = numbers.map(num -> {
if (num == 3) {
throw new RuntimeException("Error!);
}
return num;
})
.onErrorContinue((ex, val) -> {
System.out.println("Error occurred, but continuing with value : " + val);
});

Mono나 Flux에서 각가의 값을 소비하기 전에 사전 처리 수행

.doOnSuccess()

Mono<Integer> numberMono = Mono.just(42);
numberMono = numberMono.doOnSuccess(num -> System.out.println("처리"));

Flux에서 값을 집계하여 단일 결과 Mono로 반환

.reduce()

Flux<Integer> numbers = Flux.just(1,2,3,4,5);
Mono<Integer> sumMono = numbers.reduce(0, (acc,num) -> acc + num);

Flux의 모든 요소를 리스트나 맵으로 수집

.collectMap()

Flux<Integer> numbers = Flux.just(1,2,3,4,5);
Mono<List<Integer>> listMono = numbers.collectList();
Mono<Map<Integer, String>> mapMono = numbers.collectMap(num -> num, num -> "Value : " + num);

주기적으로 값을 생성하는 Flux를 생성

.interval()

Flux<Long> intervalFlux = Flux.interval(Duration.OfSeconds(1));
intervalFlux.subscribe(value -> System.out.println("Received value : " + value));

두 개의 Flux를 조합하여 새로운 Flux를 생성

.zipWith()

Flux<Integer> flux1 = Flux.just(1,2,3);
Flux<Integer> flux2 = Flux.just(10,20,30);
Flux<Tuple2<Integer, Integer>> zippedFlux = flux1.zipWith(flux2);

지정된 개수만큼의 요소를 모아 List형태로 발행

.buffer()

Flux<Integer> numbers = Flux.range(1, 10);
Flux<List<Integer>> bufferedFlux = numbers.buffer(3);
// Output: [1, 2, 3], [4, 5, 6], [7, 8, 9], [10]

지정된 개수만큼의 요소를 가진 작은 Flux들로 구분하여 발행

.window()

Flux<Integer> numbers = Flux.range(1, 10);
Flux<Flux<Integer>> windowedFlux = numbers.window(3);
// Output: Flux[1, 2, 3], Flux[4, 5, 6], Flux[7, 8, 9], Flux[10]

에러가 발생할 경우 재시도

.retry()

Flux<Integer> fluxWithRetry = someFlux.retry(3) // 최대 3번 재시도

Flux각 요소에 비동기 함수 적용, 결과Flux 순차처리

.concatMap()

Flux<Integer> numbers = Flux.just(1,2,3);
Flux<Integer> processedNumbers = numbers.concatMap(num -> Mono.just(num * 2));

Flux각 요소에 비동기 함수 적용, 결과 Flux병렬처리. 순서보장x

.flatMapSequential()

Flux<Integer> numbers = Flux.just(1,2,3);
Flux<Integer> processedNumbers = numbers.flatMapSequential(num -> Mono.just(num * 2));

Mono나 Flux실행완료 후, 다른 Mono나 Flux실행

.then()

Mono<Integer> mono1 = Mono.just(1);
Mono<Integer> mono2 = Mono.just(2);
Mono<Void> voidMono = mono1.then(mono2);

Flux에서 일정시간이내 데이터 못받으면 에러발생

.timeout()

Flux<Integer> numbers = Flux.just(1,2,3).delayElements(Duration.ofSeconds(2));
Flux<Integer> timeoutFlux = numbers.timeout(Duration.ofSeconds(1));

Flux에서 지정한 개수만큼의 요소만 취함

.take()

Flux<Integer> numbers = Flux.just(1,2,3,4,5);
Flux<Integer> limitedFlux = numbers.take(3); // 결과 1,2,3

Flux에서 지정한 개수만큼 요소를 건너뜀

.skip()

Flux<Integer> numbers = Flux.just(1,2,3,4,5);
Flux<Integer> skippedFlux = numbers.skip(2); // 결과 3,4,5

중복된 요소를 제거

.distinct()

Flux<Integer> numbers = Flux.just(1,2,3,2,4,3,5);
Flux<Integer> distinctFlux = numbers.distinct(); // 결과 1,2,3,4,5

비동기 조건으로 Flux를 필터링

.filterWhen()

Flux<Integer> numbers = Flux.just(1,2,3,4,5);
Flux<Integer> filteredFlux = numbers.filterWhen(num -> Mono.just(num % 2 == 0)); // 결과 2,3

Flux가 비어있을 때 기본값을 발행

.defaultIfEmpty()

Flux<Integer> emptyFlux = Flux.empty();
Flux<Integer> defaultFlux = emptyFlux.defaultIfEmpty(0); // 결과 0

두개의 Flux를 합침

.mergeWith()

Flux<Integer> flux1 = Flux.just(1,2,3);
Flux<Integer> flux1 = Flux.just(4,5,6);
Flux<Integer> mergedFlux = flux1.mergeWith(flux2) // 결과 1,2,3,4,5,6

에러를 다른형태로 변환

.map()

Flux<Integer> numbers = Flux.just(1,2,3);
Flux<Integer> processedNumbers = numbers.map(num -> num / 0)
.onErrorMap(ArithmeticException.class, ex -> new RuntimeException("Division by zero!));

Flux와 Iterable객체를 조합하여 Tuple형태로 결합

.zipWithIterable()

Flux<Integer> flux = Flux.just(1,2,3);
List<String> names = Arrays.asList("Alice", "Bob", "Charlie");
Flux<Tuple2<Integer, String>> zippedFlux = flux.zipWithIterable(names);

Flux의 모든 요소를 정렬된 리스트로 수집

.collectSortedList()

Flux<Integer> numbers = Flux.just(3,1,4,2,5);
Mono<List<Integer>> sortedList = numbers.collectSortedList();

새로운 값을 생성하고 최신 결과만 유지

.switchMap()

Flux<String> words = Flux.just("apple", "banana", "cherry");
Flux<Integer> switchedFlux = words.switchMap(word -> Mono.just(word.length()));
// 결과 5, 6, 6

병렬로 실행된 결과를 보장하면서 Flux요소를 처리

.flatMapSequential()

Flux<Integer> flux = Flux.just(1,2,3);
Flux<Integer> processedFlux = flux.flatMapSequential(num -> Flux.just(num * 2).delayElements(Duration.ofMillis(num * 100)));
// 결과 2, 4, 6 (딜레이에 의해 병렬로 실행되어도 순서를 보장)

Mono의 결과를 캐싱하여 다시 사용할 수 있도록 한다

.cache()

Mono<String> data = Mono.fromCallable(() -> expensiveDatabaseCall());
Mono<String> cachedData = data.cache();

여러 Flux에서 최신 값들을 조합하여 새로운 결과를 발행

.combineLatest

Flux<Integer> flux1 = Flux.just(1,2,3);
Flux<Integer> flux2 = Flux.just(10,20,30);
Flux<Integer> combinedFlux = Flux.combineLatest(flux1, flux2, (num1, num2) -> num1 + num2);
// 결과 13, 23, 33 (최신 값임 3과 30을 조합, 1+30, 2+30, 3+30)

여러 Flux중 가장 먼저 도착한 Flux를 선택

.first()

Flux<Integer> flux1 = Flux.just(1,2,3).delayElements(Duration.ofSeconds(2));
Flux<Integer> flux2 = Flux.just(10,20,30).delayElements(Duration.ofSeconds(1));
Flux<Integer> firstFlux = Flux.first(flux1, flux2);
// 결과 10, 20, 30 (flux2가 먼저도착)

지정된 주기로 가장 최근의 값만 선택

.sample()

Flux<Integer> numbers = Flux.range(1,10).delayElements(Duration.ofMillis(50));
Flux<Integer> sampledFlux = numbers.sample(Duration.ofSeconds(2));
// 결과 2, 5, 8 (2초마다 최근 값만 선책))

Flux의 요소를 비동기적으로 처리하고 결과를 Iterable형태로 반환

.concatMapIterable()

Flux<Integer> flux = Flux.just(1,2,3);
Flux<String> processedFlux = flux.concatMapIterable(num -> Arrays.asList("A" + num, "B" + num));
// 결과 : A1, B1, A2, B2, A3, B3

에러가 발생할 경우 지정된 조건에 따라 재시도

.retryWhen()

Flux<Integer> fluxWithRetry = someFlux.retryWhen(errors -> errors.zipWith(Flux.range(1, 3), (error, retryCount) -> {
if (retryCount < 3) {
return retryCount;
}
return error;
}).flatMap(errorOrRetryCount -> {
if (errorOrRetryCount instanceof Throwable) {
return Mono.error((Throwable) errorOrRetryCount);
}
return Mono.delay(Duration.ofSeconds(1));
}));

Flux 각요소에 대해 콜백을 제공하여 변환 및 필터링

.handle()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<String> processedNumbers = numbers.handle((num, sink) -> {
if (num % 2 == 0) {
sink.next("Even: " + num);
}
});

Mono의 값을 비동기 조건으로 필터링

.filterWhen()

Mono<Integer> numberMono = Mono.just(5);
Mono<Integer> filteredMono = numberMono.filterWhen(num -> Mono.just(num > 3));

첫번째 요소에대해 특별로직 수행, 나머지Flux반환

.switchOnFirst()

Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<Integer> processedFlux = numbers.switchOnFirst((signalType, value) -> {
if (signalType == SignalType.ON_NEXT) {
System.out.println("First element: " + value.get());
}
return value;
});

Flux가 완료되거나 에러가 발생했을 때 특정 동작실행

.doFinally()

Flux<Integer> numbers = Flux.just(1, 2, 3);
numbers = numbers.doFinally(signalType -> {
if (signalType == SignalType.ON_COMPLETE) {
System.out.println("Flux completed.");
}
});

Flux를 반복해서 재생

.repeat()

Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<Integer> repeatedFlux = numbers.repeat(2); // 2번 반복

Mono의 값을 무시하고 완료 신호만 유지

.ignoreElement()

Mono<Integer> numberMono = Mono.just(42);
Mono<Void> completionMono = numberMono.ignoreElement();

초기값과 함께 Flux의 요소를 집계하고 결과를 차례로 발행

.scan()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> accumulatedFlux = numbers.scan((acc, num) -> acc + num);
// 결과: 1, 3, 6, 10, 15

Flux의 각 요소를 비동기처리후 결과를 Iterable형태로 변환

.concatMapIterable()

Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<String> processedFlux = flux.concatMapIterable(num -> Arrays.asList("A" + num, "B" + num));
// 결과: A1, B1, A2, B2, A3, B3

기준에 따라 Flux요소를 그룹핑

.groupBy()

Flux<String> animals = Flux.just("cat", "dog", "cow", "elephant", "camel");
Flux<GroupedFlux<Character, String>> groupedFluxes = animals.groupBy(animal -> animal.charAt(0));

재시도 로직을 사용하고 에러를 반환

.retryWhen()

Flux<Integer> fluxWithRetryAndErrorHandling = someFlux.retryWhen(errors -> errors.zipWith(Flux.range(1, 3), (error, retryCount) -> {
if (retryCount < 3) {
return retryCount;
}
return error;
}).flatMap(errorOrRetryCount -> {
if (errorOrRetryCount instanceof Throwable) {
return Mono.error((Throwable) errorOrRetryCount);
}
return Mono.delay(Duration.ofSeconds(1));
})).onErrorMap(IllegalStateException.class, ex -> new CustomException("Custom error occurred.", ex));

Flux와 Mono를 연결하고 모든 작업이 완료될 때까지 기다림

.concatWith()

Flux<Integer> flux1 = Flux.just(1, 2, 3);
Flux<Integer> flux2 = Flux.just(4, 5, 6);
Mono<Integer> mono1 = Mono.just(7);
Mono<Integer> mono2 = Mono.just(8);

Flux<Integer> concatenatedFlux = flux1.concatWith(flux2).concatWith(Mono.when(mono1, mono2));

에러가 발생할 경우 대체 Flux로 처리

.onErrorResume()

Flux<Integer> fluxWithFallback = someFlux.map(num -> num / 0)
.onErrorResume(ex -> Flux.range(1, 5)); // 에러가 발생하면 대체 Flux로 처리

특정 조건에 따라 Flux의 요소를 버퍼링

.bufferUntil() / .bufferWhile()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9);
Flux<List<Integer>> bufferedUntilFlux = numbers.bufferUntil(num -> num % 3 == 0); // 3의 배수일 때 버퍼링
Flux<List<Integer>> bufferedWhileFlux = numbers.bufferWhile(num -> num % 2 == 0); // 짝수일 때 버퍼링

Flux의 데이터 생성과 구독이 다른스레드에서 수행

어렵..

Flux<Integer> flux = Flux.range(1, 5)
.map(num -> {
System.out.println("Mapping on thread: " + Thread.currentThread().getName());
return num * 2;
})
.publishOn(Schedulers.parallel()) // 데이터 생성과 관련된 작업을 병렬 스레드로 수행
.map(num -> {
System.out.println("Mapping again on thread: " + Thread.currentThread().getName());
return num + 1;
})
.subscribeOn(Schedulers.single()); // 구독과 관련된 작업을 단일 스레드로 수행

flux.subscribe(System.out::println);

Backpressure를 처리

.onBackpressureBuffer()

Flux<Integer> sourceFlux = Flux.range(1, 1000);
Flux<Integer> bufferedFlux = sourceFlux.onBackpressureBuffer(100); // 최대 100개의 요소만 버퍼링하도록 설정

동적으로 요청하고 응답을 처리

Flux<String> requests = Flux.just("request-1", "request-2", "request-3");
Flux<String> processedResponses = requests.switchMap(request -> {
// request를 기반으로 비동기 작업 수행 (예: 외부 API 호출)
return WebClient.create().get()
.uri("https://api.example.com/data?query=" + request)
.retrieve()
.bodyToMono(String.class);
});

Flux의 최신 값을 조합

.combineLatest()

Flux<Long> interval1 = Flux.interval(Duration.ofSeconds(1));
Flux<Long> interval2 = Flux.interval(Duration.ofMillis(500));

Flux<Tuple2<Long, Long>> combinedFlux = Flux.combineLatest(interval1, interval2, Tuple2::of);
// 두 Flux 중 하나라도 값이 갱신되면 최신 값을 조합하여 Tuple을 발행합니다.

Flux<Tuple2<Long, Long>> withLatestFromFlux = interval1.withLatestFrom(interval2, Tuple2::of);
// 첫 번째 Flux의 값이 갱신될 때마다 두 번째 Flux의 최신 값을 조합하여 Tuple을 발행합니다.

반복을 컨트롤

어렵..

AtomicInteger counter = new AtomicInteger();

Flux<String> dataFlux = Flux.defer(() -> {
int count = counter.getAndIncrement();
return count < 5 ? Flux.just("Data-" + count) : Flux.empty();
});

dataFlux.repeatWhen(companion -> companion.take(3).delayElements(Duration.ofSeconds(2)))
.subscribe(System.out::println);

구독자가 동일한 시퀀스를 공유

어렵다..

Flux<Integer> sourceFlux = Flux.range(1, 5)
.doOnSubscribe(subscription -> System.out.println("New subscriber connected"));

ConnectableFlux<Integer> connectableFlux = sourceFlux.publish();

connectableFlux.subscribe(num -> System.out.println("Subscriber 1: " + num));
connectableFlux.subscribe(num -> System.out.println("Subscriber 2: " + num));

connectableFlux.connect();

Flux의 중복 로직을 재사용

.compose()

Function<Flux<String>, Flux<String>> toUpperCase = stringFlux -> stringFlux.map(String::toUpperCase);
Function<Flux<String>, Flux<String>> trimWhitespace = stringFlux -> stringFlux.map(String::trim);

Flux<String> dataFlux = Flux.just(" hello ", " world ");
dataFlux.compose(trimWhitespace).compose(toUpperCase).subscribe(System.out::println);

Flux의 각 요소 사의 시간 간격을 측정, 새로운 Flux생성

.elapsed()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<Tuple2<Integer, Long>> elapsedFlux = numbers
.elapsed()
.doOnNext(tuple -> System.out.println("Elapsed: " + tuple.getT2() + "ms"));

두개의 Flux중 하나가 요소를 발행하면 다른 Flux취소

.takeUntilOther()

Flux<Integer> sourceFlux = Flux.just(1, 2, 3, 4, 5);
Flux<Long> durationFlux = Flux.interval(Duration.ofSeconds(3));

Flux<Integer> takeUntilOtherFlux = sourceFlux.takeUntilOther(durationFlux);
takeUntilOtherFlux.subscribe(System.out::println);

지정범위 내에서 순차적인 숫자 발행

.range()

Flux<Integer> rangeFlux = Flux.range(1, 5);
rangeFlux.subscribe(System.out::println);

에러발생시 대체값

.onErrorReturn()

Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<Integer> fluxWithFallback = numbers.map(num -> num / 0)
.onErrorReturn(0); // 에러가 발생하면 대체 값 0으로 처리

지정된 시간마다 새로운 Flux생성

.windowTimeout()

Flux<Integer> numbers = Flux.range(1, 10);
Flux<Flux<Integer>> windowedFlux = numbers.windowTimeout(3, Duration.ofSeconds(2));
windowedFlux.subscribe(window -> window.collectList().subscribe(System.out::println));

두개의 Flux를 병합

.zip()

Flux<Integer> flux1 = Flux.just(1, 2, 3);
Flux<String> flux2 = Flux.just("A", "B", "C");

Flux<Tuple2<Integer, String>> zippedFlux = Flux.zip(flux1, flux2);
zippedFlux.subscribe(tuple -> System.out.println(tuple.getT1() + tuple.getT2()));

Flux중 가장 먼저 도착한 Flux를 선택

.firstEmitting()

Flux<Integer> flux1 = Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(2));
Flux<Integer> flux2 = Flux.just(10, 20, 30).delayElements(Duration.ofSeconds(1));

Flux<Integer> firstFlux = Flux.firstEmitting(flux1, flux2);
firstFlux.subscribe(System.out::println);

Flux에서 단 하나의 요소를 가져옴

.single()

Flux<Integer> numbers = Flux.just(42);
numbers.single()
.doOnSuccess(num -> System.out.println("Single value: " + num))
.doOnError(ex -> System.out.println("Error: " + ex.getMessage()))
.subscribe();

Flux의 요소를 하나의 결과로 줄임

.reduce()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Mono<Integer> sum = numbers.reduce((x, y) -> x + y);
sum.subscribe(result -> System.out.println("Sum: " + result));

여러 Flux를 합침

.merge()

Flux<Integer> flux1 = Flux.just(1, 2, 3);
Flux<Integer> flux2 = Flux.just(4, 5, 6);
Flux<Integer> mergedFlux = Flux.merge(flux1, flux2);
mergedFlux.subscribe(System.out::println);

Flux<Integer> concatenatedFlux = Flux.concat(flux1, flux2);
concatenatedFlux.subscribe(System.out::println);

Flux에서 조건에 만족하는 요소를 확인

.any()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
numbers.any(num -> num > 3)
.subscribe(result -> System.out.println("Any number greater than 3: " + result));

Flux에서 모든 요소가 조건에 만족하는지 확인

.all()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
numbers.all(num -> num > 0)
.subscribe(result -> System.out.println("All numbers are greater than 0: " + result));

Flux의 모든 요소를 리스트로 수집

.collectList()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
numbers.collectList()
.subscribe(list -> System.out.println("List of numbers: " + list));

Flux의 요소를 키-값 쌍으로 수집하여 Map으로 반환

.collectMap()

Flux<String> animals = Flux.just("cat", "dog", "elephant", "giraffe");
animals.collectMap(animal -> animal.charAt(0))
.subscribe(map -> System.out.println("Animals map: " + map));

초기값과 함께 Flux의 요소를 집계하고 결과를 발행

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
numbers.scanWith(() -> 0, (acc, num) -> acc + num)
.subscribe(result -> System.out.println("Accumulated sum: " + result));

주어진 시간 간격으로 최근에 발행된 Flux의 요소를 샘플링

.sample()

Flux.interval(Duration.ofMillis(200))
.sample(Duration.ofSeconds(1))
.take(5)
.subscribe(System.out::println);

Flux가 완료되었을 때 지정된 조건에 따라 반복

.repeatWhenEmpty()

Flux<Integer> flux = Flux.just(1, 2, 3).concatWith(Flux.empty());
flux.repeatWhenEmpty(companion -> companion.take(2)).subscribe(System.out::println);

에러가 발생하더라도 모든 Flux를 처리

.concatMapDelayError

Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<Integer> processedFlux = numbers.concatMapDelayError(num -> {
if (num == 2) {
return Mono.error(new RuntimeException("Error processing number 2"));
}
return Mono.just(num * 2);
}, 1); // 1로 설정하여 에러 발생 시 해당 에러만 처리하고 나머지 Flux는 에러 발생 이후에 처리

지정된 시간만큼 구독을 지연

.delaySubscription()

Flux<Integer> numbers = Flux.range(1, 5);
numbers.delaySubscription(Duration.ofSeconds(3)).subscribe(System.out::println);

Flux각 요소를 비동기처리후 결과 Iterable형태로 반환

.concatMapIterable()

Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<String> processedFlux = flux.concatMapIterable(num -> Arrays.asList("A" + num, "B" + num));

지정된 조건에 따라 Flux를 반복

.repeatWhen()

Flux<Integer> flux = Flux.range(1, 3);
Flux<Integer> repeatedFlux = flux.repeatWhen(companion -> companion.take(2));

각 요청마다 새로운 Flux를 생성하여 이전요청 취소

.switchMap()

Flux<String> requests = Flux.just("request-1", "request-2", "request-3");
Flux<String> processedResponses = requests.switchMap(request -> {
// request를 기반으로 비동기 작업 수행 (예: 외부 API 호출)
return WebClient.create().get()
.uri("https://api.example.com/data?query=" + request)
.retrieve()
.bodyToMono(String.class);
});

배압 처리

.onBackpressureBuffer()

Flux<Integer> sourceFlux = Flux.range(1, 1000);
Flux<Integer> bufferedFlux = sourceFlux.onBackpressureBuffer(100); // 최대 100개의 요소만 버퍼링하도록 설정

첫번째 요소에 대해 로직 수행후, 나머지 Flux반환

.switchOnFirst()

Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<Integer> processedFlux = numbers.switchOnFirst((signalType, value) -> {
if (signalType == SignalType.ON_NEXT) {
System.out.println("First element: " + value.get());
}
return value;
});

Flux각 요소에 콜백을 제공하여 변환 또는 필터링

.handle()

Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<String> processedNumbers = numbers.handle((num, sink) -> {
if (num % 2 == 0) {
sink.next("Even: " + num);
}
});

Flux가 완료되거나 에러발생시 특정 동작 실행

.doFinally()

Flux<Integer> numbers = Flux.just(1, 2, 3);
numbers = numbers.doFinally(signalType -> {
if (signalType == SignalType.ON_COMPLETE) {
System.out.println("Flux completed.");
}
});

Flux와 Mono를 연결하고 모든 작업이 완료될때까지 기다림

.concatWith()

Flux<Integer> flux1 = Flux.just(1, 2, 3);
Flux<Integer> flux2 = Flux.just(4, 5, 6);
Mono<Integer> mono1 = Mono.just(7);
Mono<Integer> mono2 = Mono.just(8);

Flux<Integer> concatenatedFlux = flux1.concatWith(flux2).concatWith(Mono.when(mono1, mono2));

특정 기준에 따라 Flux의 요소를 그룹핑

.groupBy()

Flux<String> animals = Flux.just("cat", "dog", "cow", "elephant", "camel");
Flux<GroupedFlux<Character, String>> groupedFluxes = animals.groupBy(animal -> animal.charAt(0));