RabbitMQ con Spring AMQP y Spring Boot: RabbitTemplate, @RabbitListener y reintentos

El cliente Java de RabbitMQ te da channels, arrays de bytes y callbacks. Spring AMQP te da un RabbitTemplate que publica objetos, una anotación @RabbitListener que convierte un método en un consumidor escalable y un RabbitAdmin que crea los exchanges y las colas que describen tus beans. Spring Boot añade encima la autoconfiguración y las propiedades. El resultado es un servicio cuyo código de mensajería ocupa unas pocas decenas de líneas; esta guía muestra esas líneas junto con los ajustes que las hacen seguras bajo carga. Está pensada para Spring Boot 3.x con Spring AMQP 3.x; las ideas se trasladan sin cambios a Spring AMQP 4, aunque algunos nombres de clases difieren.

Configuración y conexión

Un único starter trae Spring AMQP y el cliente de RabbitMQ:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Boot crea una CachingConnectionFactory —una sola conexión cuyos channels se agrupan en un pool y se reutilizan entre threads, que es el patrón correcto en AMQP— y la configura a partir de propiedades:

spring:
  rabbitmq:
    addresses: amqps://rabbit-1.internal:5671,amqps://rabbit-2.internal:5671   # cluster nodes, TLS
    username: orders
    password: ${RABBITMQ_PASSWORD}
    virtual-host: shop
    connection-timeout: 5s
    publisher-confirm-type: correlated     # enable confirms with a CorrelationData callback
    publisher-returns: true                # get unroutable messages back
    template:
      mandatory: true                      # required for returns to fire
      retry:
        enabled: true                      # retry a failed publish (broker unavailable)
        max-attempts: 3
        initial-interval: 500ms
    listener:
      simple:
        acknowledge-mode: auto             # ack on success, nack on exception
        prefetch: 20
        concurrency: 2
        max-concurrency: 8
        default-requeue-rejected: false    # rejected messages go to the DLX, not back to the queue

La lista addresses le da al cliente varios nodos que probar; la factory con caché se reconecta al siguiente cuando un nodo desaparece, y los consumidores se vuelven a suscribir automáticamente. El resto del archivo se explica a medida que aparece en la guía.

Declarar la topología como beans

Spring AMQP modela exchanges, colas y bindings como beans. En la primera conexión, RabbitAdmin (autoconfigurado por Boot) declara todos los Declarable del contexto y los vuelve a declarar tras una reconexión. Las declaraciones son idempotentes, así que varias instancias del servicio pueden arrancar en cualquier orden.

@Configuration
class OrdersTopology {

  static final String EXCHANGE = "orders";
  static final String QUEUE = "fulfilment";
  static final String DLX = "orders.dlx";

  @Bean TopicExchange ordersExchange() { return ExchangeBuilder.topicExchange(EXCHANGE).durable(true).build(); }

  @Bean DirectExchange ordersDlx() { return ExchangeBuilder.directExchange(DLX).durable(true).build(); }

  @Bean Queue fulfilmentQueue() {
    return QueueBuilder.durable(QUEUE)
        .quorum()                              // x-queue-type=quorum
        .deadLetterExchange(DLX)
        .deadLetterRoutingKey(QUEUE)
        .deliveryLimit(5)                      // quorum queues: dead-letter after 5 redeliveries
        .build();
  }

  @Bean Queue fulfilmentDlq() { return QueueBuilder.durable(QUEUE + ".dlq").quorum().build(); }

  @Bean Binding fulfilmentBinding() {
    return BindingBuilder.bind(fulfilmentQueue()).to(ordersExchange()).with("order.*.placed");
  }

  @Bean Binding fulfilmentDlqBinding() {
    return BindingBuilder.bind(fulfilmentDlq()).to(ordersDlx()).with(QUEUE);
  }
}

