Filas e mensageria

Os exemplos consomem a fila pedidos.criados do RabbitMQ, com o cliente oficial ou o mais adotado de cada linguagem. O consumidor recebe a mensagem, processa de forma idempotente e só então confirma.

Subir um RabbitMQ para testar

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-management

O painel fica em http://localhost:15672 (usuário e senha guest). Em seguida, crie o destino das mensagens rejeitadas e uma política que as envia para lá, para todas as filas pedidos.*:

# A troca e a fila que recebem as mensagens rejeitadas
docker exec rabbitmq rabbitmqadmin declare exchange --name mortas --type fanout
docker exec rabbitmq rabbitmqadmin declare queue --name mensagens.mortas
docker exec rabbitmq rabbitmqadmin declare binding --source mortas \
  --destination-type queue --destination mensagens.mortas --routing-key ""

# A política que manda para "mortas" o que for rejeitado nas filas pedidos.*
docker exec rabbitmq rabbitmqctl set_policy mensagens-mortas "^pedidos\." \
  '{"dead-letter-exchange":"mortas"}' --apply-to queues

A política fica no intermediário, fora do código. Assim é possível mudá-la sem redeclarar as filas, o que o RabbitMQ não permite quando os argumentos de uma fila existente mudam.

Um consumidor que confirma só depois de processar

import amqp from "amqplib";

const FILA = "pedidos.criados";

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

await canal.assertQueue(FILA, { durable: true });
await canal.prefetch(10); // no máximo 10 mensagens sem confirmação

await canal.consume(FILA, async (mensagem) => {
  if (!mensagem) return;
  try {
    const evento = JSON.parse(mensagem.content.toString());
    await processarUmaVez(mensagem.properties.messageId, evento);
    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 });

import json
import os

import pika

FILA = "pedidos.criados"

conexao = pika.BlockingConnection(pika.URLParameters(os.environ["RABBITMQ_URL"]))
canal = conexao.channel()

canal.queue_declare(queue=FILA, durable=True)
canal.basic_qos(prefetch_count=10)


def ao_receber(canal, metodo, propriedades, corpo):
    try:
        evento = json.loads(corpo)
        processar_uma_vez(propriedades.message_id, evento)
        canal.basic_ack(delivery_tag=metodo.delivery_tag)
    except Exception:
        # vai para a fila de mensagens mortas
        canal.basic_nack(delivery_tag=metodo.delivery_tag, requeue=False)


canal.basic_consume(queue=FILA, on_message_callback=ao_receber, auto_ack=False)
canal.start_consuming()

package main

import (
	"context"
	"encoding/json"
	"log"
	"os"

	amqp "github.com/rabbitmq/amqp091-go"
)

const fila = "pedidos.criados"

func main() {
	conexao, err := amqp.Dial(os.Getenv("RABBITMQ_URL"))
	if err != nil {
		log.Fatal(err)
	}
	defer conexao.Close()

	canal, err := conexao.Channel()
	if err != nil {
		log.Fatal(err)
	}

	if _, err := canal.QueueDeclare(fila, true, false, false, false, nil); err != nil {
		log.Fatal(err)
	}
	if err := canal.Qos(10, 0, false); err != nil {
		log.Fatal(err)
	}

	// autoAck = false: a confirmação é manual
	mensagens, err := canal.Consume(fila, "", false, false, false, false, nil)
	if err != nil {
		log.Fatal(err)
	}

	ctx := context.Background()
	for mensagem := range mensagens {
		var evento PedidoCriado
		if err := json.Unmarshal(mensagem.Body, &evento); err != nil {
			mensagem.Nack(false, false) // permanente: vai para as mensagens mortas
			continue
		}
		if err := processarUmaVez(ctx, mensagem.MessageId, evento); err != nil {
			log.Println("falha ao processar:", err)
			mensagem.Nack(false, false)
			continue
		}
		mensagem.Ack(false)
	}
}

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class ConsumidorDePedidos {
    private static final String FILA = "pedidos.criados";
    private static final ObjectMapper JSON = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        var fabrica = new ConnectionFactory();
        fabrica.setUri(System.getenv("RABBITMQ_URL"));

        var conexao = fabrica.newConnection();
        Channel canal = conexao.createChannel();

        canal.queueDeclare(FILA, true, false, false, null);
        canal.basicQos(10);

        DeliverCallback aoReceber = (etiqueta, entrega) -> {
            long id = entrega.getEnvelope().getDeliveryTag();
            try {
                var evento = JSON.readValue(entrega.getBody(), PedidoCriado.class);
                processarUmaVez(entrega.getProperties().getMessageId(), evento);
                canal.basicAck(id, false);
            } catch (Exception erro) {
                // vai para a fila de mensagens mortas
                canal.basicNack(id, false, false);
            }
        };

        // autoAck = false: a confirmação é manual
        canal.basicConsume(FILA, false, aoReceber, etiqueta -> {});
    }
}

using System.Text.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

const string Fila = "pedidos.criados";

var fabrica = new ConnectionFactory
{
    Uri = new Uri(Environment.GetEnvironmentVariable("RABBITMQ_URL")!),
};

await using var conexao = await fabrica.CreateConnectionAsync();
await using var canal = await conexao.CreateChannelAsync();

await canal.QueueDeclareAsync(Fila, durable: true, exclusive: false, autoDelete: false);
await canal.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);

var consumidor = new AsyncEventingBasicConsumer(canal);
consumidor.ReceivedAsync += async (_, entrega) =>
{
    try
    {
        var evento = JsonSerializer.Deserialize<PedidoCriado>(entrega.Body.Span)!;
        await ProcessarUmaVez(entrega.BasicProperties.MessageId, evento);
        await canal.BasicAckAsync(entrega.DeliveryTag, multiple: false);
    }
    catch (Exception)
    {
        // vai para a fila de mensagens mortas
        await canal.BasicNackAsync(entrega.DeliveryTag, multiple: false, requeue: false);
    }
};

