DevGuideTK

Publicação e assinatura

Um evento, vários consumidores independentes, e o desacoplamento que isso traz.

Os exemplos usam o RabbitMQ com uma troca do tipo topic chamada pedidos. O serviço de pedidos publica os eventos pedido.criado, pedido.pago e pedido.cancelado; o serviço de estoque assina dois deles, com uma fila durável que leva o nome do serviço.

Criar as assinaturas na infraestrutura

As assinaturas são criadas antes de o publicador começar, para que nenhum evento seja publicado numa troca sem destino. Com o rabbitmqadmin 2, que acompanha o RabbitMQ 4:

rabbitmqadmin declare exchange --name pedidos --type topic

# A assinatura do serviço de estoque: uma fila e duas ligações
rabbitmqadmin declare queue --name estoque.pedidos
rabbitmqadmin declare binding --source pedidos --destination-type queue \
  --destination estoque.pedidos --routing-key pedido.criado
rabbitmqadmin declare binding --source pedidos --destination-type queue \
  --destination estoque.pedidos --routing-key pedido.cancelado

# A assinatura do serviço de análise: todos os eventos de pedido
rabbitmqadmin declare queue --name analise.pedidos
rabbitmqadmin declare binding --source pedidos --destination-type queue \
  --destination analise.pedidos --routing-key "pedido.#"

O mesmo pode ficar num arquivo de definições do RabbitMQ, ou na ferramenta de infraestrutura como código da equipe. Os assinantes também declaram a troca, a fila e as ligações ao iniciar, como garantia; declarar algo que já existe com os mesmos argumentos não tem efeito.

O assinante

Cada instância do serviço de estoque consome a mesma fila estoque.pedidos. O RabbitMQ divide as mensagens entre elas, e cada evento é processado uma vez pelo serviço.

import amqp from "amqplib";

const TROCA = "pedidos";
const FILA = "estoque.pedidos"; // nome do serviço, não da instância

const conexao = await amqp.connect(process.env.RABBITMQ_URL!);
const canal = await conexao.createChannel();

await canal.assertExchange(TROCA, "topic", { durable: true });
await canal.assertQueue(FILA, { durable: true });
await canal.bindQueue(FILA, TROCA, "pedido.criado");
await canal.bindQueue(FILA, TROCA, "pedido.cancelado");
await canal.prefetch(10);

await canal.consume(FILA, async (mensagem) => {
  if (!mensagem) return;
  try {
    const evento = JSON.parse(mensagem.content.toString());
    switch (evento.tipo) {
      case "pedido.criado":
        await reservarItens(evento.id, evento.dados);
        break;
      case "pedido.cancelado":
        await liberarItens(evento.id, evento.dados);
        break;
      // outros tipos são ignorados
    }
    canal.ack(mensagem);
  } catch (erro) {
    console.error("falha ao processar", erro);
    canal.nack(mensagem, false, false); // vai para a fila de mensagens mortas
  }
}, { noAck: false });

A rejeição sem devolver à fila (requeue: false) só leva a mensagem para a fila de mensagens mortas se houver uma política configurada, como no tópico de filas. Sem ela, a mensagem é descartada.

O que muda de uma linguagem para outra

BibliotecaCampo desconhecido no JSON
NodeamqplibJSON.parse aceita tudo; com zod, z.object descarta os campos a mais por padrão
Pythonpikajson.loads devolve um dicionário; no pydantic, o padrão é ignorar campos a mais
Goamqp091-goIgnorado por padrão; DisallowUnknownFields torna erro
Javaamqp-clientO ObjectMapper do Jackson 2 falha por padrão; o Spring Boot já desliga essa opção
C#RabbitMQ.Client 7Ignorado por padrão no System.Text.Json

A coluna de campos desconhecidos é a que importa para a evolução dos eventos. Um assinante Java com o ObjectMapper padrão é derrubado pela primeira mudança compatível do publicador, porque toda mensagem com o campo novo falha e vai para as mensagens mortas.

O erro do tutorial: uma fila por instância

O tutorial oficial do RabbitMQ sobre publicação e assinatura usa uma fila sem nome, exclusiva da conexão:

# Certo para o tutorial, errado para eventos de negócio
resultado = canal.queue_declare(queue="", exclusive=True)
fila = resultado.method.queue  # nome gerado: amq.gen-...
canal.queue_bind(exchange="pedidos", queue=fila, routing_key="pedido.criado")

