Reactor Operators: flatMap, concatMap, zip, merge & buffer

Reactor's operator API is the core of reactive programming. Choosing the wrong operator is one of the most common sources of bugs in reactive code: flatMap vs concatMap vs switchMap each have very different concurrency and ordering semantics. This article covers the operators that matter most in production: the flatMap family, stream combination, batching with buffer and window, aggregation with reduce and scan, and time-based operators.

Java

flatMap vs concatMap vs switchMap
Combining Streams: zip, merge, concat & combineLatest
Batching: buffer & window
groupBy: Splitting by Key
Aggregation: reduce & scan
Side-Effect Operators: doOn*
Time-Based Operators: timeout, delayElements & sample
Choosing the Right Operator

flatMap vs concatMap vs switchMap

All three operators apply an async function that returns a publisher for each input item. The difference is how they handle concurrency and what happens when a new item arrives before the previous inner publisher has finished.

Operator	Concurrency	Order Guaranteed	Use When
`flatMap`	Concurrent (all inner publishers run in parallel)	No	Independent async calls where order doesn't matter
`concatMap`	Sequential (waits for each inner publisher to complete)	Yes	Order matters, or inner publishers have side effects
`switchMap`	Cancels previous inner publisher on new item	Only latest	Search-as-you-type, only the most recent request matters

Java

// flatMap — concurrent, results may arrive out of order
Flux.just("user-1", "user-2", "user-3")
    .flatMap(id -> userService.findById(id)        // all 3 calls run at once
        .subscribeOn(Schedulers.boundedElastic()))
    .subscribe(user -> System.out.println("Got: " + user));

// flatMap with concurrency limit — max 2 concurrent inner subscriptions
Flux.range(1, 100)
    .flatMap(id -> fetchOrder(id), 2)              // at most 2 in-flight
    .subscribe(System.out::println);

// concatMap — sequential, guaranteed order
Flux.just("step-1", "step-2", "step-3")
    .concatMap(step -> executeStep(step))          // step-1 completes, then step-2, then step-3
    .subscribe(result -> System.out.println("Done: " + result));

// switchMap — cancels previous on new item (perfect for live search)
searchInput                                        // Flux of keystroke events
    .debounce(Duration.ofMillis(300))
    .switchMap(query -> searchService.search(query))  // cancels previous search
    .subscribe(results -> displayResults(results));

Combining Streams: zip, merge, concat & combineLatest

Reactor provides four distinct operators for combining multiple publishers. Each has different semantics around timing and what triggers emission of a combined element.

Java

// zip — pairs items by position, waits for all publishers to emit
// Completes when the shortest publisher completes
Mono<User>     userMono    = userService.findById(1L);
Mono<Account>  accountMono = accountService.findByUserId(1L);

Mono<UserProfile> profile = Mono.zip(userMono, accountMono)
    .map(tuple -> new UserProfile(tuple.getT1(), tuple.getT2()));

// zip up to 8 publishers — use Tuples
Mono.zip(fetchName(), fetchEmail(), fetchRole())
    .map(t -> new UserDTO(t.getT1(), t.getT2(), t.getT3()))
    .subscribe(System.out::println);

// merge — interleaves items from multiple Flux in emission order (concurrent)
Flux<String> fast = Flux.interval(Duration.ofMillis(100)).map(i -> "fast-" + i);
Flux<String> slow = Flux.interval(Duration.ofMillis(300)).map(i -> "slow-" + i);

Flux.merge(fast, slow)
    .take(10)
    .subscribe(System.out::println);   // fast-0, fast-1, fast-2, slow-0, ...

// concat — subscribes to publishers sequentially (second starts when first ends)
Flux<String> first  = Flux.just("a", "b", "c");
Flux<String> second = Flux.just("d", "e", "f");

Flux.concat(first, second)
    .subscribe(System.out::println);   // a, b, c, d, e, f  (guaranteed order)

// combineLatest — emits whenever any publisher emits, combining with latest from others
Flux<String> stockPrice = priceStream();   // emits on price change
Flux<String> currency   = fxStream();      // emits on FX rate change

Flux.combineLatest(stockPrice, currency,
    (price, fx) -> convertPrice(price, fx))
    .subscribe(System.out::println);

Batching: buffer & window

buffer collects items into List batches and emits each complete batch downstream: useful for bulk database inserts or batched API calls. window does the same but emits a Flux per batch instead of a List, letting you start processing a window before it closes.

Java

// buffer by count — collect into batches of 10
Flux.range(1, 100)
    .buffer(10)                              // emits List<Integer> of 10 items each
    .subscribe(batch -> bulkInsert(batch));  // 10 lists of 10

// buffer by time — collect everything received within 1 second
Flux.interval(Duration.ofMillis(200))
    .buffer(Duration.ofSeconds(1))           // batches: [0,1,2,3,4], [5,6,7,8,9], ...
    .subscribe(System.out::println);

// buffer by count OR time — whichever comes first
Flux.interval(Duration.ofMillis(200))
    .bufferTimeout(10, Duration.ofSeconds(1))
    .subscribe(System.out::println);

// window by count — each window is a Flux (start processing before window closes)
Flux.range(1, 20)
    .window(5)
    .flatMap(windowFlux -> windowFlux
        .reduce(0, Integer::sum))             // sum each window of 5
    .subscribe(System.out::println);          // 15, 40, 65, 90

// Sliding window — overlapping windows
Flux.range(1, 10)
    .buffer(3, 1)   // window size 3, step 1 → [1,2,3], [2,3,4], [3,4,5], ...
    .subscribe(System.out::println);

groupBy: Splitting by Key

groupBy partitions a Flux into multiple inner Flux streams, one per key. Each inner GroupedFlux carries its key and must be subscribed to (usually via flatMap) or items will be buffered indefinitely.

