Nivelación de Carga con Colas: Suaviza Picos de Tráfico

Descripción General

Queue-Based Load Leveling mete una cola de mensajes entre productores y consumidores de trabajo. En vez de que los productores llamen a los consumidores directamente, los productores ponen las tareas en una cola y los consumidores las extraen a un ritmo constante.

Este desacoplamiento convierte cargas de trabajo impredecibles y con ráfagas en un flujo suave. La cola actúa como amortiguador: cuando el tráfico pica, los mensajes se acumulan en lugar de colapsar al consumidor. Cuando baja el tráfico, la cola se vacía y el sistema puede reducir recursos. Vi este patrón salvar una API de pagos durante Black Friday: la cola absorbió 20,000 solicitudes de pago en 30 segundos mientras el consumidor las procesaba a un ritmo constante de 200/s, sin timeouts, sin caídas, sin transacciones perdidas.

Vas a encontrar este patrón detrás de procesadores de trabajos en segundo plano, microservicios orientados a eventos y sistemas de triggers serverless. Si estás usando throttling para rechazar tráfico excesivo, la nivelación de carga es el patrón complementario que acepta el tráfico y lo bufferea.

Cuándo Usarlo

Usalo cuando los productores generen trabajo más rápido de lo que los consumidores pueden procesar durante picos, o cuando los servicios aguas abajo tengan límites de tasa o restricciones de capacidad. También encaja cuando el trabajo puede diferirse, cuando necesitás que productores y consumidores sean independientes, cuando el tráfico es muy variable y cuando construís arquitecturas serverless o auto-escalables que ajustan capacidad según la profundidad de la cola. Si tus productores y consumidores comparten un mecanismo de back-pressure, la nivelación de carga se mete entre ellos como el buffer que absorbe las ráfagas que el back-pressure rechazaría.

Cuándo Evitarlo

Evitalo cuando el usuario espere una respuesta sincrónica, porque encolar agrega latencia. No lo uses si la cola podría crecer sin límite y desbordarse, o si el orden de los mensajes es crítico y la cola no puede garantizar FIFO. Tampoco sirve cuando el costo de serialización de la cola supera las llamadas directas, o cuando incluso un milisegundo de latencia de cola es demasiado. Vi equipos que le pusieron una cola a una API de chat en tiempo real y se preguntaban por qué los mensajes llegaban 200ms tarde; la cola agregó latencia que el caso de uso no toleraba. Si necesitás respuestas sub-10ms, no encoles.

Solución

Python (Celery con Redis)

from celery import Celery
import time

app = Celery('tasks')
app.conf.update(
    broker_url='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/0',
    worker_prefetch_multiplier=1,
    task_acks_late=True,
    task_default_rate_limit='100/m',
)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_image(self, image_url, filters):
    try:
        print(f"Processing {image_url} with filters: {filters}")
        time.sleep(2)
        call_external_api(image_url)
        return {"status": "success", "url": image_url}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

@app.task(rate_limit='10/m')
def generate_report(report_type, date_range):
    print(f"Generating {report_type} report for {date_range}")
    time.sleep(5)
    return {"report_id": f"{report_type}-{date_range}", "status": "completed"}

def call_external_api(image_url):
    pass

class ImageUploadService:
    def handle_upload(self, image_urls, filters):
        task_ids = []
        for url in image_urls:
            result = process_image.delay(url, filters)
            task_ids.append(result.id)

        return {
            "message": f"Queued {len(image_urls)} images for processing",
            "task_ids": task_ids,
        }

Java (Spring con RabbitMQ)

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.*;

@Configuration
class QueueConfig {

    @Bean
    Queue taskQueue() {
        return QueueBuilder.durable("task-queue")
            .withArgument("x-max-length", 10000)
            .withArgument("x-overflow", "reject-publish")
            .withArgument("x-message-ttl", 3600000)
            .build();
    }

    @Bean
    DirectExchange exchange() {
        return new DirectExchange("task-exchange");
    }

