Especialização em Backend com Go & Gin • Demonstrações Práticas Compiláveis

💻 Código de Exemplo 01 — Publicador e Consumidor RabbitMQ com Ack/Nack

package messaging

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

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

type PedidoCriadoEvent struct {
	PedidoID uint    `json:"pedido_id"`
	Total    float64 `json:"total"`
}

// PublicarMensagem declara a fila (idempotente) e publica com entrega
// persistente, garantindo que a mensagem sobreviva a um restart do broker.
func PublicarMensagem(ch *amqp.Channel, fila string, evento PedidoCriadoEvent) error {
	if _, err := ch.QueueDeclare(fila, true, false, false, false, nil); err != nil {
		return err
	}
	body, err := json.Marshal(evento)
	if err != nil {
		return err
	}
	return ch.PublishWithContext(context.Background(), "", fila, false, false, amqp.Publishing{
		ContentType:  "application/json",
		DeliveryMode: amqp.Persistent,
		Body:         body,
	})
}

// ConsumirComRetentativa processa cada delivery de forma idempotente e usa
// Ack/Nack manual para controlar reentrega em caso de falha do worker.
func ConsumirComRetentativa(msgs <-chan amqp.Delivery, processar func(PedidoCriadoEvent) error) {
	for d := range msgs {
		var evento PedidoCriadoEvent
		if err := json.Unmarshal(d.Body, &evento); err != nil {
			log.Printf("mensagem malformada, descartando: %v", err)
			d.Nack(false, false) // não reenfileira: vai para DLX se configurada
			continue
		}
		if err := processar(evento); err != nil {
			log.Printf("erro ao processar pedido %d, reenfileirando: %v", evento.PedidoID, err)
			d.Nack(false, true)
			continue
		}
		d.Ack(false)
	}
}

💻 Código de Exemplo 02 — Padrão Outbox Transacional

package outbox

import (
	"encoding/json"
	"time"

	"gorm.io/gorm"
)

// OutboxEvent é gravado na MESMA transação SQL que o dado de negócio,
// eliminando o problema de dual-write entre banco e broker.
type OutboxEvent struct {
	ID          uint `gorm:"primaryKey"`
	Topico      string
	Payload     string
	Processado  bool
	CriadoEm    time.Time
	ProcessadoEm *time.Time
}

// CriarPedidoComEvento grava o pedido e o evento outbox atomicamente: se a
// transação falhar, nem o pedido nem o evento existem — nunca um sem o outro.
func CriarPedidoComEvento(db *gorm.DB, pedido *Pedido, evento PedidoCriadoEvent) error {
	return db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Create(pedido).Error; err != nil {
			return err
		}
		payload, err := json.Marshal(evento)
		if err != nil {
			return err
		}
		return tx.Create(&OutboxEvent{
			Topico:   "pedidos.criado",
			Payload:  string(payload),
			CriadoEm: time.Now().UTC(),
		}).Error
	})
}

// PublicarPendentes é chamado por um poller periódico: lê eventos não
// processados e os publica no broker, marcando como processados após sucesso.
func PublicarPendentes(db *gorm.DB, publicar func(topico string, payload []byte) error) error {
	var pendentes []OutboxEvent
	if err := db.Where("processado = ?", false).Limit(100).Find(&pendentes).Error; err != nil {
		return err
	}
	for _, evento := range pendentes {
		if err := publicar(evento.Topico, []byte(evento.Payload)); err != nil {
			continue // tentativa seguinte do poller reprocessa
		}
		agora := time.Now().UTC()
		db.Model(&evento).Updates(map[string]interface{}{"processado": true, "processado_em": agora})
	}
	return nil
}

💻 Código de Exemplo 03 — Fila com Dead Letter Exchange (DLX)

package messaging

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

// DeclararFilaComDLX cria a fila principal já configurada para rotear
// mensagens para a fila morta após esgotar TTL/tentativas, isolando
// "poison messages" sem travar o processamento da fila saudável.
func DeclararFilaComDLX(ch *amqp.Channel) error {
	if _, err := ch.QueueDeclare("dlq_pedidos", true, false, false, false, nil); err != nil {
		return err
	}
	if err := ch.ExchangeDeclare("dlx_exchange", "direct", true, false, false, false, nil); err != nil {
		return err
	}
	if err := ch.QueueBind("dlq_pedidos", "pedidos_mortos", "dlx_exchange", false, nil); err != nil {
		return err
	}

	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx_exchange",
		"x-dead-letter-routing-key": "pedidos_mortos",
		"x-message-ttl":             int32(60000), // 60s sem consumo => vai para a DLQ
	}
	_, err := ch.QueueDeclare("pedidos_queue", true, false, false, false, args)
	return err
}