RabbitMQ e AMQP 0-9-1: exchanges, bindings, filas e roteamento

O RabbitMQ é o broker que implementa o AMQP 0-9-1 com mais fidelidade, e quase tudo o que confunde quem está começando vem de não conhecer o modelo desse protocolo. Um produtor nunca envia para uma fila. Ele envia para um exchange, o exchange consulta seus bindings, e cópias da mensagem chegam a zero, uma ou muitas filas, de onde os consumidores as retiram. Com essa imagem clara, a flexibilidade do RabbitMQ — a mesma mensagem indo para uma fila de trabalho, um log de auditoria e o feed de um dashboard sem que o produtor saiba — vira algo com que você projeta, em vez de algo contra o que você briga. Este guia trata do RabbitMQ 3.13 e 4.x com o cliente Java oficial.

O modelo AMQP 0-9-1

Cinco conceitos dão conta do protocolo.

Uma conexão é uma conexão TCP de longa duração, geralmente com TLS, autenticada em um virtual host (vhost) — um namespace de exchanges, filas e permissões, que permite a um mesmo broker atender várias aplicações de forma isolada. Dentro de uma conexão, as aplicações abrem channels: sessões multiplexadas e baratas nas quais acontece toda a publicação e o consumo. A regra é uma conexão por processo e um channel por thread, porque channels não são thread-safe e conexões são caras.

Um exchange recebe mensagens dos publishers e as roteia. Ele não armazena nada. Uma fila (queue) armazena mensagens e as entrega aos consumidores. Um binding liga um exchange a uma fila e pode carregar uma binding key (ou, nos exchanges headers, um conjunto de cabeçalhos). Quando um publisher envia uma mensagem, ele indica um exchange e fornece uma routing key; o tipo do exchange decide como a routing key é comparada com as binding keys.

Tudo isso é programável: as aplicações declaram exchanges, filas e bindings pelo próprio protocolo, e as declarações são idempotentes desde que os argumentos sejam os mesmos. É por isso que tanto o produtor quanto o consumidor podem declarar a fila que compartilham e subir em qualquer ordem.

publisher --(exchange, routing key)--> exchange --bindings--> queue(s) --> consumer(s)

Os quatro tipos de exchange

Direct

Um exchange direct roteia uma mensagem para cada fila cuja binding key seja exatamente igual à routing key. É a ferramenta para entrega ponto a ponto e para distribuir mensagens em um conjunto pequeno e fixo de filas, como uma fila por nível de severidade de log.

O exchange padrão é um exchange direct pré-declarado cujo nome é a string vazia. Toda fila é automaticamente ligada a ele usando o próprio nome como binding key, então basicPublish("", "orders", ...) entrega na fila chamada orders. É prático para tutoriais e filas de resposta, e uma armadilha para todo o resto: os produtores acabam fixando nomes de filas no código, que é exatamente o acoplamento que os exchanges existem para eliminar.

Fanout

Um exchange fanout ignora a routing key e copia a mensagem para todas as filas ligadas. É a primitiva de broadcast: ligue uma fila por assinante e cada um recebe a própria cópia, no próprio ritmo, com os próprios acknowledgements.

Topic

Um exchange topic compara a routing key com padrões de binding. Routing keys são palavras separadas por pontos, com até 255 bytes; em uma binding key, * corresponde a exatamente uma palavra e # a zero ou mais palavras.

Um exchange topic com binding key # se comporta como um fanout, e com uma chave sem curingas se comporta como um exchange direct. Por isso muitos times usam um único exchange topic por bounded context e expressam tudo como convenções de routing key, como <entity>.<region>.<event>.

Headers

Um exchange headers ignora a routing key e compara os cabeçalhos da mensagem com os argumentos do binding; o argumento x-match com valor all exige que todos os cabeçalhos listados correspondam, e any exige ao menos um. Ele roteia por vários atributos ao mesmo tempo, ao custo de ser mais lento e menos legível que as chaves topic. A maioria dos designs que recorrem a ele sairia ganhando se codificasse esses atributos em uma routing key.

