Message Queues

Uma message queue é um buffer persistente que fica entre produtor e consumidor — o produtor deposita mensagens sem saber quem vai processá-las, e o consumidor processa no seu próprio ritmo sem saber quem as produziu. Esse desacoplamento temporal é o benefício central: se o consumidor está lento, sobrecarregado ou indisponível, as mensagens ficam na fila esperando — sem derrubar o produtor. É a fundação de toda comunicação assíncrona entre serviços.

Anatomia de uma message queue

Os componentes fundamentais são consistentes entre brokers diferentes:

• Producer/Publisher: envia mensagens para a fila sem aguardar processamento
• Broker: armazena as mensagens persistentemente e as entrega aos consumidores
• Queue: sequência ordenada de mensagens aguardando consumo
• Consumer/Subscriber: recebe mensagens da fila para processar
• Message: envelope com payload (corpo), headers (metadados) e routing key
• Acknowledgement (ACK): confirmação de que o consumidor processou a mensagem com sucesso
• Negative Acknowledgement (NACK): rejeição — a mensagem volta para a fila ou vai para DLQ

Modelos de entrega

Point-to-point (queue)

Uma mensagem é entregue a exatamente um consumidor — mesmo que vários estejam ouvindo a mesma fila. O broker faz load balancing entre os consumidores disponíveis. Modelo correto para tarefas que devem ser executadas uma única vez: processar um pagamento, gerar um relatório, enviar um e-mail.

# Point-to-point: 1 mensagem → 1 consumidor
Producer → [Queue] → Consumer A  (ou Consumer B, não ambos)
                   → Consumer B

# Se Consumer A trava após receber mas antes de ACKar,
# o broker reentrega para Consumer B após o timeout

Publish/Subscribe (topic/exchange)

Uma mensagem é entregue a todos os consumidores subscritos. Modelo correto para eventos que múltiplos sistemas precisam conhecer: pedido criado → notificar estoque, notificar faturamento, atualizar dashboard, enviar e-mail de confirmação.

# Pub/Sub: 1 mensagem → N consumidores
Producer → [Topic/Exchange] → Queue de Estoque    → Consumer Estoque
                            → Queue de Faturamento → Consumer Faturamento
                            → Queue de Dashboard   → Consumer Dashboard

Garantias de entrega

A garantia de entrega determina quantas vezes uma mensagem pode ser recebida pelo consumidor — e é um trade-off entre complexidade e confiabilidade:

Acknowledgements e visibilidade

O mecanismo de ACK é o coração da confiabilidade das filas. Quando um consumidor recebe uma mensagem, ela fica "invisível" para outros consumidores por um período (visibility timeout). Se o ACK não chega dentro do timeout, o broker assume que o consumidor falhou e reentrega a mensagem.

# Ciclo de vida de uma mensagem
1. Producer publica mensagem → mensagem fica em estado READY
2. Consumer recebe mensagem → mensagem fica em INVISIBLE/PROCESSING (timeout: 30s)
3a. Consumer processa com sucesso → envia ACK → mensagem é DELETADA da fila
3b. Consumer falha → envia NACK (ou timeout expira) → mensagem volta para READY
3c. Após N tentativas → mensagem vai para Dead Letter Queue

# Visibility timeout no AWS SQS
# Se o processamento leva 25s, configure timeout para 60s (2-3x o tempo esperado)
# Visibility timeout muito curto → reentregas desnecessárias
# Visibility timeout muito longo → mensagens travadas por muito tempo se consumidor falhar

Dead Letter Queue (DLQ)

Uma DLQ é uma fila especial que recebe mensagens que falharam repetidamente. Em vez de tentar infinitamente ou descartar, o broker move a mensagem para a DLQ após N tentativas. Isso evita que mensagens "poison" (que sempre falham) bloqueiem processamento de mensagens saudáveis.

# RabbitMQ — configurar DLQ via políticas
rabbitmqadmin declare queue \
  name=orders.processing \
  durable=true \
  arguments='{"x-dead-letter-exchange": "dlx", "x-dead-letter-routing-key": "orders.dead", "x-message-ttl": 86400000, "x-max-length": 100000}'

rabbitmqadmin declare exchange name=dlx type=direct durable=true
rabbitmqadmin declare queue name=orders.dead durable=true
rabbitmqadmin declare binding source=dlx destination=orders.dead routing_key=orders.dead

