Overview
Knowing which Reactor operator to reach for in a given situation is the practical skill that separates reactive developers. The operator vocabulary is large, but most production code uses a small core set. This page covers the operators that appear in every non-trivial reactive pipeline, with the semantic traps that catch developers (flatMap vs concatMap, switchIfEmpty vs defaultIfEmpty).
Architecture
OPERATOR TAXONOMY
════════════════════════════════════════════════════════
TRANSFORMATION
map(fn) → sync 1:1 transform (non-blocking only)
flatMap(fn) → async 1:N, concurrent, no ordering guarantee
concatMap(fn) → async 1:N, sequential, preserves order
switchMap(fn) → like flatMap but cancels previous inner stream on new item
FILTERING
filter(pred) → keep matching items
take(n) → first N items then complete
skip(n) → skip first N
distinct() → deduplicate
distinctUntilChanged() → deduplicate consecutive identical items
COMBINING
Mono.zip(a, b) → combine 2 Monos when both resolve
Flux.merge(a, b) → combine, emits as they arrive (no order)
Flux.concat(a, b) → sequential: a completes, then b starts
Flux.combineLatest → emit when ANY source emits, include latest from all
ERROR HANDLING
onErrorResume(fn) → switch to fallback publisher on error
onErrorReturn(val) → emit fallback value on error, then complete
onErrorMap(fn) → transform exception type
retry(n) → resubscribe N times on error
retryWhen(spec) → retry with exponential backoff, jitter
SIDE EFFECTS (no value transformation)
doOnNext(fn) → peek at each item (logging, metrics)
doOnError(fn) → peek at error
doOnComplete(fn) → on completion
doFinally(fn) → runs on complete OR error OR cancel
FALLBACKS
defaultIfEmpty(val) → emit val if empty sequence
switchIfEmpty(mono) → switch to different publisher if empty
zipWhen(fn) → zip Mono with result of fn(current value)
FLATMAP vs CONCATMAP
════════════════════════════════════════════════════════
Source: [A, B, C]
fn: item → Mono(item + slow async)
flatMap: fires A, B, C concurrently → [C, A, B] (fastest finishes first)
concatMap: fires A, waits, then B, waits, then C → [A, B, C] (ordered)
flatMap(fn, concurrency=2):
→ fires A and B, waits for one to finish, then fires C
→ bounded concurrency for rate-limited downstream
Technical Implementation
Fan-Out and Fan-In with zip and flatMap
@Service
public class ProductService {
// Fetch product + reviews + inventory in parallel, combine into one response
public Mono<ProductDetail> getProductDetail(String productId) {
Mono<Product> productMono =
productRepo.findById(productId)
.switchIfEmpty(Mono.error(new ProductNotFoundException(productId)));
Mono<List<Review>> reviewsMono =
reviewService.getTopReviews(productId, 5)
.onErrorResume(ex -> {
log.warn("Reviews unavailable for {}", productId, ex);
return Mono.just(Collections.emptyList()); // graceful degrade
});
Mono<Integer> stockMono =
inventoryService.getStock(productId)
.defaultIfEmpty(0);
// All three fire concurrently, combine when all complete
return Mono.zip(productMono, reviewsMono, stockMono)
.map(tuple -> new ProductDetail(
tuple.getT1(),
tuple.getT2(),
tuple.getT3()
));
}
// Process a batch of product IDs with bounded concurrency
public Flux<Product> enrichProducts(Flux<String> productIds) {
return productIds
.flatMap(
id -> fetchAndEnrich(id), // async enrichment per product
10 // max 10 concurrent enrichments
)
.onErrorContinue((ex, item) -> {
log.error("Failed to enrich product {}: {}", item, ex.getMessage());
// continue processing remaining items, skip the failed one
});
}
}
Retry with Exponential Backoff
public Mono<PaymentResult> processPayment(PaymentRequest request) {
return paymentGateway.charge(request)
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
.maxBackoff(Duration.ofSeconds(30))
.jitter(0.3d) // ±30% jitter to avoid thundering herd
.filter(ex -> ex instanceof TransientPaymentException) // only retry transient
.onRetryExhaustedThrow((spec, signal) ->
new PaymentFailedException("Payment failed after retries", signal.failure())
)
)
.onErrorResume(PaymentFailedException.class, ex ->
auditService.logFailedAttempt(request)
.then(Mono.error(ex)) // re-throw after logging
);
}
Windowing for Micro-Batching
// Group incoming events into batches of 100 or every 500ms, whichever comes first
// Useful for: batching DB inserts, debouncing notifications
public Flux<BulkWriteResult> batchProcess(Flux<Event> events) {
return events
.windowTimeout(100, Duration.ofMillis(500)) // batch by count OR time
.flatMap(window ->
window.collectList()
.filter(batch -> !batch.isEmpty())
.flatMap(batch -> dbService.bulkInsert(batch))
);
}
// Buffer with backpressure strategy
public Flux<List<Metric>> bufferMetrics(Flux<Metric> metrics) {
return metrics
.onBackpressureBuffer(1000,
dropped -> log.warn("Metric dropped due to backpressure: {}", dropped)
)
.bufferTimeout(50, Duration.ofSeconds(1));
}
Interview Preparation
Q: When would you use flatMap vs concatMap?
flatMap fires all inner publishers concurrently — maximum throughput, no ordering guarantee. Use when: order doesn't matter and you want to maximize parallelism (enrich 100 items concurrently from an external API). concatMap fires one inner publisher at a time, waits for completion before starting the next — ordered, sequential. Use when: order matters (process audit log entries in order), the downstream has no concurrency support (a serial queue), or you're updating state that depends on previous results. flatMap(fn, concurrency) is the middle ground: bounded parallelism with a concurrency cap, useful for rate-limiting downstream calls.
Q: What is the difference between defaultIfEmpty and switchIfEmpty?
defaultIfEmpty(value) emits a single fallback value if the sequence completed without emitting anything. The value is pre-computed (it's a constant). switchIfEmpty(alternativePublisher) switches to a different publisher (Mono/Flux) when the original completes empty. The alternative publisher is lazy — it only subscribes and executes if the original was empty. Use switchIfEmpty when the fallback requires I/O (e.g., look up in cache, fall back to DB). Use defaultIfEmpty when the fallback is a static value or is already computed.
Q: How do you handle errors in a way that doesn't terminate the entire Flux?
Three techniques: (1) onErrorResume on a specific inner publisher in a flatMap — the outer Flux continues, only that one inner stream failed and was replaced by a fallback. (2) onErrorContinue(handler) — instead of propagating the error downstream, logs and skips the element that caused the error, continuing with the rest. (3) Flux.mergeDelayError — like merge but collects all errors and emits them as a CompositeException at the end rather than failing fast. The choice depends on whether you want to skip errors silently, substitute fallbacks, or collect all errors for a final report.
Learning Resources
- Project Reactor Operators Reference
- Marble diagrams — ReactiveX — visual operator guide
- Reactor — Error Handling
- Reactive Programming with RxJava — Tomasz Nurkiewicz — patterns and idioms