Dos hábitos mantienen esto manejable a medida que el servicio crece. Guarda los nombres en constantes (o en una clase @ConfigurationProperties) para que productores, listeners y tests usen los mismos valores, y agrupa las declaraciones relacionadas en un bean Declarables cuando tengas muchas, en lugar de un bean por binding. Recuerda que los argumentos de una cola son inmutables: si más adelante cambias deliveryLimit en una cola que ya existe, RabbitMQ rechaza la declaración con un error PRECONDITION_FAILED y el listener no arranca. Borra la cola o expresa el ajuste como policy.

Publicar con RabbitTemplate

RabbitTemplate.convertAndSend serializa un objeto con el MessageConverter configurado, fija el content type y lo publica en un exchange con una routing key. Por defecto el conversor es SimpleMessageConverter, que maneja strings, arrays de bytes y serialización Java; declara un conversor JSON y Boot lo inyecta tanto en el template como en la factory de listener containers:

@Bean
MessageConverter jsonMessageConverter() {
  return new Jackson2JsonMessageConverter();   // Spring AMQP 3.x; JacksonJsonMessageConverter in 4.x
}

@Service
class OrderEventPublisher {
  private final RabbitTemplate rabbit;

  OrderEventPublisher(RabbitTemplate rabbit) {
    this.rabbit = rabbit;
    rabbit.setConfirmCallback((correlation, ack, cause) -> {
      if (!ack) log.error("broker rejected {}: {}", correlation != null ? correlation.getId() : "?", cause);
    });
    rabbit.setReturnsCallback(returned ->
      log.error("unroutable message to {} with key {}: {}",
          returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()));
  }

  public void orderPlaced(OrderPlaced event) {
    rabbit.convertAndSend(OrdersTopology.EXCHANGE, "order." + event.region() + ".placed", event,
        message -> {
          message.getMessageProperties().setMessageId(event.orderId());   // for consumer-side de-duplication
          message.getMessageProperties().setHeader("schemaVersion", 2);
          return message;
        },
        new CorrelationData(event.orderId()));
  }
}

Con publisher-confirm-type: correlated, el broker confirma o rechaza cada publicación y el template invoca tu ConfirmCallback con el CorrelationData que le pasaste. Un nack significa que el broker no pudo hacerse responsable del mensaje —se perdió un nodo en plena publicación o una quorum queue se quedó sin mayoría— y lo correcto es reintentar o registrar el fallo, no escribir un log y olvidarse. El returns callback se dispara cuando mandatory es true y la routing key no coincide con ningún binding, y es la única forma de detectar una errata en una routing key.

La lambda MessagePostProcessor es donde se fijan cabeceras, identificadores de mensaje, prioridades y expiraciones. El modo de entrega persistente es el valor por defecto del template.

Consumir con @RabbitListener

Un método @RabbitListener se convierte en un consumidor respaldado por un listener container. Boot crea un SimpleMessageListenerContainer por anotación a partir de las propiedades spring.rabbitmq.listener.simple.*; el payload se convierte con el mismo conversor JSON y las cabeceras se pueden inyectar.

@Component
class FulfilmentListener {

  @RabbitListener(queues = OrdersTopology.QUEUE, concurrency = "2-8")
  public void onOrderPlaced(OrderPlaced event,
                            @Header(AmqpHeaders.MESSAGE_ID) String messageId,
                            @Header(name = "x-delivery-count", required = false) Long deliveryCount) {
    if (fulfilments.alreadyProcessed(messageId)) return;          // idempotent: duplicates are harmless
    fulfilments.fulfil(event);                                    // throwing here → nack → DLX (requeue disabled)
  }
}

concurrency = "2-8" arranca dos threads consumidores y escala hasta ocho mientras la cola tenga mensajes acumulados. Cada thread tiene su propio channel con el prefetch configurado, así que ocho consumidores con un prefetch de 20 pueden tener 160 mensajes sin ack a la vez: dimensiona el prefetch teniendo en cuenta esa multiplicación.