# AWS SQS — configurar DLQ via redrive policy
{
  "deadLetterTargetArn": "arn:aws:sqs:us-east-1:123456789:orders-dead",
  "maxReceiveCount": 3  // após 3 falhas, vai para DLQ
}

RabbitMQ em profundidade

RabbitMQ implementa o protocolo AMQP 0-9-1 e é um dos brokers de fila mais usados no mundo. Seu modelo de roteamento usa exchanges como roteadores entre producers e queues — os producers nunca publicam diretamente em filas, mas em exchanges com routing keys.

Tipos de exchange

# RabbitMQ — topologia completa para sistema de pedidos
# Exchange principal para eventos de pedidos
rabbitmqadmin declare exchange name=orders type=topic durable=true

# Filas por subsistema com bindings para routing keys específicas
rabbitmqadmin declare queue name=inventory.updates durable=true \
  arguments='{"x-dead-letter-exchange": "dlx"}'
rabbitmqadmin declare queue name=billing.events durable=true \
  arguments='{"x-dead-letter-exchange": "dlx"}'
rabbitmqadmin declare queue name=notifications.orders durable=true \
  arguments='{"x-dead-letter-exchange": "dlx"}'

# Bindings: routing key patterns
# orders.# → todas as queues (qualquer evento de pedido)
# orders.created → apenas eventos de criação
# orders.*.us → eventos de pedidos americanos

rabbitmqadmin declare binding \
  source=orders destination=inventory.updates routing_key="orders.created"
rabbitmqadmin declare binding \
  source=orders destination=billing.events routing_key="orders.#"
rabbitmqadmin declare binding \
  source=orders destination=notifications.orders routing_key="orders.created"

Prefetch e QoS

Por padrão, RabbitMQ pode enviar infinitas mensagens a um consumidor antes de receber ACKs — o que pode sobrecarregar consumidores lentos. O basicQos(prefetchCount) limita quantas mensagens o broker envia antes de receber ACK, implementando backpressure.

# prefetchCount = 1: envia uma mensagem por vez — fair dispatch
# Mais lento (mais round-trips de ACK), mas distribui carga uniformemente
channel.basic_qos(prefetch_count=1)

# prefetchCount = 10-50: bom equilíbrio entre throughput e fairness
channel.basic_qos(prefetch_count=10)

# prefetchCount = 0 (default): sem limite — risco de memory pressure no consumidor

Amazon SQS

SQS é um serviço de fila gerenciado da AWS — sem infraestrutura para operar, alta disponibilidade nativa, e integração com o ecossistema AWS (Lambda, SNS, EventBridge). Dois tipos:

• Standard Queue: throughput ilimitado, at-least-once delivery, melhor esforço de ordenação
• FIFO Queue: exatamente-uma-vez dentro de um grupo de mensagens, ordenação garantida, throughput limitado a 3000 msg/s (com batching)

# SQS — padrão de uso com Lambda consumer
# 1. Criar fila com DLQ
aws sqs create-queue \
  --queue-name orders-processing \
  --attributes '{
    "VisibilityTimeout": "60",
    "MessageRetentionPeriod": "86400",
    "RedrivePolicy": "{\"deadLetterTargetArn\":\"arn:aws:sqs:...:orders-dead\",\"maxReceiveCount\":\"3\"}"
  }'

# 2. Lambda trigger — SQS invoca Lambda com batch de mensagens
# Configurar Event Source Mapping:
aws lambda create-event-source-mapping \
  --function-name process-order \
  --event-source-arn arn:aws:sqs:us-east-1:123456789:orders-processing \
  --batch-size 10 \
  --maximum-batching-window-in-seconds 5

# 3. Lambda retorna falhas parciais (não reprocessar mensagens que já funcionaram)
# Response da Lambda para SQS:
{
  "batchItemFailures": [
    { "itemIdentifier": "msg-id-que-falhou" }
    // mensagens NÃO listadas → consideradas sucesso → deletadas
    // mensagens listadas → voltam para fila ou DLQ
  ]
}

Patterns de uso de filas

Worker queue — distribuição de trabalho

# Múltiplos workers consumindo uma fila — horizontal scaling
                     → Worker 1 (processa pedido 1)
Queue de pedidos    → Worker 2 (processa pedido 2)
                     → Worker 3 (processa pedido 3)

