Cuando una cola de RabbitMQ posee múltiples consumidores, el broker distribuye los mensajes entrantes entre ellos. Cada mensaje es entregado a un único consumidor inscrito, lo cual facilita la escalabilidad horizontal: basta con añadir más instancias consumidoras para manejar un mayor volumen de trabajo.

Por defecto, RabbitMQ emplea un algorimto round-robin para la distribución, sin considerar si el consumidor ha confirmado previamente los mensajes. Este enfoque puede ser ineficiente: si algunos consumidores procesan mensajes más lentamente que otros, se pueden producir cuellos de botella en los procesadores rápidos mientras otros permanecen ociosos, reduciendo el rendimiento genarel del sistema.

Para mitigar este problema, se puede utilizar el método channel.basicQos(int prefetchCount), el cual limita el número máximo de mensajes sin confirmar que un consumidor puede tener en su canal. Por ejemplo, si un consumidor establece prefetchCount(5), el broker mantendrá un contador. Por cada mensaje enviado, el contador se incrementa; por cada confirmación recibida, se decrementa. Al alcanzar el límite, el broker suspende el envío hasta que se reciba una confirmación. Este mecanismo es análogo a la ventana deslizante en TCP/IP.

Configuración Inicial

spring:
  application:
    name: demo-rabbitmq-avanzado
  rabbitmq:
    addresses: amqp://usuario:clave@servidor:5672/vhost
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 5

Definición de Constantes

public final class RabbitConstants {
    public static final String COLA_CONTROL_FLUJO = "control.flujo.cola";
    public static final String INTERCAMBIO_CONTROL_FLUJO = "control.flujo.intercambio";
}

Declaración de Recursos y Vínculos

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ConfiguracionControlFlujo {

    @Bean
    public Queue colaControlFlujo() {
        return QueueBuilder.durable(RabbitConstants.COLA_CONTROL_FLUJO).build();
    }

    @Bean
    public DirectExchange intercambioControlFlujo() {
        return ExchangeBuilder.directExchange(RabbitConstants.INTERCAMBIO_CONTROL_FLUJO).build();
    }

    @Bean
    public Binding vinculacionControlFlujo(Queue cola, DirectExchange intercambio) {
        return BindingBuilder.bind(cola).to(intercambio).with("flujo.clave");
    }
}

Implementación de Limitación de Velocidad (Rate Limiting)

Imaginemos un sistema de pedidos que puede manejar 5000 solicitudes por segundo bajo condiciones normales. Durante un evento de venta flash, el tráfico se dispara a 10,000 solicitudes por segundo. Si todas estas solicitudes fueran encoladas y enviadas sin control al sistema, este colapsaría. El mecanismo de prefetch permite que el consumidor solo obtenga una cantidad controlada de mensajes a la vez.

Productor de Mensajes

@RestController
@RequestMapping("/api")
public class ControladorMensajes {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/enviar-lote")
    public String enviarLoteMensajes() {
        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(
                RabbitConstants.INTERCAMBIO_CONTROL_FLUJO,
                "flujo.clave",
                "Solicitud #" + i
            );
        }
        return "Lote de 20 mensajes enviado.";
    }
}

Consumidor con Confirmación Pendiente (Efecto de Limitación)

@Component
public class ConsumidorControlFlujo {

    @RabbitListener(queues = RabbitConstants.COLA_CONTROL_FLUJO)
    public void procesarSinConfirmar(Message mensaje, Channel canal) {
        System.out.printf("[Consumidor A] Recibido: %s, Etiqueta: %d%n",
            new String(mensaje.getBody(), StandardCharsets.UTF_8),
            mensaje.getMessageProperties().getDeliveryTag());
    }
}

Resultado observado: Solo se procesan 5 mensajes, los cuales permanecen en estado no confirmado.

Consumidor con Confirmación Explícita

@Component
public class ConsumidorConfirmado {

    @RabbitListener(queues = RabbitConstants.COLA_CONTROL_FLUJO)
    public void procesarYConfirmar(Message mensaje, Channel canal) throws IOException {
        long etiqueta = mensaje.getMessageProperties().getDeliveryTag();
        System.out.printf("[Consumidor B] Procesado: %s, Etiqueta: %d%n",
            new String(mensaje.getBody(), StandardCharsets.UTF_8), etiqueta);
        canal.basicAck(etiqueta, false);
    }
}

Resultado observado: Todos los mensajes del lote se consumen secuencialmente a medida que se van confirmando.

Balanceo de Carga Dinámico

La misma configuración de prefetch sirve para equilibrar la carga entre consumidores con velocidades de procesamiento heterogéneas. Al establecer prefetch: 1, un consumidor lento no bloquea la entrega de mensajes a un consumidor rápido.

Ajuste de Configuración

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1

Consumidores con Diferentes Velocidades

@Component
public class ConsumidoresBalanceados {

    @RabbitListener(queues = RabbitConstants.COLA_CONTROL_FLUJO)
    public void consumidorLento(Message mensaje, Channel canal) throws Exception {
        long etiqueta = mensaje.getMessageProperties().getDeliveryTag();
        try {
            System.out.printf("[Lento] Inicio procesamiento: %s%n", new String(mensaje.getBody(), StandardCharsets.UTF_8));
            Thread.sleep(2000); // Simula procesamiento lento
            System.out.println("[Lento] Fin procesamiento.");
            canal.basicAck(etiqueta, false);
        } catch (Exception e) {
            canal.basicNack(etiqueta, false, true);
        }
    }

    @RabbitListener(queues = RabbitConstants.COLA_CONTROL_FLUJO)
    public void consumidorRapido(Message mensaje, Channel canal) throws Exception {
        long etiqueta = mensaje.getMessageProperties().getDeliveryTag();
        try {
            System.out.printf("[Rápido] Procesado rápido: %s%n", new String(mensaje.getBody(), StandardCharsets.UTF_8));
            Thread.sleep(500); // Simula procesamiento rápido
            canal.basicAck(etiqueta, false);
        } catch (Exception e) {
            canal.basicNack(etiqueta, false, true);
        }
    }
}

Resultado observado: El consumidor "Rápido" procesa aproximadamente el doble de mensajes que el consumidor "Lento", logrando un balanceo de carga efectivo basado en la capacidad real de procesamiento.