DEV Community

Cover image for Implementando o padrão Outbox Pattern com MySQL e usando Brighter
Rafael Andrade
Rafael Andrade

Posted on

Implementando o padrão Outbox Pattern com MySQL e usando Brighter

#braziliandevs #brighter #mysql #outbox

Introdução

Este guia demonstra como implementar o Padrão de Caixa de Saída (Outbox Pattern) com MySQL e .NET 8 usando a biblioteca Brighter para garantir consistência transacional entre atualizações de banco de dados e publicação de mensagens.

Projeto

O objetivo é processar um comando CreateNewOrder que publique dois eventos (OrderPlaced, OrderPaid) somente se a transação for bem-sucedida. Se ocorrer um erro (ex.: violação de regra de negócio), ambas as alterações no banco de dados e publicações de mensagens serão revertidas.

Requisitos

Mensagens

Para este projeto, usaremos estas 3 mensagens: CreateNewOrder, OrderPlaced e OrderPaid

public class CreateNewOrder() : Command(Guid.NewGuid()) { public decimal Value { get; set; } } public class OrderPlaced() : Event(Guid.NewGuid()) { public string OrderId { get; set; } = string.Empty; public decimal Value { get; set; } } public class OrderPaid() : Event(Guid.NewGuid()) { public string OrderId { get; set; } = string.Empty; }
Enter fullscreen mode Exit fullscreen mode

Mapeadores de Mensagens

Como apenas os eventos OrderPlaced e OrderPaid são publicados no RabbitMQ, precisamos implementar mapeadores para eles usando serialização JSON

public class OrderPlacedMapper : IAmAMessageMapper<OrderPlaced> { public Message MapToMessage(OrderPlaced request) { var header = new MessageHeader(); header.Id = request.Id; header.TimeStamp = DateTime.UtcNow; header.Topic = "order-placed"; header.MessageType = MessageType.MT_EVENT; var body = new MessageBody(JsonSerializer.Serialize(request)); return new Message(header, body); } public OrderPlaced MapToRequest(Message message) { return JsonSerializer.Deserialize<OrderPlaced>(message.Body.Bytes)!; } } public class OrderPaidMapper : IAmAMessageMapper<OrderPaid> { public Message MapToMessage(OrderPaid request) { var header = new MessageHeader(); header.Id = request.Id; header.TimeStamp = DateTime.UtcNow; header.Topic = "order-paid"; header.MessageType = MessageType.MT_EVENT; var body = new MessageBody(JsonSerializer.Serialize(request)); return new Message(header, body); } public OrderPaid MapToRequest(Message message) { return JsonSerializer.Deserialize<OrderPaid>(message.Body.Bytes)!; } }
Enter fullscreen mode Exit fullscreen mode

Manipuladores de Requisições

Para OrderPlaced e OrderPaid vamos registrar logs da mensagem recebida.

public class OrderPlaceHandler(ILogger<OrderPlaceHandler> logger) : RequestHandlerAsync<OrderPlaced> { public override Task<OrderPlaced> HandleAsync(OrderPlaced command, CancellationToken cancellationToken = default) { logger.LogInformation("{OrderId} placed with value {OrderValue}", command.OrderId, command.Value); return base.HandleAsync(command, cancellationToken); } } public class OrderPaidHandler(ILogger<OrderPaidHandler> logger) : RequestHandlerAsync<OrderPaid> { public override Task<OrderPaid> HandleAsync(OrderPaid command, CancellationToken cancellationToken = default) { logger.LogInformation("{OrderId} paid", command.OrderId); return base.HandleAsync(command, cancellationToken); } }
Enter fullscreen mode Exit fullscreen mode

Criar Novo Pedido

O manipulador CreateNewOrder vai esperar 10ms para emular um processo, depois publica o OrderPlaced, se o valor for divisível por 3 lança uma exceção (simulando um erro de negócio), caso contrário publica OrderPaid.

