RabbitMQ com Spring AMQP e Spring Boot: RabbitTemplate, @RabbitListener e retries

O cliente Java do RabbitMQ oferece channels, arrays de bytes e callbacks. O Spring AMQP oferece um RabbitTemplate que publica objetos, uma anotação @RabbitListener que transforma um método em um consumidor escalável e um RabbitAdmin que cria os exchanges e as filas descritos pelos seus beans. O Spring Boot acrescenta a autoconfiguração e as propriedades por cima. O resultado é um serviço em que o código de mensageria cabe em algumas dezenas de linhas, e este guia mostra essas linhas junto com as configurações que as tornam seguras sob carga. O alvo é o Spring Boot 3.x com Spring AMQP 3.x; as ideias valem sem mudanças para o Spring AMQP 4, embora alguns nomes de classes sejam diferentes.

Configuração e conexão

Um único starter traz o Spring AMQP e o cliente do RabbitMQ:

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

O Boot cria uma CachingConnectionFactory — uma única conexão cujos channels ficam em pool e são reutilizados entre threads, que é o padrão correto no AMQP — e a configura a partir de propriedades:

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

A lista addresses dá ao cliente vários nós para tentar; a factory com cache se reconecta ao próximo quando um nó cai, e os consumidores refazem a assinatura automaticamente. Todo o resto deste arquivo é explicado ao longo do guia.

Declarando a topologia como beans

O Spring AMQP modela exchanges, filas e bindings como beans. Na primeira conexão, o RabbitAdmin (autoconfigurado pelo Boot) declara todos os Declarable do contexto e os declara de novo após uma reconexão. As declarações são idempotentes, então várias instâncias do serviço podem subir em qualquer ordem.

@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);
  }
}

Dois hábitos mantêm isso gerenciável à medida que o serviço cresce. Coloque os nomes em constantes (ou em uma classe @ConfigurationProperties) para que publicadores, listeners e testes usem os mesmos valores, e agrupe declarações relacionadas em um bean Declarables quando forem muitas, em vez de um bean por binding. Lembre-se de que os argumentos de uma fila são imutáveis: se mais tarde você mudar deliveryLimit em uma fila que já existe, o RabbitMQ recusa a declaração com um erro PRECONDITION_FAILED e o listener não sobe. Apague a fila ou expresse a configuração como policy.

Publicando com RabbitTemplate

RabbitTemplate.convertAndSend serializa um objeto com o MessageConverter configurado, define o content type e publica em um exchange com uma routing key. Por padrão, o conversor é o SimpleMessageConverter, que lida com strings, arrays de bytes e serialização Java; declare um conversor JSON e o Boot o injeta tanto no template quanto na 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()));
  }
}

Com publisher-confirm-type: correlated, o broker confirma ou rejeita cada publicação e o template chama o seu ConfirmCallback com o CorrelationData que você passou. Um nack significa que o broker não conseguiu assumir a responsabilidade pela mensagem — um nó caiu no meio da publicação, ou uma quorum queue ficou sem maioria — e a reação certa é fazer retry ou registrar a falha, não logar e esquecer. O returns callback dispara quando mandatory é true e a routing key não bateu com nenhum binding, e é a única forma de perceber um erro de digitação em uma routing key.

A lambda MessagePostProcessor é onde se definem headers, identificadores de mensagem, prioridades e expirações. O modo de entrega persistente é o padrão do template.

Consumindo com @RabbitListener

Um método @RabbitListener vira um consumidor apoiado em um listener container. O Boot cria um SimpleMessageListenerContainer por anotação a partir das propriedades spring.rabbitmq.listener.simple.*; o payload é convertido pelo mesmo conversor JSON, e os headers podem ser injetados.

@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" inicia duas threads consumidoras e escala até oito enquanto a fila tiver backlog. Cada thread tem o próprio channel com o prefetch configurado, então oito consumidores com prefetch de 20 podem ter 160 mensagens sem ack ao mesmo tempo — dimensione o prefetch com essa multiplicação em mente.

Com acknowledge-mode: auto (o padrão), o container faz o ack depois que o método retorna e envia um acknowledgement negativo quando ele lança uma exceção. Como default-requeue-rejected é false na configuração acima, uma mensagem rejeitada vai para o dead-letter exchange em vez de voltar para o início da fila. Para rejeitar uma mensagem de propósito independentemente dessa configuração, lance AmqpRejectAndDontRequeueException; para pedir que ela volte para a fila quando um sistema downstream estiver brevemente indisponível, lance ImmediateRequeueAmqpException.

Acknowledgement manual

Algumas cargas de trabalho precisam fazer o ack em um ponto preciso — depois de um commit no banco de dados, ou em lotes. Mude o container para AcknowledgeMode.MANUAL e injete o channel e o 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
  }
}

No modo manual, o container não faz o ack por você, então um método que esquece o ack deixa a mensagem sem ack até o channel ser fechado — e o consumer timeout do broker o fecha depois de trinta minutos, com erro.

