Patrón Back-Pressure

Descripción General

El Patrón Back-Pressure evita que un productor sature a un consumidor lento propagando señales de control de flujo hacia atrás a través del pipeline de datos. Cuando el consumidor no llega al ritmo, le pide al productor que baje la velocidad o pause, evitando que la memoria crezca sin límite, timeouts y fallos en cascada.

Sin back-pressure, productores rápidos y consumidores lentos llevan a:

• Errores de memoria agotada por buffers sin límite
• Picos de latencia a medida que las colas crecen
• Mensajes descartados cuando los buffers se desbordan
• Fallos en cascada cuando los servicios de destino colapsan bajo carga

El back-pressure es fundamental en sistemas reactivos, procesamiento de streams (Kafka, Flink) y frameworks de E/S asíncrona (Node.js streams, ReactiveX).

Cuándo Usar

Usa el Patrón Back-Pressure cuando:

• Productores y consumidores operan a velocidades diferentes de forma sostenida
• Necesitas uso de memoria acotado en pipelines de streaming o código asíncrono
• Los servicios de destino tienen límites duros de capacidad (tasa de escritura de la base de datos, cuotas de API)
• El sistema debe mantenerse estable ante picos de carga impredecibles

Cuándo Evitar

• Todas las etapas procesan datos a ritmos similares (un pipeline síncrono basta)
• La latencia importa más que el throughput (el back-pressure añade retraso)
• Un buffer fijo con política de descarte del más nuevo o el más viejo es aceptable
• El consumidor escala sin límites (auto-scaling serverless) y bufferear es más barato

Solución

Python

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

class BackPressuredQueue:
    """Cola acotada; produce() se bloquea mientras esté llena, frenando al productor"""
    def __init__(self, max_size: int = 10):
        self._queue = queue.Queue(maxsize=max_size)
        self._shutdown = threading.Event()

    def produce(self, item) -> bool:
        """Espera a que haya hueco en vez de fallar: esa espera ES el back-pressure"""
        while not self._shutdown.is_set():
            try:
                self._queue.put(item, block=True, timeout=0.5)
                return True
            except queue.Full:
                continue  # sigue llena: el productor se mantiene frenado
        return False

    def consume(self) -> Optional:
        """Retorna None solo tras el apagado y con la cola ya vacía"""
        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  # el productor puede seguir activo

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

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


class DataPipeline:
    """Pipeline con back-pressure entre productor y consumidor"""
    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  # el pipeline se está apagando
        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()


# Uso
def fast_producer():
    for i in range(20):
        print(f"Produciendo {i}")
        yield f"data-{i}"
        time.sleep(0.1)  # Rápido: 10 items/seg

def slow_consumer(item):
    print(f"Consumiendo {item}")
    time.sleep(0.5)  # Lento: 2 items/seg

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

Java

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

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); // Bloquea cuando lleno — 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();
        }
    }
}

JavaScript

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

// Transform stream con back-pressure en Node.js
const slowTransform = new Transform({
  highWaterMark: 3, // Buffer pequeño para ejercer back-pressure rápidamente
  transform(chunk, encoding, callback) {
    console.log(`Procesando: ${chunk.toString().trim()}`);
    setTimeout(() => {
      callback(null, chunk);
    }, 500);
  }
});

// Productor rápido (stdin o generator)
process.stdin.pipe(slowTransform).pipe(process.stdout);

// O con async generators y back-pressure manual
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('Canal cerrado');
    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;
    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()); // no dejar senders esperando
  }
}

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

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

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

demo().catch(console.error);

Explicación

El back-pressure funciona haciendo que la operación de envío del productor dependa de la capacidad del consumidor:

1. Buffer acotado: La cola entre productor y consumidor tiene un tamaño máximo
2. Send bloqueante: Cuando el buffer está lleno, el productor bloquea o recibe una señal de “ralentizar”
3. Flujo basado en créditos: El consumidor otorga “créditos” (permiso para enviar N items más) al productor
4. Pull reactivo: El consumidor solicita items a su propio ritmo (ej. Reactive Streams request(n))

En Reactive Streams, esto se formaliza con el mecanismo Subscription.request(n). En Kafka, el consumidor controla el ritmo de lectura: max.poll.records acota cada lote de poll(), pause() detiene el fetch de las particiones asignadas mientras un worker procesa el lote, y los offsets se commitean solo cuando el trabajo terminó. El back-pressure también combina bien con el Patrón de Throttling, que limita la tasa del productor de forma proactiva, y con Queue-Based Load Leveling, que usa la cola como amortiguador.

Variantes

Variante	Mecanismo	Mejor Para
Blocking queue	Thread bloquea cuando lleno	Pipelines basados en threads
Reactive Streams	request(n) basado en créditos	Pipelines async componibles
TCP windowing	Protocolo de ventana deslizante	Control de flujo de red
Token bucket	Productor necesita token para enviar	APIs con rate limit
Pause/resume	Consumidor envía señal de pausa	Sistemas de procesamiento batch

Lo que Funciona

• Usa buffers acotados en todas partes. Las colas sin límite son la causa raíz de casi todos los fallos de back-pressure.
• Dimensiona buffers según la latencia p99, no la media. Un buffer ajustado para la carga media se desborda durante los picos.
• Monitoriza la profundidad y la edad del buffer. Alerta según el retardo tolerado por tu carga, no sobre un porcentaje fijo de capacidad.
• Prefiere el pull reactivo sobre el push. Deja que el consumidor marque el ritmo, no el productor.
• Aplica back-pressure en cada capa. Base de datos → servicio → API gateway → cliente.

Errores Comunes

• Colas sin límite. LinkedBlockingQueue sin capacidad consume en silencio toda la memoria disponible.
• Ignorar back-pressure en código asíncrono. Promise.all() con arrays sin límite crea el mismo problema.
• Tragar excepciones de bloqueo. Los timeouts en put() deben reintentar o propagarse, no descartar el elemento en silencio.
• Buffer de talla única. Cada etapa del pipeline necesita su propio tamaño de buffer.
• Asumir que el consumidor siempre es el lento. En un pipeline de más de un salto, cada etapa puede frenar a la anterior.

Ejemplos del Mundo Real

Apache Kafka

El broker de Kafka nunca empuja mensajes al consumidor: es el cliente el que pide. Un consumidor corto de capacidad usa max.poll.records para acotar cada lote, pause() para dejar de hacer fetch de las particiones asignadas mientras un worker las procesa, y commits manuales para que los offsets solo avancen cuando el trabajo terminó — ver la documentación de KafkaConsumer.

gRPC Streaming

gRPC hereda el control de flujo de HTTP/2: el receptor concede capacidad con frames WINDOW_UPDATE, y la llamada de escritura simplemente espera mientras la ventana de recepción está agotada — ver la guía de control de flujo de gRPC.

Node.js Streams

readable.pipe(writable) de Node.js maneja el back-pressure automáticamente: cuando el writable alcanza su highWaterMark, pipe() pausa el readable hasta que se dispara drain — ver Node.js sobre backpressure en streams.

Akka Streams / Project Reactor

Las implementaciones de Reactive Streams usan señalización de demanda explícita: el subscriber llama a request(n) y concede permiso al upstream para enviar n elementos más.