Esse desenho serve para o que o tutorial mostra, a distribuição de logs para quem estiver olhando. Num serviço de negócio, ele tem dois defeitos:

DefeitoConsequência
Cada instância cria a sua filaCom três réplicas, cada evento é processado três vezes
A fila exclusiva some quando a conexão caiOs eventos publicados durante uma reinicialização se perdem

A correção é a do exemplo acima: um nome fixo, igual em todas as instâncias, com durable: true e sem exclusive.

O publicador

O publicador monta o envelope, publica na troca com o tipo do evento como chave de roteamento e espera a confirmação do intermediário.

import amqp from "amqplib";
import { randomUUID } from "node:crypto";

const conexao = await amqp.connect(process.env.RABBITMQ_URL!);
const canal = await conexao.createConfirmChannel();
await canal.assertExchange("pedidos", "topic", { durable: true });

// "mandatory" devolve a mensagem quando nenhuma fila a recebe
canal.on("return", (mensagem) => {
  console.error("evento sem assinante:", mensagem.fields.routingKey);
});

export async function publicar(tipo: string, dados: unknown) {
  const evento = {
    id: randomUUID(),
    tipo,
    versao: 1,
    ocorridoEm: new Date().toISOString(),
    origem: "pedidos",
    dados,
  };

  canal.publish("pedidos", tipo, Buffer.from(JSON.stringify(evento)), {
    messageId: evento.id,
    type: tipo,
    contentType: "application/json",
    persistent: true,
    mandatory: true,
  });
  await canal.waitForConfirms();
}