Declarando a topologia em Java

O cliente oficial é o com.rabbitmq:amqp-client (5.x). Este produtor declara um exchange topic durável e uma quorum queue durável, faz o binding entre eles e publica uma mensagem persistente:

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class OrderPublisher {
  public static void main(String[] args) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri("amqps://orders:[email protected]:5671/shop");  // vhost "shop"

    try (Connection connection = factory.newConnection("order-service");
         Channel channel = connection.createChannel()) {

      channel.exchangeDeclare("orders", BuiltinExchangeType.TOPIC, /* durable */ true);
      channel.queueDeclare("fulfilment", /* durable */ true, /* exclusive */ false,
          /* autoDelete */ false, Map.of("x-queue-type", "quorum"));
      channel.queueBind("fulfilment", "orders", "order.*.placed");

      channel.confirmSelect();                      // enable publisher confirms on this channel

      AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
          .contentType("application/json")
          .deliveryMode(2)                          // 2 = persistent: written to disk by durable queues
          .messageId("o-9911")
          .build();

      channel.basicPublish("orders", "order.eu.placed", /* mandatory */ true, props,
          "{\"orderId\":\"o-9911\",\"total\":149.90}".getBytes(StandardCharsets.UTF_8));

      channel.waitForConfirmsOrDie(5_000);          // throws if the broker did not confirm in 5 s
    }
  }
}

Três flags nesse código decidem a durabilidade, e as três são necessárias para que uma mensagem sobreviva a um restart do broker: o exchange precisa ser durável (senão ele some junto com seus bindings), a fila precisa ser durável e a mensagem precisa ser publicada com delivery mode 2. Quorum queues são sempre duráveis e sempre persistem as mensagens; a flag de delivery mode continua valendo para filas clássicas.

Tipos de fila

O RabbitMQ tem três tipos de fila, escolhidos com o argumento x-queue-type na declaração e imutáveis depois.

Uma quorum queue mantém suas mensagens na maioria dos nós do cluster e elege um líder via Raft; ela sobrevive à perda de uma minoria dos nós sem perder mensagens confirmadas, e foi essa propriedade que a tornou a escolha padrão recomendada. Ela também registra quantas vezes uma mensagem foi reentregue no cabeçalho x-delivery-count e pode mover mensagens para um dead-letter exchange depois de um x-delivery-limit configurável, algo que filas clássicas não fazem. O custo é mais disco e memória por mensagem e latência maior do que a de uma fila clássica em um único nó.

Um stream é um log append-only: os consumidores leem a partir de um offset, vários consumidores podem ler as mesmas mensagens e a retenção é definida por tamanho ou idade, não por consumo. Streams falam AMQP 0-9-1 pelo cliente normal, mas o protocolo e o cliente dedicados de stream são muito mais rápidos. Quando você se pegar querendo replay no estilo Kafka dentro do RabbitMQ, streams são a resposta.

O RabbitMQ 4.0 também tornou o AMQP 1.0 um protocolo central ao lado do 0-9-1, então uma aplicação Java pode usar o cliente Qpid JMS ou o cliente AMQP 1.0 do próprio RabbitMQ nas mesmas filas; o modelo de exchanges e bindings descrito aqui continua o mesmo.

Consumo, acknowledgements e prefetch

Um consumidor se inscreve em uma fila com basicConsume, e o broker empurra as entregas para ele. O que acontece em seguida depende do modo de acknowledgement escolhido.

Com autoAck = true, o broker esquece a mensagem assim que a escreve no socket. Se o consumidor cair, tudo o que estava em trânsito se perde. Com autoAck = false, a mensagem continua na fila, marcada como unacked, até o consumidor enviar basicAck; se o channel fechar antes, o broker a reenfileira para outro consumidor. Isso é entrega at-least-once.

channel.basicQos(20);   // prefetch: at most 20 unacknowledged deliveries per consumer on this channel