Con acknowledge-mode: auto (el valor por defecto), el container hace ack cuando el método retorna y envía un ack negativo cuando lanza una excepción. Como default-requeue-rejected es false en la configuración anterior, un mensaje rechazado va al dead-letter exchange en vez de volver al principio de la cola. Para rechazar un mensaje a propósito independientemente de ese ajuste, lanza AmqpRejectAndDontRequeueException; para pedir que se reencole cuando un sistema downstream no está disponible momentáneamente, lanza ImmediateRequeueAmqpException.

Acknowledgement manual

Algunas cargas de trabajo necesitan hacer el ack en un punto concreto: después de un commit en la base de datos, o por lotes. Cambia el container a AcknowledgeMode.MANUAL e inyecta el channel y el delivery tag:

@RabbitListener(queues = OrdersTopology.QUEUE, ackMode = "MANUAL")
public void onOrderPlaced(OrderPlaced event, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
  try {
    fulfilments.fulfil(event);
    channel.basicAck(tag, false);
  } catch (DownstreamUnavailable e) {
    channel.basicNack(tag, false, true);      // requeue: another instance may succeed
  } catch (RuntimeException e) {
    channel.basicNack(tag, false, false);     // dead-letter it
  }
}

En modo manual el container no hace el ack por ti, así que un método que olvida el ack deja el mensaje sin ack hasta que se cierra el channel, y el consumer timeout del broker lo cierra a los treinta minutos con un error.

Declarar la cola en el listener

En servicios pequeños resulta cómodo declarar el binding en la propia anotación; RabbitAdmin lo recoge como cualquier otro declarable:

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "notifications", durable = "true",
                   arguments = @Argument(name = "x-queue-type", value = "quorum")),
    exchange = @Exchange(value = "orders", type = ExchangeTypes.TOPIC),
    key = "order.#"))
public void notify(OrderPlaced event) { ... }

Reintentos y dead letters

Un mensaje fallido se puede reintentar en tres niveles, y un servicio robusto usa dos de ellos.

El reintento a nivel de container es un simple cambio de propiedades:

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 4          # the original delivery plus three retries
          initial-interval: 1s
          multiplier: 2.0          # 1 s, 2 s, 4 s
          max-interval: 10s
        default-requeue-rejected: false

Cuando se agotan los reintentos, el MessageRecoverer del container decide el destino del mensaje. El RejectAndDontRequeueRecoverer por defecto lo rechaza sin reencolar, lo que lo envía al dead-letter exchange configurado en la cola. RepublishMessageRecoverer es una alternativa que publica el mensaje en el exchange que elijas con cabeceras que describen la excepción y la stack trace, algo mucho más útil cuando alguien abre después la dead-letter queue para entender qué salió mal:

@Bean
MessageRecoverer messageRecoverer(RabbitTemplate template) {
  return new RepublishMessageRecoverer(template, OrdersTopology.DLX, OrdersTopology.QUEUE);
}

Sea cual sea el mecanismo, una dead-letter queue necesita un responsable: una alerta sobre su profundidad, un listener que registre el fallo en un lugar visible y un procedimiento documentado para reprocesar una vez corregido el bug, normalmente la función move messages de la interfaz de management (con el plugin shovel) o un pequeño endpoint de administración que lee la dead-letter queue y vuelve a publicar.

Request-reply

RabbitTemplate.convertSendAndReceive implementa llamadas a procedimiento remoto sobre AMQP: publica una petición, espera una respuesta con el mismo identificador de correlación y convierte el resultado. Por defecto usa la pseudocola direct reply-to de RabbitMQ (amq.rabbitmq.reply-to), que evita crear una cola temporal por llamada y es lo bastante rápida para uso interactivo.

PriceQuote quote = rabbit.convertSendAndReceiveAsType(
    "pricing", "quote.request", new QuoteRequest("A-100", 3),
    new ParameterizedTypeReference<PriceQuote>() {});