await publicar("pedido.criado", { pedidoId: 981, itens: [{ produtoId: 7, quantidade: 2 }] });
DetalhePor quê
A chave de roteamento é o tipo do eventoOs assinantes filtram por ela (pedido.criado, pedido.#)
id no envelope e em messageIdCada assinante descarta reentregas pelo mesmo identificador
persistent: trueA mensagem sobrevive a uma reinicialização do intermediário, desde que a fila seja durável
mandatory: trueUma mensagem que não chega a nenhuma fila volta ao publicador, em vez de sumir sem aviso
Canal com confirmaçãowaitForConfirms só termina quando o intermediário aceitou a mensagem

O return não é um erro de negócio: um evento pode não ter assinantes. Mas, se um tipo que deveria ter assinantes começa a voltar, uma ligação foi removida ou nunca foi criada, e o aviso aparece no log em vez de virar dados perdidos.

Na aplicação real, publicar não é chamada logo depois de gravar o pedido, e sim pelo processo que lê a tabela de saída do padrão outbox.

Idempotência por assinante

Cada assinante registra os eventos que já processou, como no tópico de filas. Se vários serviços compartilham o mesmo banco, a chave inclui o nome do assinante, porque o mesmo evento precisa ser processado uma vez por cada um:

CREATE TABLE eventos_processados (
  assinante     TEXT NOT NULL,
  evento_id     TEXT NOT NULL,
  processado_em TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (assinante, evento_id)
);
-- Dentro da mesma transação do efeito
INSERT INTO eventos_processados (assinante, evento_id)
VALUES ('estoque', $1)
ON CONFLICT DO NOTHING;
-- 0 linhas inseridas: o evento já foi processado, não faz nada

O mesmo desenho no Kafka

No Kafka, o papel da fila nomeada é do group.id. A configuração do assinante de estoque, com os nomes de propriedade aceitos tanto pelo cliente Java quanto pelos clientes baseados no librdkafka (Python, Go, C# e Node):

# O nome do serviço, igual em todas as instâncias
group.id=estoque
# Um grupo novo começa do início do tópico, e não só pelos eventos novos
auto.offset.reset=earliest
# A posição é confirmada pelo código, depois de processar
enable.auto.commit=false

As instâncias do grupo dividem as partições do tópico. Use o id do pedido como chave da mensagem, para que os eventos de um mesmo pedido caiam na mesma partição e cheguem em ordem.

Lista de revisão

Assinatura

  • Cada serviço tem uma assinatura com nome fixo, igual em todas as instâncias?
  • A assinatura é durável, e não exclusiva ou temporária?
  • As assinaturas existem antes de o publicador começar a publicar?
  • Dois serviços diferentes nunca compartilham a mesma assinatura (fila ou group.id)?

Evento

  • O nome está no passado e descreve um fato, não um comando?
  • O envelope tem id, tipo, versao e ocorridoEm?
  • O conteúdo traz o que a maioria dos assinantes precisa, sem despejar a tabela inteira?
  • Mudanças no formato só acrescentam campos, ou publicam uma versão nova ao lado da antiga?

Assinante

  • Campos e tipos de evento desconhecidos são ignorados, e não tratados como erro?
  • O processamento é idempotente, com o registro na mesma transação do efeito?
  • A confirmação acontece só depois de processar, com limite de mensagens sem confirmação?
  • Há fila de mensagens mortas e alguém é avisado quando ela recebe algo?

Publicador

  • Os eventos saem pelo padrão outbox, e não logo depois da gravação?
  • A publicação é confirmada pelo intermediário?
  • Um evento que não chega a nenhuma assinatura gera um aviso?
  • Existe um catálogo de quem assina cada tipo de evento?
← Anterior: SoluçõesPróximo tópico: Escalabilidade →

Prompts para aprofundar

Ao abrir no ChatGPT ou no Claude, o prompt é enviado ao serviço escolhido pela URL e também fica copiado na sua área de transferência.

  • Construir publicador e assinantes

    Quero construir um publicador e dois assinantes de eventos, testando cada comportamento.

Pergunte qual linguagem e qual intermediário eu uso.

Vamos em etapas. Em cada uma, deixe que eu tente e depois mostre a sua versão:
1. Subir o intermediário com Docker e criar o tópico e as duas assinaturas pela linha de comando, antes de qualquer código.
2. Um publicador que monta o envelope do evento e espera a confirmação do intermediário.
3. Um assinante com confirmação manual, que trata dois tipos de evento e ignora os demais.
4. Subir três instâncias do mesmo assinante e confirmar que cada evento é processado uma vez.
5. Repetir com uma fila exclusiva por instância e observar o processamento triplicado.
6. Derrubar um assinante, publicar eventos e confirmar que eles esperam por ele.
7. Acrescentar um campo novo no evento e confirmar que os assinantes continuam funcionando.

Em cada etapa, diga o que observar no painel ou nos logs.
    Abrir no ChatGPTAbrir no Claude

  • Migrar um evento para uma versão nova

    Preciso fazer uma mudança incompatível num evento que vários serviços assinam, como trocar o campo total, em centavos, por um objeto com valor e moeda.

Pergunte qual linguagem e qual intermediário eu uso, e quantos serviços assinam o evento.

Conduza o plano por etapas, com código:
1. Como publicar a versão nova ao lado da antiga, pelo tipo do evento ou por um campo de versão.
2. Como cada assinante passa a ler a versão nova, um de cada vez.
3. Como descobrir quais assinantes ainda dependem da versão antiga.
4. Quando e como parar de publicar a versão antiga.
5. Como um registro de esquemas teria impedido a mudança incompatível de ser publicada por engano.

Em cada etapa, pergunte o que pode dar errado antes de seguir.
    Abrir no ChatGPTAbrir no Claude

  • Revisar código de publicação e assinatura

    Vou colar abaixo código que publica ou assina eventos. Revise com foco em entrega, duplicidade e evolução.

Verifique, nesta ordem, e para cada item diga se está correto, incorreto ou ausente, citando a linha:
1. A assinatura tem nome fixo, igual em todas as instâncias, e é durável? Há filas exclusivas, nomes gerados ou group.id com valor aleatório ou nome da máquina?
2. Dois serviços diferentes podem estar usando a mesma assinatura?
3. A assinatura existe antes de o publicador começar, ou só é criada quando o assinante inicia?
4. Há uso de um mecanismo efêmero, como o Redis Pub/Sub, para eventos que não podem se perder?
5. O assinante ignora campos e tipos de evento desconhecidos?
6. O assinante é idempotente e confirma só depois de processar?
7. O evento tem id, tipo, versão e data do fato? O nome descreve um fato ou um comando?
8. A publicação é confirmada, e há risco de gravar no banco e não publicar?

Para cada problema, descreva o cenário concreto em que ele perde, duplica ou quebra o processamento de eventos, e mostre a correção. Não reescreva o que já está correto.

Código:
[cole aqui]
    Abrir no ChatGPTAbrir no Claude