channel.basicConsume("fulfilment", /* autoAck */ false, "fulfilment-worker-1",
    new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String tag, Envelope env, AMQP.BasicProperties props, byte[] body) throws IOException {
        long deliveryTag = env.getDeliveryTag();
        try {
          fulfil(new String(body, StandardCharsets.UTF_8));
          channel.basicAck(deliveryTag, /* multiple */ false);
        } catch (TransientFailure e) {
          channel.basicNack(deliveryTag, false, /* requeue */ true);    // try again, possibly elsewhere
        } catch (RuntimeException e) {
          channel.basicNack(deliveryTag, false, /* requeue */ false);   // dead-letter it, if a DLX is set
        }
      }
    });

basicQos define o prefetch count: quantas entregas o broker pode empurrar para um consumidor antes de receber acknowledgements. Sem ele, o broker envia tão rápido quanto o socket permite, um consumidor acumula milhares de mensagens na memória enquanto os outros ficam ociosos, e uma queda reenfileira todas elas. Um prefetch entre 10 e 100 atende à maioria das cargas; 1 garante equidade perfeita ao custo de uma ida e volta por mensagem.

Dois limites protegem o broker de consumidores que se comportam mal. Reentregar com requeue = true uma mensagem que falha toda vez cria um loop infinito que queima CPU; verifique env.isRedeliver() ou, em quorum queues, o cabeçalho x-delivery-count, e pare de reenfileirar depois de algumas tentativas. Além disso, um consumidor que segura uma entrega sem confirmá-la por mais tempo que o consumer timeout (30 minutos por padrão) tem seu channel fechado pelo broker, uma surpresa comum em consumidores no estilo batch.

Publisher confirms

Os acknowledgements protegem o lado do consumidor; os publisher confirms protegem o publisher. Depois de confirmSelect, o broker envia basic.ack para cada mensagem assim que ela é roteada e, no caso de mensagens persistentes em filas duráveis, gravada em disco, ou basic.nack se não conseguiu tratá-la. waitForConfirmsOrDie é a forma síncrona simples; para ter throughput, publique em lotes ou registre um ConfirmListener e acompanhe os números de sequência pendentes, republicando tudo o que receber nack. Junto com mandatory = true e um ReturnListener, que devolve as mensagens que não corresponderam a nenhum binding, os confirms dão a você um publisher que sabe o que o broker aceitou.

TTL, limites e dead-lettering

As filas aceitam argumentos opcionais que controlam o que acontece com as mensagens que não são consumidas.

• x-message-ttl — milissegundos que uma mensagem pode esperar antes de ser descartada ou enviada para dead letter. A propriedade expiration de cada mensagem faz o mesmo para uma única mensagem.
• x-expires — apaga a própria fila depois desse número de milissegundos sem consumidores nem atividade; útil para filas de resposta por cliente.
• x-max-length / x-max-length-bytes com x-overflow — limita o tamanho da fila; drop-head descarta as mensagens mais antigas e reject-publish faz com que novas publicações recebam nack quando há confirms, para que o produtor perceba o backpressure.
• x-dead-letter-exchange e x-dead-letter-routing-key — para onde vão as mensagens rejeitadas, expiradas ou que excederam o limite.

Um dead-letter exchange (DLX) é um exchange comum; ligar uma fila a ele dá a você um lugar onde as mensagens com falha se acumulam, com um cabeçalho x-death que descreve o motivo e quantas vezes aconteceu. A combinação de TTL e DLX também é a forma como o RabbitMQ implementa retry com atraso sem plugin: faça nack da mensagem com falha para uma fila de “espera” cujo TTL é de trinta segundos e cujo dead-letter exchange roteia de volta para a fila de trabalho.

Map<String, Object> args = Map.of(
    "x-queue-type", "quorum",
    "x-dead-letter-exchange", "orders.dlx",
    "x-delivery-limit", 5,               // quorum queues: dead-letter after 5 redeliveries
    "x-message-ttl", 86_400_000);        // one day
channel.queueDeclare("fulfilment", true, false, false, args);