Declarando a fila no listener

Em serviços pequenos é prático declarar o binding na própria anotação; o RabbitAdmin o reconhece como qualquer outro 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) { ... }

Retries e dead letters

Há três camadas em que uma mensagem com falha pode passar por retry, e um serviço robusto usa duas delas.

O retry no nível do container é só uma mudança de propriedades:

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

Quando os retries se esgotam, o MessageRecoverer do container decide o destino da mensagem. O RejectAndDontRequeueRecoverer padrão a rejeita sem requeue, o que a envia para o dead-letter exchange configurado na fila. RepublishMessageRecoverer é uma alternativa que publica a mensagem em um exchange à sua escolha com headers descrevendo a exceção e a stack trace, algo muito mais útil quando alguém abre a dead-letter queue depois para entender o que deu errado:

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

Seja qual for o mecanismo, uma dead-letter queue precisa de um dono: um alerta sobre a profundidade da fila, um listener que registre a falha em algum lugar visível e uma forma documentada de reprocessar depois que o bug for corrigido — normalmente o recurso move messages da interface de management (com o plugin shovel) ou um pequeno endpoint administrativo que lê a dead-letter queue e republica.

Request-reply

RabbitTemplate.convertSendAndReceive implementa chamadas de procedimento remoto sobre AMQP: publica uma requisição, espera uma resposta com o identificador de correlação correspondente e converte o resultado. Por padrão, usa a pseudofila direct reply-to do RabbitMQ (amq.rabbitmq.reply-to), que evita criar uma fila temporária por chamada e é rápida o bastante para uso interativo.

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

No lado do servidor, um método @RabbitListener simplesmente retorna um valor; o container o envia para o replyTo da requisição com o mesmo correlationId:

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

Defina spring.rabbitmq.template.reply-timeout (cinco segundos por padrão) com um valor compatível com o timeout do próprio chamador, e lembre-se de que o request-reply reintroduz o acoplamento temporal que a mensageria deveria eliminar: se o serviço de preços estiver fora do ar, o chamador falha depois do timeout exatamente como faria com uma chamada HTTP.

Testando contra um broker real

Mockar o RabbitTemplate não testa nada sobre routing keys, conversores ou acknowledgements. O Testcontainers sobe um RabbitMQ real no Docker, e o @ServiceConnection do Boot (3.1+) preenche as propriedades de conexão sem nenhuma configuração 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 os beans de topologia declaram tudo na inicialização, o teste roda contra um broker vazio, sem fixtures. Use a tag de imagem -management quando quiser abrir a interface na porta mapeada enquanto depura um teste que está falhando.

Erros comuns

• Esquecer o conversor JSON. Os listeners recebem byte[] ou falham com uma ClassCastException; os publicadores enviam payloads com serialização Java que consumidores não Java não conseguem ler. Declare o bean conversor uma única vez.
• Deixar default-requeue-rejected em true. Um listener que lança exceção diante de uma mensagem malformada fica girando a toda velocidade na mesma mensagem para sempre. Defina como false e use dead letter.
• Ignorar os callbacks de confirm e return. As publicações somem em silêncio diante de uma instabilidade momentânea do broker ou de uma routing key digitada errado. Ative os confirms correlated e mandatory, e registre em log ou faça retry nos callbacks.
• Alterar argumentos de uma fila existente. PRECONDITION_FAILED na inicialização. Apague a fila fora de produção; use policies em produção.
• Retry bloqueante com intervalos longos. Uma thread de listener dormindo minutos entre retries é uma thread que você não tem mais. Mantenha curtos os retries do container e leve as esperas longas para uma fila com TTL.
• Compartilhar estado mutável entre threads de listener concorrentes. concurrency = "2-8" significa que o seu bean listener é chamado por oito threads ao mesmo tempo; mantenha-o sem estado ou sincronize corretamente.

Principais conclusões

• O spring-boot-starter-amqp do Boot entrega uma connection factory com cache, um RabbitTemplate, um RabbitAdmin e listener containers configurados pelas propriedades spring.rabbitmq.*.
• Declare exchanges, filas e bindings como beans; eles são criados de forma idempotente na primeira conexão, e quorum queues com um dead-letter exchange deveriam ser o formato padrão.
• Publique com um MessageConverter JSON, confirms correlacionados e returns com mandatory para saber o que o broker aceitou.
• O @RabbitListener escala com concurrency, é limitado pelo prefetch e faz o ack automaticamente no retorno ou rejeita em caso de exceção; use o modo manual só quando o ponto de acknowledgement precisar ser preciso.
• Faça retries curtos no container e depois mande para dead letter via broker; garanta que default-requeue-rejected seja false e que alguém seja dono da dead-letter queue.
• Teste o fluxo real com Testcontainers e @ServiceConnection; mocks não pegam erros de roteamento nem de serialização.