Back-Pressure Pattern

Overview

The Back-Pressure Pattern stops a fast producer from burying a slow consumer: flow-control signals travel backward through the pipeline so the sender waits instead of the receiver drowning. When the consumer can’t keep up, it asks the producer to slow down or pause — which is how you avoid unbounded memory growth, timeouts, and cascading failures.

Without it, pairing a fast producer with a slow consumer ends in:

• Out-of-memory errors from unbounded buffers
• Latency spikes as queues grow
• Dropped messages when buffers overflow
• Cascading failures as downstream services collapse under load

Back-pressure is fundamental in reactive systems, stream processing (Kafka, Flink), and async I/O frameworks (Node.js streams, ReactiveX).

When to Use

Use the Back-Pressure Pattern when:

• Producers and consumers operate at different speeds in a sustained way
• You need bounded memory usage in streaming or async pipelines
• Downstream services have hard capacity limits (database write rates, API quotas)
• The system must remain stable under unpredictable load spikes

When to Avoid

• All stages process data at roughly the same rate (synchronous pipelining suffices)
• Latency is more important than throughput (back-pressure adds delay)
• A simple fixed-size buffer with drop-newest/drop-oldest policy is acceptable
• The consumer is infinitely scalable (serverless auto-scaling) and buffering is cheaper

Solution

Python

import queue
import threading
import time
from typing import Callable, Optional

class BackPressuredQueue:
    """Bounded queue; produce() blocks while full, so producers back off"""
    def __init__(self, max_size: int = 10):
        self._queue = queue.Queue(maxsize=max_size)
        self._shutdown = threading.Event()

    def produce(self, item) -> bool:
        """Waits for room instead of failing: that wait IS the back-pressure"""
        while not self._shutdown.is_set():
            try:
                self._queue.put(item, block=True, timeout=0.5)
                return True
            except queue.Full:
                continue  # still full: keep holding the producer back
        return False

    def consume(self) -> Optional:
        """Returns None only after shutdown AND the queue is drained"""
        while True:
            if self._shutdown.is_set() and self._queue.empty():
                return None
            try:
                return self._queue.get(block=True, timeout=0.5)
            except queue.Empty:
                continue  # the producer may still be running

    def mark_done(self):
        self._queue.task_done()

    def shutdown(self):
        self._shutdown.set()


class DataPipeline:
    """Pipeline with back-pressure between producer and consumer"""
    def __init__(self, buffer_size: int = 5):
        self.buffer = BackPressuredQueue(max_size=buffer_size)
        self.producer_thread: Optional[threading.Thread] = None
        self.consumer_thread: Optional[threading.Thread] = None

    def start(self, producer_fn: Callable, consumer_fn: Callable):
        self.producer_thread = threading.Thread(
            target=self._run_producer, args=(producer_fn,)
        )
        self.consumer_thread = threading.Thread(
            target=self._run_consumer, args=(consumer_fn,)
        )
        self.producer_thread.start()
        self.consumer_thread.start()

    def _run_producer(self, producer_fn):
        try:
            for item in producer_fn():
                if not self.buffer.produce(item):
                    break  # pipeline is shutting down
        finally:
            self.buffer.shutdown()

    def _run_consumer(self, consumer_fn):
        while True:
            item = self.buffer.consume()
            if item is None:
                break
            consumer_fn(item)
            self.buffer.mark_done()

    def join(self):
        self.producer_thread.join()
        self.consumer_thread.join()


# Usage
def fast_producer():
    for i in range(20):
        print(f"Producing {i}")
        yield f"data-{i}"
        time.sleep(0.1)  # Fast: 10 items/sec

def slow_consumer(item):
    print(f"Consuming {item}")
    time.sleep(0.5)  # Slow: 2 items/sec

pipeline = DataPipeline(buffer_size=3)
pipeline.start(fast_producer, slow_consumer)
pipeline.join()

Java