    @Bean
    Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("task.routing.key");
    }
}

@RestController
class TaskController {

    private final RabbitTemplate rabbitTemplate;

    public TaskController(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @PostMapping("/tasks")
    public String enqueueTask(@RequestBody TaskRequest request) {
        rabbitTemplate.convertAndSend(
            "task-exchange",
            "task.routing.key",
            request
        );
        return "Task queued successfully";
    }
}

@Service
class TaskConsumer {

    @RabbitListener(queues = "task-queue",
                    concurrency = "4-8",
                    containerFactory = "rabbitListenerContainerFactory")
    public void processTask(TaskRequest task) {
        System.out.println("Processing task: " + task.getId());

        try {
            process(task);
        } catch (Exception e) {
            throw new AmqpRejectAndDontRequeueException("Failed: " + e.getMessage());
        }
    }

    private void process(TaskRequest task) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

class TaskRequest {
    private String id;
    private String type;
    private Object payload;
}

JavaScript (BullMQ con Redis)

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis({ maxRetriesPerRequest: null });

const taskQueue = new Queue('tasks', { connection });

const worker = new Worker('tasks', async (job) => {
    console.log(`Processing job ${job.id}: ${job.name}`);

    switch (job.name) {
        case 'send-email':
            return await sendEmail(job.data);
        case 'process-payment':
            return await processPayment(job.data);
        case 'generate-report':
            return await generateReport(job.data);
        default:
            throw new Error(`Unknown job type: ${job.name}`);
    }
}, {
    connection,
    concurrency: 5,
    limiter: {
        max: 50,
        duration: 60000,
    },
});

worker.on('failed', (job, err) => {
    console.error(`Job ${job.id} failed:`, err.message);
});

class TaskProducer {
    async enqueueEmail(emailData) {
        return await taskQueue.add('send-email', emailData, {
            priority: 2,
            attempts: 3,
            backoff: { type: 'exponential', delay: 2000 },
            removeOnComplete: true,
        });
    }

    async enqueuePayment(paymentData) {
        return await taskQueue.add('process-payment', paymentData, {
            priority: 1,
            attempts: 5,
            backoff: { type: 'fixed', delay: 5000 },
        });
    }

    async enqueueReport(reportData) {
        return await taskQueue.add('generate-report', reportData, {
            priority: 3,
            delay: 60000,
            attempts: 2,
        });
    }