Java

// Group orders by status and count each group
Flux.fromIterable(orders)
    .groupBy(Order::getStatus)
    .flatMap(group -> group
        .count()
        .map(count -> group.key() + ": " + count))
    .subscribe(System.out::println);
// PENDING: 12, CONFIRMED: 8, SHIPPED: 25

// Group incoming events by type and process each type differently
eventFlux
    .groupBy(Event::getType)
    .flatMap(group -> {
        if (group.key() == EventType.PAYMENT) {
            return group.flatMap(paymentHandler::handle);
        } else if (group.key() == EventType.REFUND) {
            return group.flatMap(refundHandler::handle);
        }
        return group.flatMap(defaultHandler::handle);
    })
    .subscribe();

Every GroupedFlux must be subscribed to (e.g., via flatMap). If you ignore a group, its items buffer in memory until the upstream completes, causing memory leaks on long-running streams. Always use flatMap or concatMap directly on the groupBy result.

Aggregation: reduce & scan

reduce aggregates all items into a single value and emits it when the sequence completes, like a stream reduce(). scan emits the running accumulator value after each item: useful for running totals or state machines that emit intermediate results.

Java

// reduce — sum all items, emit final result when stream completes
Flux.range(1, 10)
    .reduce(0, Integer::sum)              // Mono<Integer>
    .subscribe(System.out::println);      // 55

// reduce — collect into a map
Flux.fromIterable(orders)
    .reduce(new HashMap<String, Long>(), (acc, order) -> {
        acc.merge(order.getCustomerId(), 1L, Long::sum);
        return acc;
    })
    .subscribe(map -> System.out.println("Order counts: " + map));

// scan — emit running total after each item (Flux, not Mono)
Flux.range(1, 5)
    .scan(0, Integer::sum)
    .subscribe(System.out::println);      // 0, 1, 3, 6, 10, 15

// scan as a state machine — accumulate event log
Flux.fromIterable(events)
    .scan(OrderState.CREATED, (state, event) -> state.apply(event))
    .subscribe(state -> System.out.println("State: " + state));

Side-Effect Operators: doOn*

The doOn* operators let you attach side effects (logging, metrics, debugging) at specific lifecycle points without modifying the stream. They receive the item or signal and return void: the original value passes through unchanged.

Java

Flux.range(1, 5)
    .doOnSubscribe(sub  -> log.info("Subscribed"))
    .doOnNext(item      -> log.debug("Received: {}", item))
    .doOnError(ex       -> log.error("Error: {}", ex.getMessage()))
    .doOnComplete(()    -> log.info("Stream completed"))
    .doOnTerminate(()   -> log.info("Terminated (complete or error)"))
    .doOnCancel(()      -> log.warn("Subscription cancelled"))
    .doFinally(signal   -> log.info("Final signal: {}", signal))  // always runs
    .map(n -> n * 2)
    .subscribe(System.out::println);

// doOnNext for metrics
Flux.fromIterable(orders)
    .doOnNext(order -> meterRegistry.counter("orders.processed",
        "status", order.getStatus().name()).increment())
    .flatMap(orderService::process)
    .subscribe();

Time-Based Operators: timeout, delayElements & sample

Reactor integrates with time naturally: you can set deadlines on publishers, introduce artificial delays, and sample high-frequency streams to reduce processing load.

Java

// timeout — error if no item arrives within the duration
Mono.fromCallable(() -> slowRemoteCall())
    .timeout(Duration.ofSeconds(3))
    .onErrorReturn("fallback-value")
    .subscribe(System.out::println);

// timeout with fallback publisher instead of error
Mono.fromCallable(() -> primaryService.fetch())
    .timeout(Duration.ofSeconds(2),
        Mono.fromCallable(() -> fallbackService.fetch()))
    .subscribe(System.out::println);

// delayElements — introduce a pause between items
Flux.just("email-1", "email-2", "email-3")
    .delayElements(Duration.ofMillis(500))  // 500ms between each send
    .flatMap(emailService::send)
    .subscribe();

// delaySubscription — wait before even starting
Flux.range(1, 5)
    .delaySubscription(Duration.ofSeconds(1))  // start 1s from now
    .subscribe(System.out::println);

// sample — emit most recent item in each time window (lossy)
Flux.interval(Duration.ofMillis(100))       // emits every 100ms
    .sample(Duration.ofSeconds(1))           // emit latest value every 1s
    .subscribe(System.out::println);         // 0 items per second sampled to 1

// throttleFirst — emit first item in each window, drop the rest
Flux.interval(Duration.ofMillis(100))
    .sample(Flux.interval(Duration.ofSeconds(1)))
    .subscribe(System.out::println);

Choosing the Right Operator

Goal	Operator	Notes
Transform each item (sync)	`map`	1-to-1, no async work
Async transform, order unimportant	`flatMap`	Runs concurrently
Async transform, order required	`concatMap`	Sequential, slower
Only last async result matters	`switchMap`	Cancels previous
Combine two Mono results	`Mono.zip`	Waits for both
Merge concurrent streams	`Flux.merge`	Interleaved by time
Append streams in order	`Flux.concat`	Sequential, ordered
Batch items for bulk ops	`buffer`	Emits `List<T>`
Running total / state machine	`scan`	Emits each accumulation
Final aggregate	`reduce`	Emits single `Mono`
Deadline on slow source	`timeout`	Errors or falls back

Contents

flatMap vs concatMap vs switchMap

Combining Streams: zip, merge, concat & combineLatest

Batching: buffer & window

groupBy: Splitting by Key

Aggregation: reduce & scan

Side-Effect Operators: doOn*

Time-Based Operators: timeout, delayElements & sample

Choosing the Right Operator