# Escalar workers conforme tamanho da fila:
# queue.depth > 1000 → adicionar workers
# queue.depth < 100  → remover workers
# Com Kubernetes HPA + KEDA (Kubernetes Event-Driven Autoscaling)

Request/Reply via fila — RPC assíncrono

# Padrão correlation ID + reply-to queue
# Útil quando o resultado é necessário mas a latência pode ser alta

# 1. Producer envia request com correlation_id e reply_to
{
  "payload": {"order_id": "abc"},
  "correlation_id": "uuid-123",
  "reply_to": "responses.api-gateway-1"  # fila privada do producer
}

# 2. Worker processa e envia response para reply_to
{
  "payload": {"status": "approved"},
  "correlation_id": "uuid-123"  # producer correlaciona com request original
}

# 3. Producer aguarda na fila de responses com timeout
# Desvantagem: complexidade alta — prefira REST/gRPC para request-response

Competing consumers — escalonamento horizontal

O padrão mais comum: múltiplos consumidores idênticos lendo da mesma fila. O broker distribui as mensagens entre eles. Para escalar, adicionar mais consumidores — sem alterar producers ou a fila.

Comparação por linguagem

Publicação e consumo de mensagens com RabbitMQ em cada linguagem, incluindo confirmações, DLQ e tratamento de falhas.

// MassTransit abstrai o broker — mesmo código para RabbitMQ, SQS, Azure Service Bus

// Definir a mensagem (contrato)
public record OrderCreated(
    string OrderId,
    string CustomerId,
    decimal TotalAmount,
    DateTimeOffset CreatedAt);

// Consumer — implementa IConsumer<T>
public class OrderCreatedConsumer : IConsumer<OrderCreated>
{
    private readonly IInventoryService _inventory;
    private readonly ILogger<OrderCreatedConsumer> _logger;

    public OrderCreatedConsumer(IInventoryService inventory, ILogger<OrderCreatedConsumer> logger)
    {
        _inventory = inventory;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var msg = context.Message;
        _logger.LogInformation("Processando pedido {OrderId}", msg.OrderId);

        try
        {
            await _inventory.ReserveStockAsync(msg.OrderId, context.CancellationToken);
            // ACK implícito — se o método retornar sem exceção, MassTransit faz ACK
        }
        catch (StockUnavailableException ex)
        {
            // Falha de negócio — não retentável, mover para DLQ imediatamente
            _logger.LogWarning("Estoque indisponível para {OrderId}", msg.OrderId);
            throw new Exception($"Estoque indisponível: {ex.Message}");
            // MassTransit faz NACK → após retry_count → DLQ
        }
    }
}

// Program.cs — configuração
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderCreatedConsumer>()
        .Endpoint(e =>
        {
            e.Name = "inventory.order-created";
            e.PrefetchCount = 10;
        });

    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host("rabbitmq://rabbitmq:5672", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        // Configurar DLQ automático
        cfg.ReceiveEndpoint("inventory.order-created", e =>
        {
            e.ConfigureDeadLetterQueueDeadLetterTransport();
            e.ConfigureDeadLetterQueueErrorTransport();
            e.UseMessageRetry(r =>
                r.Exponential(3, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(5)));
            e.ConfigureConsumer<OrderCreatedConsumer>(ctx);
        });
    });
});

// Publicar evento
public class OrderService
{
    private readonly IPublishEndpoint _bus;

    public OrderService(IPublishEndpoint bus) => _bus = bus;

    public async Task<Order> CreateOrderAsync(CreateOrderCommand cmd, CancellationToken ct)
    {
        var order = await _repo.CreateAsync(cmd, ct);

        // Publicar evento — MassTransit roteia para o exchange correto
        await _bus.Publish(new OrderCreated(
            order.Id, order.CustomerId, order.Total, DateTimeOffset.UtcNow), ct);

        return order;
    }
}

import asyncio
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
import aio_pika
from aio_pika import DeliveryMode, Message

logger = logging.getLogger(__name__)

@dataclass
class OrderCreated:
    order_id: str
    customer_id: str
    total_amount: float
    created_at: str

