Mensageria Prática em Go: RabbitMQ, Apache Kafka e Resiliência

Em sistemas distribuídos e arquiteturas de microsserviços, a comunicação puramente síncrona baseada em HTTP REST ou gRPC cria acoplamento temporal: se o serviço de faturamento estiver indisponível ou enfrentando lentidão, o checkout do cliente falha imediatamente.

A Mensageria Assíncrona e o paradigma Event-Driven Architecture (EDA) desacoplam produtores e consumidores no tempo e no espaço. O Go, graças ao seu modelo de concorrência leve baseado em goroutines e canais, é uma das linguagens mais eficientes do mundo para produzir, transformar e consumir fluxos massivos de mensagens com baixíssimo consumo de memória.

1. RabbitMQ vs Apache Kafka: Qual Escolher?

Embora ambos sejam frequentemente classificados como “sistemas de mensageria”, RabbitMQ e Kafka possuem modelos conceituais completamente distintos:

Tabela Comparativa de Engenharia:

2. RabbitMQ na Prática com amqp091-go

O driver oficial da comunidade é o github.com/rabbitmq/amqp091-go.

2.1 Publicação Segura de Mensagens

package rabbitmq

import (
	"context"
	"time"

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

type Producer struct {
	conn    *amqp.Connection
	channel *amqp.Channel
}

func NewProducer(amqpURI string) (*Producer, error) {
	conn, err := amqp.Dial(amqpURI)
	if err != nil {
		return nil, err
	}

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

	// Declara uma Exchange do tipo "topic" ou "direct"
	err = ch.ExchangeDeclare(
		"orders.v1", // nome da exchange
		"topic",     // tipo
		true,        // durable (persiste a reinicializações do RabbitMQ)
		false,       // auto-deleted
		false,       // internal
		false,       // no-wait
		nil,         // argumentos
	)
	if err != nil {
		return nil, err
	}

	return &Producer{conn: conn, channel: ch}, nil
}

func (p *Producer) PublishOrderCreated(ctx context.Context, routingKey string, payload []byte) error {
	return p.channel.PublishWithContext(ctx,
		"orders.v1", // exchange
		routingKey,  // routing key (ex: "orders.br.created")
		false,       // mandatory
		false,       // immediate
		amqp.Publishing{
			DeliveryMode: amqp.Persistent, // Grava mensagem em disco no broker
			ContentType:  "application/json",
			Timestamp:    time.Now().UTC(),
			Body:         payload,
		},
	)
}

2.2 Consumidor com Manual ACK e Dead Letter Queue (DLQ)

package rabbitmq

import (
	"context"
	"log"

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

func StartConsumer(ctx context.Context, ch *amqp.Channel) error {
	// 1. Configura QoS: limita a quantidade de mensagens que o RabbitMQ envia antes do ACK
	// Sem QoS, o RabbitMQ sobrecarrega a goroutine despejando milhares de mensagens na memória
	if err := ch.Qos(10, 0, false); err != nil {
		return err
	}

	// 2. Declara fila com Dead Letter Exchange para isolar mensagens envenenadas
	args := amqp.Table{
		"x-dead-letter-exchange": "orders.dlx",
	}
	q, err := ch.QueueDeclare("orders.process_payment", true, false, false, false, args)
	if err != nil {
		return err
	}

	msgs, err := ch.Consume(
		q.Name,
		"payment-worker-1", // consumer tag
		false,              // auto-ack = FALSE (Manual ACK mandatório)
		false,              // exclusive
		false,              // no-local
		false,              // no-wait
		nil,
	)
	if err != nil {
		return err
	}

	go func() {
		for {
			select {
			case <-ctx.Done():
				log.Println("Consumidor RabbitMQ encerrando loop...")
				return
			case d, ok := <-msgs:
				if !ok {
					return
				}

				// Processa a mensagem
				err := processPayment(ctx, d.Body)
				if err != nil {
					log.Printf("Falha no pagamento: %v. Enviando para DLQ.\n", err)
					// Nack com requeue=false direciona a mensagem para a DLQ
					_ = d.Nack(false, false)
					continue
				}

				// Confirmação de sucesso
				_ = d.Ack(false)
			}
		}
	}()

	return nil
}

func processPayment(ctx context.Context, payload []byte) error {
	// Regra de negócio do pagamento...
	return nil
}

3. Apache Kafka na Prática com segmentio/kafka-go

O pacote github.com/segmentio/kafka-go é uma implementação 100% Go nativa (sem dependência da biblioteca C librdkafka), simplificando compilações em containers e pipelines de CI.

3.1 Publicando Eventos em Batch no Kafka

package kafka

import (
	"context"
	"time"

	"github.com/segmentio/kafka-go"
)

type EventProducer struct {
	writer *kafka.Writer
}

func NewEventProducer(brokers []string, topic string) *EventProducer {
	return &EventProducer{
		writer: &kafka.Writer{
			Addr:         kafka.TCP(brokers...),
			Topic:        topic,
			Balancer:     &kafka.Hash{}, // Garante mesma partição para a mesma chave
			MaxAttempts:  5,
			BatchSize:    100,                  // Agrupa até 100 mensagens
			BatchTimeout: 10 * time.Millisecond, // Ou aguarda no máximo 10ms
			RequiredAcks: kafka.RequireAll,     // acks=-1 (garante réplica em todos os in-sync replicas)
			Async:        false,
		},
	}
}

func (p *EventProducer) EmitUserSignedUp(ctx context.Context, userID string, payload []byte) error {
	return p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(userID), // Mensagens com mesmo UserID vão para a mesma partição (ordem estrita)
		Value: payload,
		Time:  time.Now().UTC(),
	})
}