En el lado servidor, un método @RabbitListener simplemente devuelve un valor; el container lo envía al replyTo de la petición con el mismo correlationId:

@RabbitListener(queues = "pricing.requests")
public PriceQuote quote(QuoteRequest request) {
  return pricing.quote(request.sku(), request.quantity());
}

Ajusta spring.rabbitmq.template.reply-timeout (cinco segundos por defecto) a un valor acorde con el timeout del propio llamante, y recuerda que request-reply reintroduce el acoplamiento temporal que la mensajería pretendía eliminar: si el servicio de precios está caído, el llamante falla tras el timeout exactamente igual que con una llamada HTTP.

Pruebas contra un broker real

Mockear RabbitTemplate no prueba nada sobre routing keys, conversores ni acknowledgements. Testcontainers arranca un RabbitMQ real en Docker, y el @ServiceConnection de Boot (3.1+) rellena las propiedades de conexión sin ninguna configuración manual:

@SpringBootTest
@Testcontainers
class FulfilmentFlowTest {

  @Container @ServiceConnection
  static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:4.0-management");

  @Autowired OrderEventPublisher publisher;
  @Autowired FulfilmentRepository fulfilments;

  @Test
  void placedOrderIsFulfilled() {
    publisher.orderPlaced(new OrderPlaced("o-1", "eu", 149.90));
    Awaitility.await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() -> assertThat(fulfilments.find("o-1")).isPresent());
  }
}

Como los beans de topología lo declaran todo al arrancar, el test se ejecuta contra un broker vacío sin fixtures. Usa la etiqueta de imagen -management cuando quieras abrir la interfaz en el puerto mapeado mientras depuras un test que falla.

Errores habituales

• Falta el conversor JSON. Los listeners reciben byte[] o fallan con una ClassCastException; los productores envían payloads serializados en Java que los consumidores no Java no pueden leer. Declara el bean conversor una sola vez.
• Dejar default-requeue-rejected en true. Un listener que lanza una excepción ante un mensaje malformado se queda dando vueltas a toda velocidad sobre ese mismo mensaje indefinidamente. Ponlo a false y usa dead letter.
• Ignorar los callbacks de confirm y return. Las publicaciones desaparecen en silencio ante un fallo momentáneo del broker o una routing key mal escrita. Activa los confirms correlated y mandatory, y registra o reintenta en los callbacks.
• Cambiar los argumentos de una cola existente. PRECONDITION_FAILED al arrancar. Borra la cola fuera de producción; usa policies en producción.
• Reintentos bloqueantes con intervalos largos. Un thread de listener que duerme minutos entre reintentos es un thread que ya no tienes. Mantén cortos los reintentos del container y lleva las esperas largas a una cola con TTL.
• Compartir estado mutable entre threads de listener concurrentes. concurrency = "2-8" significa que tu bean listener se invoca desde ocho threads a la vez; mantenlo sin estado o sincroniza como es debido.

Puntos clave

• El spring-boot-starter-amqp de Boot te da una connection factory con caché, un RabbitTemplate, un RabbitAdmin y listener containers configurados con las propiedades spring.rabbitmq.*.
• Declara exchanges, colas y bindings como beans; se crean de forma idempotente en la primera conexión, y las quorum queues con un dead-letter exchange deberían ser la forma por defecto.
• Publica con un MessageConverter JSON, confirms correlacionados y returns con mandatory para saber qué ha aceptado el broker.
• @RabbitListener escala con concurrency, está acotado por el prefetch y hace ack automáticamente al retornar o rechaza ante una excepción; usa el modo manual solo cuando el punto de ack deba ser preciso.
• Reintenta brevemente en el container y después manda a dead letter a través del broker; asegúrate de que default-requeue-rejected sea false y de que alguien se haga cargo de la dead-letter queue.
• Prueba el flujo real con Testcontainers y @ServiceConnection; los mocks no detectan errores de enrutamiento ni de serialización.