The Transactional Outbox Pattern

https://jdaniel1987.github.io/TransactionalOutboxPattern

The Transactional Outbox Pattern is a powerful technique used in distributed systems to ensure message delivery reliability when integrating with external systems or event-based architectures. In this guide, we will explore how to implement the Outbox Pattern in .NET.

What is the Transactional Outbox Pattern?

The Transactional Outbox Pattern ensures that database operations and message publishing are performed atomically, meaning they are treated as a single, indivisible unit of work. This guarantees that either all operations succeed together or none of them take effect. This is especially important when working with event-driven systems to avoid inconsistencies caused by partial failures.

How It Works

1. Transactional Writing: Events or messages to be sent are written to an Outbox table in the same database transaction as the business operation.
2. Background Processing: A separate process reads messages from the Outbox table and publishes them to the external system (e.g., a message broker).

这是一篇关于**事务性发件箱模式（Transactional Outbox Pattern）**的技术博客文章翻译。这篇文章详细介绍了如何在 .NET 分布式系统中利用该模式确保消息传递的可靠性。

以下是该网页内容的中文翻译：

事务性发件箱模式

事务性发件箱模式是分布式系统中一种强大的技术，用于在集成外部系统或基于事件的架构时确保消息传递的可靠性。在本指南中，我们将探讨如何在 .NET 中实现发件箱模式。

什么是事务性发件箱模式？

事务性发件箱模式确保数据库操作和消息发布是原子性地执行的，这意味着它们被视为一个单一的、不可分割的工作单元。这保证了要么所有操作都一起成功，要么都不生效。这在处理事件驱动系统时尤为重要，可以避免因部分失败而导致的不一致。

它是如何工作的：

1. 事务性写入：待发送的事件或消息会与业务操作写在同一个数据库事务中，存入一个 Outbox（发件箱）表。
2. 后台处理：一个独立的进程从 Outbox 表中读取消息，并将其发布到外部系统（例如消息代理）。

设置事务性发件箱模式

第一步：创建发件箱表
在数据库中定义一个 OutboxMessage 表：

CREATE TABLE OutboxMessage (
    Id UNIQUEIDENTIFIER PRIMARY KEY,
    EventType NVARCHAR(256) NOT NULL,
    Payload NVARCHAR(MAX) NOT NULL,
    CreatedAt DATETIME NOT NULL DEFAULT GETDATE(),
    ProcessedAt DATETIME NULL
);

第二步：添加发件箱实体
在项目中创建对应的模型：

public class OutboxMessage 
{
    public Guid Id { get; set; } = Guid.NewGuid();
    public string EventType { get; set; }
    public string Payload { get; set; }
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
    public DateTime? ProcessedAt { get; set; }
}

第三步：在事务中保存发件箱消息
在执行业务逻辑时，将事件保存到 OutboxMessage 表中：

public async Task SaveOrderAsync(Order order) 
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();

    try 
    {
        _dbContext.Orders.Add(order);

        var outboxMessage = new OutboxMessage
        {
            EventType = nameof(OrderCreatedEvent),
            Payload = JsonSerializer.Serialize(new OrderCreatedEvent(order.Id, order.TotalAmount))
        };

        _dbContext.OutboxMessages.Add(outboxMessage);

        await _dbContext.SaveChangesAsync();
        await transaction.CommitAsync();
    } 
    catch 
    {
        await transaction.RollbackAsync();
        throw;
    }
}

第四步：处理发件箱消息
使用后台服务从 OutboxMessage 表中读取并发布消息：

public class OutboxProcessor : BackgroundService 
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<OutboxProcessor> _logger;

    public OutboxProcessor(IServiceProvider serviceProvider, ILogger<OutboxProcessor> logger) 
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken cancellationToken) 
    {
        while (!cancellationToken.IsCancellationRequested) 
        {
            try 
            {
                using var scope = _serviceProvider.CreateScope();
                var dbContext = scope.ServiceProvider.GetRequiredService<ApplicationDbContext>();

                var messages = await dbContext.OutboxMessages
                    .Where(m => !m.Processed)
                    .ToListAsync(stoppingToken);

                foreach (var message in messages) 
                {
                    // 发布到消息代理
                    await PublishMessageAsync(message);

                    // 标记为已处理
                    message.Processed = true;
                }

                await dbContext.SaveChangesAsync(stoppingToken);
            } 
            catch (Exception ex) 
            {
                _logger.LogError(ex, "处理发件箱消息时出错");
            }
        }
    }
}

第五步：注册后台服务
将服务添加到你的 Program.cs 或 Startup.cs 中：

builder.Services.AddHostedService<OutboxProcessor>();

事务性发件箱模式的优势

• 原子性：确保业务操作和消息发布同时发生。
• 可靠性：降低分布式系统中消息丢失的风险。
• 灵活性：允许异步处理，而不会阻塞主要业务逻辑。

结论

事务性发件箱模式是确保分布式系统中可靠消息传递的稳健解决方案。通过在 .NET 中实现它，你可以构建具有弹性和可扩展性的应用程序，以一致且高效的方式处理消息传递。