Argumentos definidos na declaração ficam congelados; mudá-los exige apagar a fila. Prefira policies para tudo que possa mudar: rabbitmqctl set_policy dlx "^fulfilment$" '{"dead-letter-exchange":"orders.dlx"}' --apply-to queues aplica a mesma configuração sem redeclarar nada e pode ser atualizado em tempo real.

Administrando o broker

O plugin de gerenciamento (management) oferece uma interface web e uma API HTTP na porta 15672 (o AMQP em si fica na 5672, o TLS na 5671 e as métricas do Prometheus na 15692). Ele mostra exchanges, bindings, profundidade das filas, taxas de mensagens, conexões e channels, e permite publicar e buscar mensagens manualmente para depuração. A conta padrão guest/guest só consegue fazer login a partir de localhost; crie usuários de verdade e conceda permissões por vhost antes que qualquer outra coisa se conecte:

rabbitmqctl add_vhost shop
rabbitmqctl add_user orders 'change-me'
rabbitmqctl set_permissions -p shop orders '^orders.*' '^orders$' '^(fulfilment|orders.*)$'
#                                          configure     write       read     (regular expressions on names)
rabbitmqctl list_queues -p shop name type messages messages_unacknowledged consumers
rabbitmq-diagnostics status

Os três padrões de permissão são expressões regulares sobre os nomes dos recursos para configure (declarar e apagar), write (publicar em um exchange) e read (consumir de uma fila, fazer binding de uma fila). A ferramenta de linha de comando rabbitmqadmin encapsula a API HTTP para automatizar declarações em scripts, e o rabbitmq-diagnostics informa a saúde dos nós, os alarmes e o status do cluster. O primeiro alarme que você vai encontrar na prática é o alarme de recursos de memória ou disco: quando um nó ultrapassa seu limite máximo (high watermark), ele bloqueia todas as conexões que publicam até os consumidores colocarem a fila em dia, o que, para o publisher, parece um travamento.

Erros comuns

• Publicar com o nome de uma fila em um exchange customizado. A routing key precisa corresponder a um binding; sem binding, a mensagem é descartada e nada é registrado em log. Use mandatory com um return listener durante o desenvolvimento.
• Uma conexão por mensagem, ou um channel compartilhado entre threads. O primeiro esgota os file descriptors e o broker; o segundo corrompe o fluxo do protocolo. Uma conexão por processo, um channel por thread.
• Fila durável, mensagem transitória. Delivery mode 1 em uma fila clássica não sobrevive a um restart, mesmo que a fila sobreviva. Use o modo 2, ou quorum queues, que ignoram essa distinção.
• Sem prefetch. Um consumidor pega tudo, os outros ficam ociosos e uma queda gera milhares de reentregas. Sempre chame basicQos.
• Loops infinitos de reenfileiramento. Uma mensagem envenenada que recebe nack com requeue = true volta na mesma hora. Limite as reentregas com x-delivery-limit ou uma verificação de x-death e mande a mensagem para dead letter.
• Fixar configurações mutáveis nos argumentos da fila. TTLs, limites e nomes de DLX mudam; argumentos de fila, não. Use policies.

Principais conclusões

• Publishers enviam para exchanges com uma routing key; os bindings decidem quais filas recebem cópias; os consumidores leem das filas. Nada mais no RabbitMQ faz sentido sem esse modelo.
• Exchanges direct comparam chaves exatas, fanout copia para todos, topic aplica padrões * e #, headers compara atributos; um exchange topic por contexto com uma convenção de routing key atende à maioria dos designs.
• Quorum queues são a escolha padrão para trabalho durável; filas clássicas cobrem os usos transitórios e exclusivos; streams acrescentam logs com replay.
• Entrega at-least-once exige, em conjunto, exchange durável, fila durável, mensagem persistente, publisher confirms e acknowledgements manuais.
• O prefetch do basicQos distribui a carga de forma justa entre os consumidores e torna as quedas baratas; consumer timeouts e delivery limits impedem que mensagens travadas ou envenenadas derrubem o broker.
• Expresse TTLs, limites de tamanho e roteamento de dead letter como policies, para que possam mudar sem redeclarar as filas.