// autoAck = false: a confirmação é manual
await canal.BasicConsumeAsync(Fila, autoAck: false, consumer: consumidor);
await Task.Delay(Timeout.Infinite);

O que muda de uma linguagem para outra

	Biblioteca	Modelo do consumidor	Confirmação manual
Node	`amqplib`	Função chamada a cada mensagem	`{ noAck: false }`
Python	`pika`	Função chamada a cada mensagem, numa conexão bloqueante	`auto_ack=False`
Go	`amqp091-go`	Canal Go, lido num laço	Terceiro argumento de `Consume` igual a `false`
Java	`amqp-client`	`DeliverCallback`	Segundo argumento de `basicConsume` igual a `false`
C#	`RabbitMQ.Client` 7	Evento `ReceivedAsync`, totalmente assíncrono	`autoAck: false`

Três detalhes que costumam passar despercebidos:

Confirmação automática (noAck: true, auto_ack=True) remove a mensagem no momento da entrega. Se o consumidor cair durante o processamento, a mensagem se perde. Use sempre a confirmação manual, depois do trabalho feito.
prefetch limita quantas mensagens um consumidor recebe sem ter confirmado. Sem limite, um consumidor pode receber milhares de mensagens de uma vez, ficar sem memória e, ao cair, devolver todas para a fila.
No Python, a conexão bloqueante do pika não é segura para uso entre threads. Para processar em paralelo, use um processo por consumidor, ou entregue o trabalho a outra thread e confirme pela própria conexão com add_callback_threadsafe.

Nos exemplos, toda falha vai direto para a fila de mensagens mortas. Para falhas temporárias, o nível Soluções descreve as novas tentativas com atraso, que no RabbitMQ usam uma fila de espera com tempo de expiração ou o limite de entregas das filas quorum.

Processar uma vez: idempotência com o banco

A função processarUmaVez registra o identificador da mensagem numa tabela com chave única, na mesma transação do efeito. Se a mensagem chegar de novo, a inserção falha por duplicidade e o efeito não se repete.

CREATE TABLE mensagens_processadas (
  mensagem_id    TEXT PRIMARY KEY,
  processada_em  TIMESTAMPTZ NOT NULL DEFAULT now()
);

export async function processarUmaVez(mensagemId: string, evento: PedidoCriado) {
  if (!mensagemId) throw new Error("mensagem sem identificador");

  await db.transaction(async (tx) => {
    const registro = await tx.query(
      `INSERT INTO mensagens_processadas (mensagem_id)
       VALUES ($1)
       ON CONFLICT (mensagem_id) DO NOTHING`,
      [mensagemId],
    );

    if (registro.rowCount === 0) {
      return; // já foi processada antes: não faz nada
    }

    await tx.query(
      "UPDATE produtos SET reservado = reservado + $1 WHERE id = $2",
      [evento.quantidade, evento.produtoId],
    );
  });
}

Detalhe	Por quê
O registro e o efeito na mesma transação	Se o efeito falhar, o registro também é desfeito, e a mensagem pode ser processada de novo
`ON CONFLICT DO NOTHING`	Uma reentrega não gera erro, só é ignorada
O identificador vem do produtor	O produtor define o `messageId` ao publicar. Sem ele, não há como reconhecer a mesma mensagem
Limpeza periódica	A tabela cresce; registros mais antigos que o tempo máximo de reentrega podem ser apagados

Quando o efeito está fora do banco, como enviar um e-mail ou chamar uma API de pagamento, a transação não o cobre. Nesses casos, use a chave de idempotência do próprio serviço externo, quando ele oferecer, ou aceite que, numa falha rara, o efeito pode se repetir.

Publicar com identificador e persistência

Do lado do produtor, duas propriedades são essenciais: um identificador único por mensagem e a marcação como persistente, para que ela sobreviva a uma reinicialização do intermediário.

import { randomUUID } from "node:crypto";

canal.sendToQueue(
  "pedidos.criados",
  Buffer.from(JSON.stringify({ pedidoId: 981, produtoId: 7, quantidade: 2 })),
  {
    messageId: randomUUID(),
    persistent: true,
    contentType: "application/json",
  },
);

Para garantir que o intermediário recebeu a mensagem, use um canal com confirmação de publicação (createConfirmChannel no amqplib), que avisa quando a mensagem foi gravada. E, para não perder a mensagem entre a gravação no banco e a publicação, o padrão outbox descrito no nível Soluções.

Lista de revisão

Consumidor

A confirmação é manual e acontece só depois do processamento?
Existe limite de mensagens sem confirmação (prefetch)?
O processamento é idempotente, com o registro na mesma transação do efeito?
Falhas permanentes vão para a fila de mensagens mortas, sem voltar para a fila principal?
Falhas temporárias são tentadas de novo com atraso e com limite de tentativas?

Produtor

Cada mensagem tem um identificador único definido pelo produtor?
As mensagens são persistentes e as filas são duráveis?
A publicação é confirmada pelo intermediário?
A mensagem e os dados são gravados juntos, pelo padrão outbox?

Operação

Alguém é avisado quando a fila de mensagens mortas recebe mensagens?
O tamanho da fila e a idade da mensagem mais antiga são monitorados?
Os consumidores toleram mensagens fora de ordem?

Subir um RabbitMQ para testar

Um consumidor que confirma só depois de processar

O que muda de uma linguagem para outra

Processar uma vez: idempotência com o banco

Publicar com identificador e persistência

Lista de revisão

Prompts para aprofundar

Construir um consumidor confiável

Ordem das mensagens por chave

Revisar código de mensageria