public class CreateNewOrderHandler(IAmACommandProcessor commandProcessor, IUnitOfWork unitOfWork, ILogger<CreateNewOrderHandler> logger) : RequestHandlerAsync<CreateNewOrder> { public override async Task<CreateNewOrder> HandleAsync(CreateNewOrder command, CancellationToken cancellationToken = default) { await unitOfWork.BeginTransactionAsync(cancellationToken); try { string id = Guid.NewGuid().ToString(); logger.LogInformation("Creating a new order: {OrderId}", id); await Task.Delay(10, cancellationToken); // emulando um processo _ = await commandProcessor.DepositPostAsync(new OrderPlaced { OrderId = id, Value = command.Value }, cancellationToken: cancellationToken); if (command.Value % 3 == 0) { throw new InvalidOperationException("invalid value"); } _ = await commandProcessor.DepositPostAsync(new OrderPaid { OrderId = id }, cancellationToken: cancellationToken); await unitOfWork.CommitAsync(cancellationToken); return await base.HandleAsync(command, cancellationToken); } catch { logger.LogError("Invalid data"); await unitOfWork.RollbackAsync(cancellationToken); throw; } } }
Enter fullscreen mode Exit fullscreen mode

Entendimento Chave:

  • DepositPostAsync armazena mensagens na caixa de saída dentro da mesma transação dos dados de negócio.

  • Se ocorrer uma exceção (ex.: InvalidOperationException), a transação é revertida, garantindo que não haja mensagens órfãs.

Configurando MySQL

Para integrar o Padrão de Caixa de Saída com MySQL, primeiro garanta que a tabela outbox_messages exista.

1. Esquema da Tabela SQL 

CREATE TABLE IF NOT EXISTS `outbox_messages`( `MessageId` CHAR(36) NOT NULL, `Topic` VARCHAR(255) NOT NULL, `MessageType` VARCHAR(32) NOT NULL, `Timestamp` TIMESTAMP(3) NOT NULL, `CorrelationId` CHAR(36) NULL, `ReplyTo` VARCHAR(255) NULL, `ContentType` VARCHAR(128) NULL, `Dispatched` TIMESTAMP(3) NULL, `HeaderBag` TEXT NOT NULL, `Body` TEXT NOT NULL, `Created` TIMESTAMP(3) NOT NULL DEFAULT NOW(3), `CreatedID` INT(11) NOT NULL AUTO_INCREMENT, UNIQUE(`CreatedID`), PRIMARY KEY (`MessageId`) );
Enter fullscreen mode Exit fullscreen mode

2. Configuração de Injeção de Dependência

Registre a caixa de saída e transação.

services .AddServiceActivator(opt => { /* Configuração de assinatura (ver artigo anterior) */ }) .UseMySqlOutbox(new MySqlConfiguration(ConnectionString, "outbox_messages"), typeof(MySqlConnectionProvider), ServiceLifetime.Scoped)) .UseMySqTransactionConnectionProvider(typeof(MySqlConnectionProvider)) .UseOutboxSweeper(opt => opt.BatchSize = 10);
Enter fullscreen mode Exit fullscreen mode

Por Que Funciona:

