Reactive Circuit Breaker — Resilience Patterns for Reactive Streams and Project Reactor

In this tutorial, you will learn about Reactive Circuit Breaker. We cover key concepts, practical examples, and best practices to help you master this topic.

Reactive circuit breaker patterns integrate with reactive streams (Project Reactor, RxJava) using non-blocking operators that honor backpressure, provide reactive fallback chains, and maintain circuit breaker state without blocking reactive pipelines.

flowchart LR
    Flux[Reactive Stream] --> CB[CircuitBreaker Operator]
    CB -->|Closed| Call[flatMap service.call()]
    CB -->|Open| Fallback[fallbackReturn/cache]
    CB -->|Half-Open| Sample[sample() probe]
    Call -->|onError| Count[Increment Failure]
    Count -->|Threshold| Open[Open State]

What You'll Learn

• Reactor CircuitBreaker operators
• RxJava circuit breaker integration
• Backpressure-aware state management
• Reactive fallback chains
• Metrics from reactive streams

Why It Matters

Blocking circuit breakers in reactive pipelines defeat the purpose of Reactive Programming. Reactive-native circuit breakers integrate as operators (flatMap, transform), honor backpressure during half-open probes, and provide non-blocking fallbacks that compose with other reactive operators.

Real-World Use

DodaTech's reactive Java services use Resilience4j's Reactor operators. Each service has circuit breakers composed into the reactive pipeline: .transformDeferred(CircuitBreakerOperator.of(cb)). This adds circuit breaking without blocking any threads.

Reactor Circuit Breaker

import io.github.resilience4j.circuitbreaker.*;
import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;

public class ReactiveCircuitBreakerExample {
    public static void main(String[] args) {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .slidingWindowSize(10)
            .failureRateThreshold(50)
            .waitDurationInOpenState(Duration.ofSeconds(30))
            .permittedNumberOfCallsInHalfOpenState(3)
            .build();

        CircuitBreaker circuitBreaker = CircuitBreaker
            .of("product-service", config);

        circuitBreaker.getEventPublisher()
            .onStateTransition(event ->
                System.out.println("State: " + event.getStateTransition()));

        Flux.range(1, 20)
            .flatMap(i -> callProductService(i)
                .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
                .onErrorResume(e -> {
                    System.out.println("Fallback for " + i + ": " + e.getMessage());
                    return Mono.just("Cached product " + i);
                })
            )
            .subscribe(result -> System.out.println("Result: " + result));

        try { Thread.sleep(5000); } catch (InterruptedException e) {}
    }

    static Mono<String> callProductService(int id) {
        return Mono.fromSupplier(() -> {
            if (Math.random() < 0.6) throw new RuntimeException("Service error");
            return "Product " + id;
        });
    }
}

Expected output:

State: CLOSED_TO_OPEN
Result: Product 1
Result: Product 3
Fallback for 5: CircuitBreaker 'product-service' is open
Fallback for 6: CircuitBreaker 'product-service' is open
State: OPEN_TO_HALF_OPEN
Result: Product 8
State: HALF_OPEN_TO_CLOSED

RxJava Integration

import io.github.resilience4j.circuitbreaker.*;
import io.github.resilience4j.rxjava3.circuitbreaker.operator.CircuitBreakerOperator;
import io.reactivex.rxjava3.core.*;
import java.time.Duration;

public class RxJavaCircuitBreakerExample {
    public static void main(String[] args) {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .slidingWindowSize(20)
            .failureRateThreshold(50)
            .build();

        CircuitBreaker circuitBreaker = CircuitBreaker
            .of("search-service", config);

        Observable.range(1, 15)
            .flatMapSingle(i -> searchProducts("query-" + i)
                .compose(CircuitBreakerOperator.of(circuitBreaker))
                .onErrorReturn(e -> "Cached results for " + i)
            )
            .subscribe(
                result -> System.out.println("Search: " + result),
                error -> System.out.println("Error: " + error)
            );

        try { Thread.sleep(3000); } catch (InterruptedException e) {}
    }