func (p *EventProducer) Close() error {
	return p.writer.Close()
}

3.2 Consumidor Kafka com Consumer Groups e Manual Commit

No Kafka, um grupo de consumidores divide as partições do tópico entre si. O controle de quais mensagens já foram processadas é feito através do Offset Commit:

package kafka

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

type EventConsumer struct {
	reader *kafka.Reader
}

func NewEventConsumer(brokers []string, topic, groupID string) *EventConsumer {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        brokers,
		Topic:          topic,
		GroupID:        groupID, // Consumer Group
		MinBytes:       10e3,    // 10KB
		MaxBytes:       10e6,    // 10MB
		CommitInterval: 0,       // 0 desativa o auto-commit de tempo (queremos controle manual)
		StartOffset:    kafka.FirstOffset,
	})

	return &EventConsumer{reader: reader}
}

func (c *EventConsumer) StartConsuming(ctx context.Context) {
	for {
		// FetchMessage recupera a mensagem sem comitar o offset no Kafka
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if errors.Is(err, context.Canceled) {
				log.Println("Context cancelado. Finalizando leitura do Kafka.")
				return
			}
			log.Printf("Erro ao buscar mensagem: %v\n", err)
			time.Sleep(500 * time.Millisecond)
			continue
		}

		// Processa a mensagem de forma síncrona
		if err := handleEvent(ctx, msg.Value); err != nil {
			log.Printf("Erro ao processar mensagem offset %d: %v\n", msg.Offset, err)
			// Decide se descarta ou grava em dead-letter table
		}

		// Commita manualmente o offset somente após a conclusão do processamento
		if err := c.reader.CommitMessages(ctx, msg); err != nil {
			log.Printf("Erro ao comitar offset no Kafka: %v\n", err)
		}
	}
}

func handleEvent(ctx context.Context, data []byte) error {
	// Lógica do consumidor...
	return nil
}

4. O Ciclo de Vida Resiliente: Graceful Shutdown

Ao receber um sinal SIGTERM ou SIGINT do Kubernetes ou Docker, o serviço não deve abortar bruscamente no meio do processamento de uma mensagem em trânsito.

Implementação de Graceful Shutdown em Go

package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

func main() {
	// Intercepta sinais do sistema operacional
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	var wg sync.WaitGroup

	// Dispara o pool de workers
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			runWorker(ctx, workerID)
		}(i)
	}

	// Bloqueia a main até receber o sinal de parada do SO
	<-ctx.Done()
	log.Println("Sinal de encerramento recebido. Drenando mensagens ativas...")

	// Cria timeout máximo para a drenagem (ex: 15 segundos)
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		log.Println("Todos os workers finalizaram com segurança. Encerrando aplicação.")
	case <-shutdownCtx.Done():
		log.Println("AVISO: Timeout de drenagem esgotado! Forçando desligamento.")
	}
}

func runWorker(ctx context.Context, id int) {
	// Loop de consumo de mensageria monitorando ctx.Done()...
}

5. Consistência e o Padrão Transactional Outbox

Um dos problemas mais perigosos em microsserviços é a falha no dual-write:

// ❌ PROBLEMA DE CONSISTÊNCIA CRÍTICO:
tx := db.Begin()
db.CreateOrder(tx, order)
tx.Commit() // O banco confirmou a criação

// Se o broker cair ou a rede falhar exatamente aqui:
rabbitProducer.Publish(order) // 💥 FALHA! O banco criou o pedido, mas o evento NUNCA foi emitido!

A Solução: Transactional Outbox

1. Crie uma tabela outbox_events no mesmo banco de dados relacional.
2. Na mesma transação ACID em que o pedido é salvo, insira uma linha na tabela outbox_events.
3. Um processo em segundo plano (ou leitor de CDC via PostgreSQL com pgx) lê a tabela outbox_events, publica no RabbitMQ/Kafka e marca a linha como processada.

6. Resumo Prático

• RabbitMQ destaca-se pelo roteamento avançado via Exchanges e entrega pontual em filas descartáveis;
• Apache Kafka destaca-se como um log distribuído imutável de altíssima vazão com suporte nativo a reprocessamento (replay);
• No RabbitMQ, defina sempre Qos(prefetchCount) e use Manual ACK com d.Ack(false);
• No Kafka, utilize chaves de partição consistentes para preservar a ordem dos eventos por entidade (ex: UserID);
• Implemente Graceful Shutdown com signal.NotifyContext e sync.WaitGroup para garantir que mensagens em voo sejam concluídas antes do processo morrer;
• Previna inconsistências de estado entre o banco e o broker utilizando o padrão Transactional Outbox.