# Producer
async def publish_order_created(
    channel: aio_pika.Channel,
    order: OrderCreated,
) -> None:
    message = Message(
        body=json.dumps({
            "order_id": order.order_id,
            "customer_id": order.customer_id,
            "total_amount": order.total_amount,
            "created_at": order.created_at,
        }).encode(),
        delivery_mode=DeliveryMode.PERSISTENT,  # sobrevive a restart do broker
        content_type="application/json",
        headers={
            "event_type": "order.created",
            "version": "1",
        },
    )
    # Publicar no exchange topic — routing key define o roteamento
    await channel.default_exchange.publish(
        message, routing_key="orders.created"
    )
    logger.info("Evento order.created publicado: %s", order.order_id)

# Consumer com retry e DLQ
async def consume_order_created(connection: aio_pika.Connection) -> None:
    channel = await connection.channel()
    await channel.set_qos(prefetch_count=10)  # processar 10 por vez

    # Declarar DLX e DLQ
    dlx = await channel.declare_exchange(
        "dlx", aio_pika.ExchangeType.DIRECT, durable=True
    )
    dlq = await channel.declare_queue(
        "inventory.order-created.dead", durable=True
    )
    await dlq.bind(dlx, routing_key="inventory.order-created")

    # Fila principal com DLQ configurado
    queue = await channel.declare_queue(
        "inventory.order-created",
        durable=True,
        arguments={
            "x-dead-letter-exchange": "dlx",
            "x-dead-letter-routing-key": "inventory.order-created",
            "x-max-delivery-attempts": 3,
        },
    )
    await queue.bind("orders", routing_key="orders.created")

    async def handle_message(message: aio_pika.IncomingMessage) -> None:
        async with message.process(requeue=False):  # NACK sem requeue → DLQ
            try:
                data = json.loads(message.body)
                order = OrderCreated(**data)

                # Verificar se é reentrega (idempotência)
                delivery_count = message.headers.get("x-death", [{}])[0].get("count", 0)
                if delivery_count > 0:
                    logger.warning(
                        "Reentrega #%d para pedido %s",
                        delivery_count, order.order_id
                    )

                await process_inventory(order)
                # ACK automático ao sair do context manager sem exceção

            except StockUnavailableError as e:
                logger.error("Estoque indisponível: %s", e)
                raise  # NACK → DLQ após max attempts
            except Exception as e:
                logger.exception("Erro ao processar: %s", e)
                raise  # NACK → retry → DLQ

    await queue.consume(handle_message)
    logger.info("Consumindo fila inventory.order-created...")
    await asyncio.Future()  # aguardar indefinidamente

async def main():
    connection = await aio_pika.connect_robust(
        "amqp://guest:guest@rabbitmq:5672/",
        # Reconexão automática em caso de queda do broker
    )
    await consume_order_created(connection)

package messaging

import (
    "context"
    "encoding/json"
    "fmt"
    "log/slog"
    "time"

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

type OrderCreated struct {
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    Total      float64   `json:"total_amount"`
    CreatedAt  time.Time `json:"created_at"`
}

// Publisher
type Publisher struct {
    conn    *amqp.Connection
    channel *amqp.Channel
}

func NewPublisher(url string) (*Publisher, error) {
    conn, err := amqp.Dial(url)
    if err != nil {
        return nil, fmt.Errorf("conectar ao RabbitMQ: %w", err)
    }
    ch, err := conn.Channel()
    if err != nil {
        return nil, fmt.Errorf("abrir channel: %w", err)
    }
    // Confirm mode — garantia de que o broker recebeu a mensagem
    if err := ch.Confirm(false); err != nil {
        return nil, fmt.Errorf("ativar confirm mode: %w", err)
    }
    return &Publisher{conn: conn, channel: ch}, nil
}

func (p *Publisher) PublishOrderCreated(ctx context.Context, order OrderCreated) error {
    body, err := json.Marshal(order)
    if err != nil {
        return fmt.Errorf("serializar mensagem: %w", err)
    }

    // PublishWithDeferredConfirmWithContext — aguarda confirmação do broker
    confirm, err := p.channel.PublishWithDeferredConfirmWithContext(ctx,
        "orders",        // exchange
        "orders.created", // routing key
        true,            // mandatory — retorna erro se não há binding
        false,           // immediate
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent, // sobrevive a restart
            Timestamp:    time.Now(),
            Body:         body,
            Headers: amqp.Table{
                "event_type": "order.created",
                "version":    "1",
            },
        },
    )
    if err != nil {
        return fmt.Errorf("publicar mensagem: %w", err)
    }

    // Aguardar ACK do broker (confirm mode)
    if !confirm.Wait() {
        return fmt.Errorf("broker não confirmou a mensagem")
    }

    slog.Info("evento publicado", "order_id", order.OrderID)
    return nil
}