    static Single<String> searchProducts(String query) {
        return Single.fromCallable(() -> {
            if (Math.random() < 0.4) throw new RuntimeException("Search failed");
            return "Results for " + query;
        });
    }
}

Expected output:

Search: Results for query-1
Search: Results for query-3
Search: Cached results for 5
Search: Cached results for 7

Reactive Fallback Chain

import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.time.Duration;
import java.util.function.Function;

public class ReactiveFallbackChain {
    public static void main(String[] args) {
        Mono<String> result = callPrimaryService(42)
            .transform(applyCircuitBreaker())
            .onErrorResume(e -> callCacheService(42))
            .onErrorResume(e -> callStaleCache(42))
            .onErrorReturn("Default product");

        result.subscribe(
            r -> System.out.println("Result: " + r),
            e -> System.out.println("All fallbacks failed: " + e)
        );

        try { Thread.sleep(2000); } catch (InterruptedException e) {}
    }

    static Function<Mono<String>, Mono<String>> applyCircuitBreaker() {
        CircuitBreaker cb = CircuitBreaker.ofDefaults("catalog");
        return mono -> mono.transformDeferred(CircuitBreakerOperator.of(cb));
    }

    static Mono<String> callPrimaryService(int id) {
        return Mono.error(new RuntimeException("Primary down"));
    }

    static Mono<String> callCacheService(int id) {
        return Mono.just("Cached product " + id);
    }

    static Mono<String> callStaleCache(int id) {
        return Mono.just("Stale product " + id);
    }
}

Expected output:

Result: Cached product 42

Common Mistakes

• Blocking inside reactive operators -- calling Thread.sleep() or synchronous blocking APIs inside flatMap defeats reactive benefits. Use reactive timeouts (Mono.timeout) and non-blocking fallbacks.
• Circuit breaker per subscriber instead of per service -- creating a new CircuitBreaker for every subscription resets state. Create one breaker per downstream service and share across all calls.
• Not composing with retry -- reactive retry (Retry.withBackoff) should be composed before the circuit breaker. Retry transient failures first, then circuit break for persistent failures.
• Ignoring backpressure in half-open -- half-open probes should respect downstream demand. Use .limitRate() on probes to avoid overwhelming the recovering service with requests.
• No timeout on reactive circuit breakers -- circuit breaker operators do not add timeouts. Always compose with Mono.timeout() or Flux.timeout() to prevent reactive pipelines from hanging indefinitely.

Practice Questions

1. How does CircuitBreakerOperator integrate with Reactor's transformDeferred?
2. Why should retry be composed before the circuit breaker in reactive pipelines?
3. How do reactive circuit breakers handle backpressure?
4. What is the advantage of reactive-native circuit breakers over wrapping sync ones?
5. How do you handle fallbacks in reactive circuit breaker pipelines?

Challenge

Build a reactive resilience pipeline: (1) Mono/Flux operators for circuit breaker, retry (exponential backoff), timeout, and bulkhead, (2) compose operators in correct order: timeout > retry > circuit breaker > bulkhead, (3) reactive fallback chain: try primary > cache > stale cache > default, (4) backpressure-aware half-open probes limited to 1 request at a time, (5) Micrometer metrics captured from the reactive pipeline without blocking, (6) testing with StepVerifier to verify circuit breaker transitions in reactive streams.

FAQ

Mini Project

Build a reactive microservice with full resilience: (1) Spring WebFlux REST endpoints with circuit breakers per downstream service, (2) reactive retry with exponential backoff (3 attempts, jitter), (3) reactive timeout (5s) before circuit breaker, (4) reactive fallback that serves cached data from Redis, (5) all resilience operators composed in the correct order, (6) Micrometer + Prometheus metrics for circuit breaker states and call outcomes, (7) StepVerifier integration tests for each circuit breaker scenario, (8) Chaos Monkey that injects random failures to verify reactive resilience.

What's Next

Continue with Database Connection to learn database circuit breaker patterns. Then explore API Gateway for gateway-level circuit breaking.