  • UseMySqlOutbox vincula a caixa de saída ao SQL Server.
  • UseOutboxSweeper configura a verificação em segundo plano para mensagens não entregues.

3. Gestão de Transações

Para garantir atomicidade entre lógica de negócio e publicação de mensagens na Brighter, implemente IMySqlTransactionConnectionProvider e IUnitOfWork para compartilhar o contexto de transação. Isso garante que mensagens só sejam armazenadas na caixa de saída se a transação do banco de dados for confirmada com sucesso.

a. SqlConnectionProvider 
public class MySqlConnectionProvider(MySqlUnitOfWork sqlConnection) : IMySqlTransactionConnectionProvider { private readonly MySqlUnitOfWork _sqlConnection = sqlConnection; public MySqlConnection GetConnection() { return _sqlConnection.Connection; } public Task<MySqlConnection> GetConnectionAsync(CancellationToken cancellationToken = default) { return Task.FromResult(_sqlConnection.Connection); } public MySqlTransaction? GetTransaction() { return _sqlConnection.Transaction; } public bool HasOpenTransaction => _sqlConnection.Transaction != null; public bool IsSharedConnection => true; }
Enter fullscreen mode Exit fullscreen mode
b. Unit of Work

E finalmente precisamos criar uma nova interface chamada IUnitOfWork

public interface IUnitOfWork { Task BeginTransactionAsync(CancellationToken cancellationToken, IsolationLevel isolationLevel = IsolationLevel.Serializable); Task CommitAsync(CancellationToken cancellationToken); Task RollbackAsync(CancellationToken cancellationToken); }
Enter fullscreen mode Exit fullscreen mode
c. Implementação do SqlUnitOfWork 
public class MySqlUnitOfWork : IUnitOfWork { public MySqlUnitOfWork(MySqlConfiguration configuration) { Connection = new(configuration.ConnectionString); Connection.Open(); } public MySqlConnection Connection { get; } public MySqlTransaction? Transaction { get; private set; } public async Task BeginTransactionAsync(CancellationToken cancellationToken, IsolationLevel isolationLevel = IsolationLevel.Serializable) { if (Transaction == null) { Transaction = await Connection.BeginTransactionAsync(isolationLevel); } } public async Task CommitAsync(CancellationToken cancellationToken) { if (Transaction != null) { await Transaction.CommitAsync(cancellationToken); } } public async Task RollbackAsync(CancellationToken cancellationToken) { if (Transaction != null) { await Transaction.RollbackAsync(cancellationToken); } } public Task<MySqlCommand> CreateSqlCommandAsync(string sql, MySqlParameter[] parameters, CancellationToken cancellationToken) { var command = Connection.CreateCommand(); if (Transaction != null) { command.Transaction = Transaction; } command.CommandText = sql; if (parameters.Length > 0) { command.Parameters.AddRange(parameters); } return Task.FromResult(command); } }
Enter fullscreen mode Exit fullscreen mode
d. Registrar Serviços na Injeção de Dependência 
services .AddScoped<MySqlUnitOfWork, MySqlUnitOfWork>() .TryAddScoped<IUnitOfWork>(provider => provider.GetRequiredService<SqlUnitOfWork>());
Enter fullscreen mode Exit fullscreen mode

Conclusão

Ao implementar o Padrão de Caixa de Saída com Brighter e MySQL, demonstramos como alcançar consistência transacional entre atualizações de banco de dados e publicação de mensagens. Essa abordagem garante que:

  1. Mensagens são publicadas somente se a transação for confirmada com sucesso
    • Usando DepositPostAsync, mensagens como OrderPlaced e OrderPaid são armazenadas na tabela outbox_messages dentro da mesma transação dos dados de negócio. Se o manipulador falhar (ex.: erro simulado), a transação é revertida e nenhuma mensagem órfã é enviada.
    • O IMySqlTransactionConnectionProvider da Brighter garante que atualizações de banco de dados e depósitos de mensagens compartilhem a mesma transação.
  2. Tolerância a Falhas via Outbox Sweeper
    • O UseOutboxSweeper verifica periodicamente mensagens não entregues e tenta reenviá-las até que sejam reconhecidas pelo RabbitMQ. Isso desacopla a publicação de mensagens da execução do manipulador, garantindo confiabilidade mesmo que o broker esteja temporariamente indisponível.
  3. Arquitetura Desacoplada
    • Aplicações focam em transações locais, enquanto a Brighter trata a entrega de mensagens de forma assíncrona. Isso evita acoplamento com a infraestrutura de mensagens e simplifica a escalabilidade. Esta implementação mostra como a Brighter abstrai complexidades, permitindo que desenvolvedores se concentrem na lógica de negócio enquanto garantem confiabilidade em sistemas distribuídos. Para uso em produção, combine esse padrão com ferramentas de monitoramento (ex.: Prometheus), filas de mensagens mortas (DLQs) para tratar mensagens corrompidas e adicione índices na tabela de caixa de saída nas colunas Dispatched e Timestamp.

Referência

Top comments (0)