// Consumer com retry e reconexão
type Consumer struct {
    url     string
    handler func(context.Context, OrderCreated) error
}

func (c *Consumer) Start(ctx context.Context) error {
    for {
        if err := c.consume(ctx); err != nil {
            slog.Error("erro no consumer, reconectando em 5s", "err", err)
            select {
            case <-ctx.Done():
                return ctx.Err()
            case <-time.After(5 * time.Second):
            }
        }
    }
}

func (c *Consumer) consume(ctx context.Context) error {
    conn, err := amqp.Dial(c.url)
    if err != nil {
        return err
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        return err
    }
    defer ch.Close()

    ch.Qos(10, 0, false) // prefetch = 10

    deliveries, err := ch.Consume(
        "inventory.order-created",
        "",    // consumer tag gerado automaticamente
        false, // autoAck=false — ACK manual
        false, false, false, nil,
    )
    if err != nil {
        return err
    }

    for {
        select {
        case <-ctx.Done():
            return nil
        case d, ok := <-deliveries:
            if !ok {
                return fmt.Errorf("channel fechado")
            }
            c.processDelivery(ctx, d)
        }
    }
}

func (c *Consumer) processDelivery(ctx context.Context, d amqp.Delivery) {
    var order OrderCreated
    if err := json.Unmarshal(d.Body, &order); err != nil {
        slog.Error("mensagem inválida, movendo para DLQ", "err", err)
        d.Nack(false, false) // requeue=false → DLQ
        return
    }

    if err := c.handler(ctx, order); err != nil {
        slog.Error("falha ao processar", "order_id", order.OrderID, "err", err)
        d.Nack(false, false) // NACK → DLQ após max-attempts
        return
    }

    d.Ack(false) // sucesso → deletar da fila
}

Exercícios práticos

1. Configure um RabbitMQ local com Docker e crie a topologia: exchange topic "orders", três filas (inventory, billing, notifications) com bindings para "orders.#". Publique eventos com routing keys diferentes e verifique o roteamento no Management UI. Critério: Management UI mostra exchange "orders" com os 3 bindings; publicar "orders.created" entrega nas 3 filas; publicar "orders.cancelled" entrega somente nas filas com binding "orders.#"; reiniciar o container com volume persistido não perde mensagens enfileiradas.
2. Implemente DLQ completa: configure maxReceiveCount=3, crie a DLQ, e escreva um consumidor que falha nas 3 primeiras tentativas. Implemente também um script de replay que reprocessa mensagens da DLQ. Critério: após 3 NACKs, a mensagem aparece na DLQ visível no Management UI com header x-death mostrando histórico de tentativas; script de replay republica a mensagem na fila original; segunda execução (sem falha) processa com sucesso e remove da DLQ.
3. Meça o impacto do prefetchCount: consuma uma fila com 1000 mensagens onde cada processamento leva 10ms simulado. Compare throughput e distribuição de carga entre 3 workers com prefetch=1, prefetch=10, prefetch=100. Critério: com prefetch=1, distribuição entre workers ≤10% de desvio e throughput total calculado; com prefetch=100, throughput ≥3× maior mas distribuição com desvio >30% (um worker acumula); tabela com valores numéricos reais medidos.
4. Implemente idempotência: adicione um campo idempotency_key à mensagem e armazene keys processadas em Redis (TTL de 30 segundos para facilitar o teste). Critério: publicar a mesma mensagem (mesmo idempotency_key) duas vezes em sequência: apenas uma operação de negócio ocorre (verificável via log ou banco); após o TTL de 30s, publicar novamente a mensagem resulta em reprocessamento normal (segunda janela de deduplicação); throughput não cai mais que 5% vs consumidor sem Redis.
5. Configure SQS FIFO com message group IDs para garantir ordenação por customer_id: 10 pedidos do cliente A e 10 do cliente B enviados intercalados. Critério: pedidos do cliente A chegam ao consumidor em ordem sequencial (sequence_number 1-10); pedidos de A e B são processados em paralelo (timestamps de processamento sobrepostos); publicar pedido com deduplication_id já usado na janela de 5 minutos não gera duplicata no consumidor.