import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class BackPressuredPipeline<T> {
    private final BlockingQueue<T> queue;
    private final ExecutorService executor = Executors.newFixedThreadPool(2);

    public BackPressuredPipeline(int capacity) {
        this.queue = new ArrayBlockingQueue<>(capacity);
    }

    public void run(Supplier<T> producer, Consumer<T> consumer) {
        Future<?> producerTask = executor.submit(() -> {
            while (true) {
                T item = producer.get();
                if (item == null) break;
                try {
                    queue.put(item); // Blocks when full — back-pressure
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });

        Future<?> consumerTask = executor.submit(() -> {
            while (true) {
                try {
                    T item = queue.poll(1, TimeUnit.SECONDS);
                    if (item == null && producerTask.isDone()) break;
                    if (item != null) consumer.accept(item);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });

        try {
            producerTask.get();
            consumerTask.get();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
    }

    // Usage
    public static void main(String[] args) {
        BackPressuredPipeline<String> pipeline = new BackPressuredPipeline<>(5);

        pipeline.run(
            () -> {
                try {
                    Thread.sleep(100);
                    return "item-" + System.currentTimeMillis();
                } catch (InterruptedException e) {
                    return null;
                }
            },
            item -> {
                System.out.println("Consumed: " + item);
                try { Thread.sleep(500); } catch (InterruptedException e) {}
            }
        );
    }
}

JavaScript

const { Transform } = require('stream');

// Back-pressured transform stream in Node.js
const slowTransform = new Transform({
  highWaterMark: 3, // Small buffer to exert back-pressure quickly
  transform(chunk, encoding, callback) {
    console.log(`Processing: ${chunk.toString().trim()}`);
    // Simulate slow processing
    setTimeout(() => {
      callback(null, chunk);
    }, 500);
  }
});

// Fast producer (stdin or generator)
process.stdin.pipe(slowTransform).pipe(process.stdout);

// Or with async generators and manual back-pressure
class BackPressuredChannel {
  constructor(capacity = 5) {
    this.buffer = [];
    this.capacity = capacity;
    this.waitingConsumers = [];
    this.waitingProducers = [];
    this.closed = false;
  }

  async send(value) {
    while (!this.closed && this.buffer.length >= this.capacity) {
      await new Promise(resolve => this.waitingProducers.push(resolve));
    }
    if (this.closed) throw new Error('Channel closed');
    this.buffer.push(value);
    const waiter = this.waitingConsumers.shift();
    if (waiter) waiter();
  }

  async receive() {
    while (this.buffer.length === 0 && !this.closed) {
      await new Promise(resolve => this.waitingConsumers.push(resolve));
    }
    if (this.buffer.length === 0) return undefined; // closed and empty
    const value = this.buffer.shift();
    const waiter = this.waitingProducers.shift();
    if (waiter) waiter();
    return value;
  }

  close() {
    this.closed = true;
    this.waitingConsumers.forEach(w => w());
    this.waitingProducers.forEach(w => w()); // don't leave senders parked
  }
}

// Usage
async function demo() {
  const channel = new BackPressuredChannel(3);

  // Producer
  (async () => {
    for (let i = 0; i < 10; i++) {
      console.log(`Sending ${i}`);
      await channel.send(i);
    }
    channel.close();
  })();

  // Consumer
  for await (const value of (async function* () {
    while (true) {
      const v = await channel.receive();
      if (v === undefined) break;
      yield v;
    }
  })()) {
    console.log(`Received ${value}`);
    await new Promise(r => setTimeout(r, 800));
  }
}

demo().catch(console.error);

Explanation

Back-pressure works by making the producer’s send operation dependent on the consumer’s capacity:

1. Bounded buffer: The queue between producer and consumer has a maximum size
2. Blocking send: When the buffer is full, the producer blocks or receives a “slow down” signal
3. Credit-based flow: The consumer grants “credits” (permission to send N more items) to the producer
4. Reactive pull: The consumer requests items at its own pace (e.g., Reactive Streams request(n))

In Reactive Streams, this is formalized through the Subscription.request(n) mechanism. In Kafka, consumers pace the fetch loop themselves: max.poll.records bounds each poll(), pause() stops fetching assigned partitions while a worker thread processes the batch, and offsets are committed only after processing finishes. Back-pressure also pairs naturally with the Throttling Pattern, which caps producer rate proactively, and with Queue-Based Load Leveling, which uses the queue itself as a shock absorber.

Variants

Variant	Mechanism	Best For
Blocking queue	Thread blocks when full	Thread-based pipelines
Reactive Streams	request(n) credit-based	Composable async pipelines
TCP windowing	Sliding window protocol	Network flow control
Token bucket	Producer needs token to send	Rate-limited APIs
Pause/resume	Consumer sends pause signal	Batch processing systems

What Works

• Use bounded buffers everywhere. Unbounded queues are the root cause of most back-pressure failures.
• Set buffer sizes based on p99 latency, not average. A buffer sized for average load will overflow during spikes.
• Monitor buffer depth and item age. Alert on the queueing delay your workload tolerates, not on a fixed percentage of capacity.
• Prefer reactive pull over push. Let the consumer drive the rate, not the producer.
• Handle back-pressure at every layer. Database → service → API gateway → client.

Common Mistakes

• Unbounded queues. LinkedBlockingQueue without a capacity silently consumes all available memory.
• Ignoring back-pressure in async code. Promise.all() with unbounded arrays creates the same problem.
• Swallowing blocking exceptions. Timeouts on put() should propagate or retry, not silently drop.
• One-size-fits-all buffer. Each pipeline stage deserves its own buffer size, tuned to that stage’s latency.
• Assuming consumers are always slower. In a pipeline with more than one hop, every stage can push back on the one before it.

Real-World Examples

Apache Kafka

Kafka’s broker never pushes messages to a consumer; the client pulls. A consumer bound for capacity uses max.poll.records to bound each batch, pause() to stop fetching assigned partitions while a worker processes them, and manual commits so offsets only advance after work completes — see the KafkaConsumer docs.

gRPC Streaming

gRPC inherits HTTP/2 flow control: the receiver grants capacity with WINDOW_UPDATE frames, and a write call simply waits while the receive window is exhausted — see the gRPC flow-control guide.

Node.js Streams

Node.js readable.pipe(writable) handles back-pressure automatically: when the writable reaches its highWaterMark, pipe() pauses the readable until drain fires — see Node.js on backpressuring in streams.

Akka Streams / Project Reactor

Reactive Streams implementations use explicit demand signaling. A downstream subscriber calls request(n), granting the upstream permission to send n more elements.