Introduction to Backpressure
Backpressure is a crucial concept in reactive programming that deals with the problem of controlling the data flow between a fast producer and a slow consumer. It prevents overwhelming the consumer with more data than it can handle, thus avoiding memory issues and system instability.
Reactive Streams Specification
Core Interfaces
The Reactive Streams specification defines four main interfaces:
// Publisher: Data producer
public interface Publisher<T> {
void subscribe(Subscriber<? super T> subscriber);
}
// Subscriber: Data consumer
public interface Subscriber<T> {
void onSubscribe(Subscription subscription);
void onNext(T item);
void onError(Throwable throwable);
void onComplete();
}
// Subscription: Control flow between Publisher and Subscriber
public interface Subscription {
void request(long n);
void cancel();
}
// Processor: Both Publisher and Subscriber
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}
Backpressure Strategies
1. Pull-Based Backpressure
import org.reactivestreams.*;
import java.util.concurrent.atomic.*;
public class PullBackpressureExample {
static class SimplePublisher implements Publisher<Integer> {
@Override
public void subscribe(Subscriber<? super Integer> subscriber) {
subscriber.onSubscribe(new SimpleSubscription(subscriber));
}
}
static class SimpleSubscription implements Subscription {
private final Subscriber<? super Integer> subscriber;
private final AtomicLong requested = new AtomicLong(0);
private final AtomicBoolean cancelled = new AtomicBoolean(false);
private int currentValue = 0;
public SimpleSubscription(Subscriber<? super Integer> subscriber) {
this.subscriber = subscriber;
}
@Override
public void request(long n) {
if (n <= 0) {
subscriber.onError(new IllegalArgumentException(
"Request must be positive"));
return;
}
// Add the requested amount
long previousRequested = requested.getAndAdd(n);
if (previousRequested == 0) {
// Start producing if we weren't already
produceItems();
}
}
@Override
public void cancel() {
cancelled.set(true);
}
private void produceItems() {
// Use a separate thread to avoid blocking
new Thread(() -> {
while (!cancelled.get() && requested.get() > 0) {
// Produce one item
subscriber.onNext(currentValue++);
// Decrement the requested count
requested.decrementAndGet();
// Simulate some work
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
// If we've produced all requested items and not cancelled, complete
if (!cancelled.get() && requested.get() == 0) {
subscriber.onComplete();
}
}).start();
}
}
static class SlowSubscriber implements Subscriber<Integer> {
private Subscription subscription;
private final String name;
public SlowSubscriber(String name) {
this.name = name;
}
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
System.out.println(name + " subscribed");
// Request first 3 items
subscription.request(3);
}
@Override
public void onNext(Integer item) {
System.out.println(name + " received: " + item);
// Simulate slow processing
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Request one more item after processing
subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
System.out.println(name + " error: " + throwable.getMessage());
}
@Override
public void onComplete() {
System.out.println(name + " completed");
}
}
public static void main(String[] args) throws InterruptedException {
SimplePublisher publisher = new SimplePublisher();
SlowSubscriber subscriber = new SlowSubscriber("SlowConsumer");
publisher.subscribe(subscriber);
// Keep main thread alive
Thread.sleep(5000);
}
}
2. Buffer-Based Backpressure
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class BufferBackpressureExample {
static class BufferedPublisher implements Publisher<String> {
private final ExecutorService executor = Executors.newFixedThreadPool(2);
private final int bufferSize;
public BufferedPublisher(int bufferSize) {
this.bufferSize = bufferSize;
}
@Override
public void subscribe(Subscriber<? super String> subscriber) {
subscriber.onSubscribe(new BufferedSubscription(subscriber, bufferSize, executor));
}
}
static class BufferedSubscription implements Subscription {
private final Subscriber<? super String> subscriber;
private final BlockingQueue<String> buffer;
private final AtomicLong requested = new AtomicLong(0);
private final AtomicBoolean cancelled = new AtomicBoolean(false);
private final ExecutorService executor;
private volatile boolean producing = false;
public BufferedSubscription(Subscriber<? super String> subscriber,
int bufferSize, ExecutorService executor) {
this.subscriber = subscriber;
this.buffer = new LinkedBlockingQueue<>(bufferSize);
this.executor = executor;
}
@Override
public void request(long n) {
if (n <= 0) {
subscriber.onError(new IllegalArgumentException(
"Request must be positive"));
return;
}
requested.getAndAdd(n);
drainBuffer();
}
@Override
public void cancel() {
cancelled.set(true);
buffer.clear();
}
public void startProducing() {
if (producing) return;
producing = true;
executor.submit(() -> {
int count = 0;
while (!cancelled.get()) {
try {
String item = "Item-" + count++;
// Try to add to buffer with timeout
boolean added = buffer.offer(item, 100, TimeUnit.MILLISECONDS);
if (added) {
System.out.println("Produced: " + item +
" (Buffer size: " + buffer.size() + ")");
drainBuffer();
} else {
// Buffer full - apply backpressure
System.out.println("Buffer full! Slowing down production...");
Thread.sleep(500); // Slow down producer
}
// Simulate production time
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
});
}
private void drainBuffer() {
while (!cancelled.get() && requested.get() > 0 && !buffer.isEmpty()) {
String item = buffer.poll();
if (item != null) {
subscriber.onNext(item);
requested.decrementAndGet();
}
}
}
}
static class BufferedSubscriber implements Subscriber<String> {
private Subscription subscription;
private final String name;
private final int processingDelay;
public BufferedSubscriber(String name, int processingDelay) {
this.name = name;
this.processingDelay = processingDelay;
}
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
System.out.println(name + " subscribed");
// Request initial batch
subscription.request(5);
// Start production
if (subscription instanceof BufferedSubscription) {
((BufferedSubscription) subscription).startProducing();
}
}
@Override
public void onNext(String item) {
System.out.println(name + " processing: " + item);
// Simulate variable processing time
try {
Thread.sleep(processingDelay);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Request next item
subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
System.out.println(name + " error: " + throwable.getMessage());
}
@Override
public void onComplete() {
System.out.println(name + " completed");
}
}
public static void main(String[] args) throws InterruptedException {
BufferedPublisher publisher = new BufferedPublisher(10); // Buffer size 10
// Fast producer, slow consumer
BufferedSubscriber slowSubscriber = new BufferedSubscriber("SlowConsumer", 500);
publisher.subscribe(slowSubscriber);
Thread.sleep(10000);
}
}
Project Reactor Backpressure Handling
Reactor Backpressure Operators
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
public class ReactorBackpressureExamples {
public static void main(String[] args) throws InterruptedException {
// Example 1: onBackpressureBuffer - Buffer items when consumer is slow
System.out.println("=== onBackpressureBuffer ===");
Flux.range(1, 1000)
.onBackpressureBuffer(50) // Buffer up to 50 items
.doOnNext(i -> System.out.println("Produced: " + i))
.publishOn(Schedulers.single(), 10) // Consumer with prefetch of 10
.subscribe(i -> {
// Slow consumer
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Consumed: " + i);
});
Thread.sleep(2000);
System.out.println("\n=== onBackpressureDrop ===");
// Example 2: onBackpressureDrop - Drop items when buffer is full
AtomicInteger droppedCount = new AtomicInteger();
Flux.range(1, 100)
.onBackpressureDrop(item -> {
droppedCount.incrementAndGet();
System.out.println("Dropped: " + item);
})
.publishOn(Schedulers.single(), 5) // Small buffer
.subscribe(item -> {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Processed: " + item);
});
Thread.sleep(5000);
System.out.println("Total dropped: " + droppedCount.get());
System.out.println("\n=== onBackpressureLatest ===");
// Example 3: onBackpressureLatest - Keep only the latest item
Flux.interval(Duration.ofMillis(10))
.onBackpressureLatest()
.publishOn(Schedulers.single(), 1)
.subscribe(item -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Latest: " + item);
});
Thread.sleep(2000);
System.out.println("\n=== onBackpressureError ===");
// Example 4: onBackpressureError - Throw error on overflow
Flux.range(1, 1000)
.onBackpressureError()
.publishOn(Schedulers.single(), 5)
.subscribe(
item -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Item: " + item);
},
error -> System.err.println("Error: " + error)
);
Thread.sleep(3000);
}
}
Custom Backpressure Strategies with Reactor
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;
import java.util.concurrent.atomic.*;
public class CustomBackpressureStrategies {
// Strategy 1: Adaptive Request Strategy
public static class AdaptiveBackpressureStrategy {
private final AtomicLong requestSize = new AtomicLong(1);
private final AtomicLong processedCount = new AtomicLong(0);
private final long windowSize;
private final long maxRequestSize;
public AdaptiveBackpressureStrategy(long windowSize, long maxRequestSize) {
this.windowSize = windowSize;
this.maxRequestSize = maxRequestSize;
}
public Flux<Integer> createAdaptiveStream() {
return Flux.create(sink -> {
AtomicInteger counter = new AtomicInteger(0);
AtomicLong currentRequested = new AtomicLong(0);
sink.onRequest(n -> {
currentRequested.addAndGet(n);
produceItems(sink, counter, currentRequested);
});
// Initial request
sink.request(requestSize.get());
});
}
private void produceItems(FluxSink<Integer> sink, AtomicInteger counter,
AtomicLong currentRequested) {
while (currentRequested.get() > 0 && !sink.isCancelled()) {
int item = counter.incrementAndGet();
sink.next(item);
currentRequested.decrementAndGet();
processedCount.incrementAndGet();
// Adaptive logic: increase request size if processing is fast
if (processedCount.get() % windowSize == 0) {
long currentSize = requestSize.get();
if (currentSize < maxRequestSize) {
requestSize.set(Math.min(currentSize * 2, maxRequestSize));
System.out.println("Increased request size to: " + requestSize.get());
}
}
// Simulate production time
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
sink.error(e);
return;
}
}
}
}
// Strategy 2: Time-Based Backpressure
public static class TimeBasedBackpressureStrategy {
public static Flux<String> createTimeBasedStream(Duration productionInterval,
Duration consumptionRate) {
return Flux.interval(productionInterval)
.onBackpressureBuffer(100,
drop -> System.out.println("Buffer full, dropping: " + drop))
.map(i -> "Item-" + i)
.publishOn(Schedulers.parallel())
.doOnNext(item -> {
// Simulate processing time based on consumption rate
try {
Thread.sleep(consumptionRate.toMillis());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}
// Strategy 3: Load-Based Backpressure
public static class LoadBasedBackpressureStrategy {
private final double maxCpuThreshold;
private final long maxMemoryThreshold;
public LoadBasedBackpressureStrategy(double maxCpuThreshold, long maxMemoryThreshold) {
this.maxCpuThreshold = maxCpuThreshold;
this.maxMemoryThreshold = maxMemoryThreshold;
}
public Flux<Integer> createLoadAwareStream() {
return Flux.generate(
() -> 0, // Initial state
(state, sink) -> {
// Check system load
if (isSystemOverloaded()) {
System.out.println("System overloaded, applying backpressure");
try {
Thread.sleep(1000); // Slow down production
} catch (InterruptedException e) {
sink.error(e);
return state;
}
}
sink.next(state);
return state + 1;
});
}
private boolean isSystemOverloaded() {
Runtime runtime = Runtime.getRuntime();
long usedMemory = runtime.totalMemory() - runtime.freeMemory();
long maxMemory = runtime.maxMemory();
double memoryUsage = (double) usedMemory / maxMemory;
// Simple simulation - in real system, you'd use proper metrics
return memoryUsage > 0.8 || Math.random() < 0.1; // 10% chance of simulated overload
}
}
public static void main(String[] args) throws InterruptedException {
System.out.println("=== Adaptive Backpressure ===");
AdaptiveBackpressureStrategy adaptiveStrategy =
new AdaptiveBackpressureStrategy(10, 20);
adaptiveStrategy.createAdaptiveStream()
.take(50)
.subscribe(item -> {
// Variable processing time
try {
Thread.sleep(50 + (long)(Math.random() * 100));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Processed: " + item);
});
Thread.sleep(5000);
System.out.println("\n=== Time-Based Backpressure ===");
TimeBasedBackpressureStrategy.createTimeBasedStream(
Duration.ofMillis(50), Duration.ofMillis(200))
.take(20)
.subscribe(System.out::println);
Thread.sleep(5000);
System.out.println("\n=== Load-Based Backpressure ===");
LoadBasedBackpressureStrategy loadStrategy =
new LoadBasedBackpressureStrategy(0.8, 1000000000);
loadStrategy.createLoadAwareStream()
.take(30)
.subscribe(item -> System.out.println("Load-aware item: " + item));
Thread.sleep(5000);
}
}
RxJava Backpressure Handling
RxJava Backpressure Operators
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.schedulers.Schedulers;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class RxJavaBackpressureExamples {
public static void main(String[] args) throws InterruptedException {
// Example 1: Flowable with backpressure support
System.out.println("=== Flowable with Backpressure ===");
Flowable.range(1, 1000)
.onBackpressureBuffer(50, // buffer size
() -> System.out.println("Buffer overflow!"),
BackpressureOverflowStrategy.ERROR)
.observeOn(Schedulers.io(), false, 10) // prefetch 10
.subscribe(
item -> {
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Processed: " + item);
},
error -> System.err.println("Error: " + error)
);
TimeUnit.SECONDS.sleep(3);
System.out.println("\n=== onBackpressureDrop ===");
// Example 2: Drop strategy
AtomicInteger dropCount = new AtomicInteger();
Flowable.interval(10, TimeUnit.MILLISECONDS)
.onBackpressureDrop(item -> {
dropCount.incrementAndGet();
System.out.println("Dropped: " + item);
})
.observeOn(Schedulers.computation(), false, 5)
.subscribe(
item -> {
try {
TimeUnit.MILLISECONDS.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Received: " + item);
}
);
TimeUnit.SECONDS.sleep(3);
System.out.println("Total dropped: " + dropCount.get());
System.out.println("\n=== onBackpressureLatest ===");
// Example 3: Latest strategy
Flowable.interval(10, TimeUnit.MILLISECONDS)
.onBackpressureLatest()
.observeOn(Schedulers.computation(), false, 1)
.subscribe(
item -> {
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Latest: " + item);
}
);
TimeUnit.SECONDS.sleep(2);
}
}
Real-World Backpressure Scenarios
1. Database Query Results
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import java.sql.*;
import java.util.concurrent.atomic.AtomicLong;
public class DatabaseBackpressureExample {
static class DatabaseService {
public Flux<String> streamLargeQuery(String query, int fetchSize) {
return Flux.create(sink -> {
Connection conn = null;
Statement stmt = null;
ResultSet rs = null;
try {
conn = DriverManager.getConnection("jdbc:h2:mem:test");
stmt = conn.createStatement();
stmt.setFetchSize(fetchSize);
rs = stmt.executeQuery(query);
AtomicLong requested = new AtomicLong(0);
sink.onRequest(n -> {
requested.addAndGet(n);
try {
while (requested.get() > 0 && rs.next() && !sink.isCancelled()) {
String row = rs.getString(1); // Simplified
sink.next(row);
requested.decrementAndGet();
}
if (!rs.next()) {
sink.complete();
}
} catch (SQLException e) {
sink.error(e);
}
});
} catch (SQLException e) {
sink.error(e);
} finally {
// Cleanup would go here
}
});
}
}
public static void main(String[] args) {
DatabaseService service = new DatabaseService();
service.streamLargeQuery("SELECT * FROM large_table", 100)
.onBackpressureBuffer(1000) // Buffer up to 1000 rows
.publishOn(Schedulers.boundedElastic())
.subscribe(
row -> {
// Process each row
try {
Thread.sleep(10); // Simulate processing time
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Processed: " + row);
},
error -> System.err.println("Error: " + error),
() -> System.out.println("Query processing completed")
);
}
}
2. WebSocket Message Processing
import reactor.core.publisher.*;
import org.springframework.web.reactive.socket.*;
import java.util.concurrent.*;
public class WebSocketBackpressureExample {
static class WebSocketHandler {
private final Flux<String> messageFlux;
private final FluxSink<String> messageSink;
private final AtomicLong pendingMessages = new AtomicLong(0);
private final int maxBufferSize;
public WebSocketHandler(int maxBufferSize) {
this.maxBufferSize = maxBufferSize;
Flux<String> flux = Flux.create(sink -> {
this.messageSink = sink;
sink.onRequest(n -> {
long pending = pendingMessages.addAndGet(-n);
System.out.println("Consumer requested " + n + " messages. Pending: " + pending);
});
});
this.messageFlux = flux.onBackpressureBuffer(maxBufferSize,
item -> System.out.println("Dropping message due to backpressure: " + item));
}
public void handleMessage(String message) {
long pending = pendingMessages.get();
if (pending >= maxBufferSize) {
System.out.println("Backpressure: Rejecting message - " + message);
return;
}
pendingMessages.incrementAndGet();
messageSink.next(message);
}
public Flux<String> getMessageStream() {
return messageFlux;
}
}
public static void main(String[] args) throws InterruptedException {
WebSocketHandler handler = new WebSocketHandler(100);
// Start processing messages
handler.getMessageStream()
.publishOn(Schedulers.parallel(), 10) // Prefetch 10
.subscribe(message -> {
// Simulate message processing
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Processed: " + message);
});
// Simulate incoming messages
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
executor.scheduleAtFixedRate(() -> {
String message = "Msg-" + System.currentTimeMillis();
handler.handleMessage(message);
}, 0, 50, TimeUnit.MILLISECONDS); // Fast producer
Thread.sleep(5000);
executor.shutdown();
}
}
Monitoring and Metrics for Backpressure
import reactor.core.publisher.*;
import java.util.concurrent.atomic.*;
import java.time.Duration;
public class BackpressureMonitoring {
static class MonitoredFlux<T> {
private final Flux<T> flux;
private final AtomicLong producedCount = new AtomicLong();
private final AtomicLong consumedCount = new AtomicLong();
private final AtomicLong droppedCount = new AtomicLong();
private final AtomicLong bufferSize = new AtomicLong();
public MonitoredFlux(Flux<T> source) {
this.flux = source
.doOnNext(item -> producedCount.incrementAndGet())
.onBackpressureBuffer(100,
item -> droppedCount.incrementAndGet())
.doOnNext(item -> {
consumedCount.incrementAndGet();
bufferSize.decrementAndGet();
})
.doOnRequest(n -> bufferSize.addAndGet(n));
}
public Flux<T> getFlux() {
return flux;
}
public void printStats() {
System.out.printf(
"Produced: %d, Consumed: %d, Dropped: %d, Buffer: %d, Lag: %d%n",
producedCount.get(), consumedCount.get(), droppedCount.get(),
bufferSize.get(), producedCount.get() - consumedCount.get()
);
}
}
public static void main(String[] args) throws InterruptedException {
MonitoredFlux<Integer> monitoredFlux = new MonitoredFlux<>(
Flux.interval(Duration.ofMillis(10))
.map(Long::intValue)
.take(1000)
);
// Start statistics reporting
Flux.interval(Duration.ofSeconds(1))
.subscribe(tick -> monitoredFlux.printStats());
// Slow consumer
monitoredFlux.getFlux()
.publishOn(Schedulers.single(), 5)
.subscribe(item -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread.sleep(10000);
}
}
Best Practices for Backpressure Handling
1. Choose the Right Strategy
public class BackpressureStrategyGuide {
public enum BackpressureScenario {
REAL_TIME_DATA, // Use DROP or LATEST
BATCH_PROCESSING, // Use BUFFER with large size
CRITICAL_DATA, // Use BUFFER with ERROR strategy
RESOURCE_CONSTRAINED // Use adaptive backpressure
}
public static <T> Flux<T> applyStrategy(Flux<T> source,
BackpressureScenario scenario) {
switch (scenario) {
case REAL_TIME_DATA:
return source.onBackpressureLatest();
case BATCH_PROCESSING:
return source.onBackpressureBuffer(10_000);
case CRITICAL_DATA:
return source.onBackpressureBuffer(1000,
BufferOverflowStrategy.ERROR);
case RESOURCE_CONSTRAINED:
return source.onBackpressureBuffer(100,
item -> System.out.println("Dropping under pressure: " + item));
default:
return source.onBackpressureBuffer();
}
}
}
2. Proper Resource Cleanup
import reactor.core.publisher.*;
import java.util.concurrent.*;
public class ResourceCleanupExample {
public static Flux<String> createResourceBoundStream() {
return Flux.using(
// Resource supplier
() -> {
System.out.println("Acquiring resource...");
return new ExpensiveResource();
},
// Stream generator
resource -> Flux.interval(Duration.ofMillis(50))
.map(i -> resource.getData(i))
.onBackpressureBuffer(100)
.doOnCancel(() -> System.out.println("Stream cancelled")),
// Resource cleanup
resource -> {
System.out.println("Cleaning up resource...");
resource.cleanup();
}
);
}
static class ExpensiveResource {
private final AtomicBoolean closed = new AtomicBoolean(false);
public String getData(long index) {
if (closed.get()) {
throw new IllegalStateException("Resource closed");
}
return "Data-" + index;
}
public void cleanup() {
closed.set(true);
System.out.println("Resource cleaned up");
}
}
public static void main(String[] args) throws InterruptedException {
createResourceBoundStream()
.take(10) // Automatically cleans up after 10 items
.subscribe(System.out::println);
Thread.sleep(1000);
}
}
Conclusion
Backpressure handling is essential for building robust reactive systems. Key takeaways:
- Understand Your Use Case: Choose the right backpressure strategy based on data criticality and system constraints
- Monitor and Measure: Implement monitoring to understand backpressure patterns
- Use Framework Support: Leverage built-in backpressure operators in Reactor/RxJava
- Consider Resource Limits: Always bound buffers and manage resources properly
- Test Under Load: Ensure your backpressure strategies work under production-like conditions
Proper backpressure handling ensures that your reactive applications remain responsive and stable even under heavy load conditions.