    async getQueueStatus() {
        const waiting = await taskQueue.getWaitingCount();
        const active = await taskQueue.getActiveCount();
        const completed = await taskQueue.getCompletedCount();
        const failed = await taskQueue.getFailedCount();

        return { waiting, active, completed, failed };
    }
}

process.on('SIGTERM', async () => {
    await worker.close();
    await taskQueue.close();
    await connection.quit();
});

module.exports = { TaskProducer, taskQueue };

Explicación

La cola se mete entre productores y consumidores y absorbe las ráfagas. Si llegan 10,000 solicitudes en un segundo, los consumidores siguen extrayendo a su ritmo configurado mientras el exceso espera. Cuando la profundidad de la cola cruza un umbral, el auto-escalamiento levanta más consumidores; cuando la cola se vacía, se reduce.

Los productores no esperan; encolan y retornan inmediatamente. Si un consumidor se cae, los mensajes quedan en la cola y el siguiente consumidor los toma. Esa es la ganancia de resiliencia: un consumidor muerto no pierde trabajo.

El trade-off central: aceptás un pequeño delay predecible a cambio de throughput estable y recuperación de crashes. Sin cola, un pico o sobrecarga al consumidor o se rechaza en el borde. Con cola, el pico se bufferiza y se drena con el tiempo. Si ese delay es aceptable depende de tu presupuesto de latencia.

Un detalle que la gente se confunde: la profundidad de la cola es una señal rezagada. Para cuando notás que sube y provisionás más consumidores, el backlog ya es profundo. Poné el umbral de auto-escalado bajo: escalá al 40-50% de capacidad, no al 80%, así tenés margen antes de que pegue la próxima ráfaga.

Variantes

Para comunicación de un solo proceso con baja latencia, usá colas en memoria como BlockingQueue o canales. RabbitMQ o ActiveMQ son brokers de mensajes para sistemas distribuidos con entrega garantizada. SQS, Azure Queue y Pub/Sub son colas en la nube ideales para serverless e infraestructura gestionada. Kafka o Kinesis son streams que soportan event sourcing y reproducción. Celery, BullMQ y Hangfire son colas de tareas: agregan programación de trabajos, reintentos y seguimiento de resultados. Usé todos estos en distintos momentos: Redis para un prototipo rápido, RabbitMQ cuando necesitaba enrutamiento complejo, y SQS cuando no quería gestionar infraestructura. La elección correcta depende de si necesitás orden, reproducción o solo fire-and-forget.

Mejores Prácticas

Poné un límite a la profundidad de la cola. Las colas sin límite esconden problemas y consumen memoria, así que definí una longitud máxima y un comportamiento de desbordamiento como rechazar, dead-letter o descartar. Monitoreá la profundidad continuamente; una cola en crecimiento es la señal más clara de que necesitás más consumidores. Usá dead-letter queues para aislar mensajes fallidos en lugar de bloquear la línea. Implementá backpressure para que, cuando la cola esté llena, el upstream reciba un 503 Service Unavailable y pueda reducir la carga. Establecé un TTL para que el trabajo viejo expire en lugar de procesarse. Una vez debugueé una cola que creció a 2 millones de mensajes durante la noche porque nadie había seteado un TTL; trabajos de reporte viejos de las 3am seguían procesándose a las 9am, bloqueando trabajo nuevo. Un TTL de 1 hora los habría descartado y mantenido la cola sana.

Errores Comunes

Las colas sin límite eventualmente agotan la memoria y caen al broker. Un solo mensaje envenenado puede bloquear la cola si no lo movés a una dead-letter queue. Asumir FIFO sin verificar rompe las garantías de orden. Ignorar las alarmas de profundidad convierte al backlog en una caída. Encolar sincrónicamente desde los productores mata el beneficio del desacoplamiento. Otro que vi: equipos que setean la cola pero se olvidan de monitorear la salud del consumidor. La cola parece bien (profundidad cerca de cero) porque el consumidor murió y nada se está desencolando, pero todo el trabajo está atascado. Monitoreá tanto la profundidad de la cola como el throughput del consumidor; una cola plana con cero procesados es una bandera roja.

Resumen

El patrón Queue-Based Load Leveling desacopla productores de consumidores con una cola de mensajes que absorbe picos de tráfico. Los productores encolan y siguen; los consumidores extraen trabajo a un ritmo constante. La cola actúa como amortiguador: cuando el tráfico pica, los mensajes se bufferizan en lugar de colapsar al consumidor; cuando baja, la cola se vacía y el sistema escala hacia abajo. Usalo cuando tengas cargas con ráfagas, servicios aguas abajo con rate limit, o infraestructura auto-escalable que reacciona a la profundidad de la cola. Poné límites de profundidad, usá una dead-letter queue para fallos, monitoreá la profundidad como indicador rezagado, y escalá al 40-50% de capacidad, no al 80%. Saltealo para respuestas sincrónicas o cuando incluso un milisegundo de latencia es demasiado.

See Also

La documentación de Celery cubre colas de tareas, reintentos y rate limiting en Python. La guía de confiabilidad de RabbitMQ explica colas durables, dead-letter exchanges y configuración de TTL. Para JavaScript, los docs de BullMQ cubren colas con prioridad, estrategias de backoff y control de concurrencia. Si necesitás priorizar ciertos mensajes sobre otros, el patrón priority queue se construye sobre la nivelación de carga. Para procesamiento ordenado donde la secuencia importa más que el throughput, el patrón sequential convoy garantiza el orden de los mensajes.