dotnet-outbox-pattern

Outbox 模式实现

概述

Outbox 模式可确保可靠地处理事件：

• 原子持久化 - 在与聚合相同的事务中保存事件
• 保证投递 - 即使应用崩溃，也会处理事件
• 最终一致性 - 异步处理并重试
• 幂等性 - 妥善处理重复处理

快速参考

组件	用途
OutboxMessage	已持久化的事件实体
OutboxMessageConfiguration	EF Core 映射
ProcessOutboxMessagesJob	后台处理器（Quartz）
IdempotentDomainEventHandler	去重处理器包装器
OutboxConsumer	直接轮询数据库的替代方案

---

Outbox 结构

/Infrastructure/
├── Outbox/
│   ├── OutboxMessage.cs
│   ├── OutboxMessageConfiguration.cs
│   ├── ProcessOutboxMessagesJob.cs
│   ├── ProcessOutboxMessagesJobSetup.cs
│   └── IdempotentDomainEventHandler.cs
└── ApplicationDbContext.cs

---

模板：Outbox 消息实体

// src/{name}.infrastructure/Outbox/OutboxMessage.cs
namespace {name}.infrastructure.outbox;

/// <summary>
/// Represents a domain event stored for reliable delivery
/// </summary>
public sealed class OutboxMessage
{
    public OutboxMessage()
    {
    }

    public OutboxMessage(Guid id, string type, string content, DateTime occurredOnUtc)
    {
        Id = id;
        Type = type;
        Content = content;
        OccurredOnUtc = occurredOnUtc;
    }

    /// <summary>
    /// Unique identifier for this message
    /// </summary>
    public Guid Id { get; set; }

    /// <summary>
    /// Assembly-qualified type name of the domain event
    /// </summary>
    public string Type { get; set; } = string.Empty;

    /// <summary>
    /// JSON-serialized event content
    /// </summary>
    public string Content { get; set; } = string.Empty;

    /// <summary>
    /// When the event originally occurred
    /// </summary>
    public DateTime OccurredOnUtc { get; set; }

    /// <summary>
    /// When the message was successfully processed (null if not yet processed)
    /// </summary>
    public DateTime? ProcessedOnUtc { get; set; }

    /// <summary>
    /// Error message if processing failed
    /// </summary>
    public string? Error { get; set; }

    /// <summary>
    /// Number of processing attempts
    /// </summary>
    public int RetryCount { get; set; }
}

---

模板：EF Core 配置

// src/{name}.infrastructure/Outbox/OutboxMessageConfiguration.cs
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Metadata.Builders;

namespace {name}.infrastructure.outbox;

internal sealed class OutboxMessageConfiguration 
    : IEntityTypeConfiguration<OutboxMessage>
{
    public void Configure(EntityTypeBuilder<OutboxMessage> builder)
    {
        builder.ToTable("outbox_message");

        builder.HasKey(o => o.Id);

        builder.Property(o => o.Id)
            .ValueGeneratedNever();

        builder.Property(o => o.Type)
            .HasMaxLength(500)
            .IsRequired();

        builder.Property(o => o.Content)
            .HasColumnType("jsonb")  // PostgreSQL JSONB
            .IsRequired();

        builder.Property(o => o.OccurredOnUtc)
            .IsRequired();

        builder.Property(o => o.ProcessedOnUtc);

        builder.Property(o => o.Error)
            .HasColumnType("text");

        builder.Property(o => o.RetryCount)
            .HasDefaultValue(0);

        // Index for efficient polling of unprocessed messages
        builder.HasIndex(o => o.ProcessedOnUtc)
            .HasFilter("processed_on_utc IS NULL")
            .HasDatabaseName("ix_outbox_message_unprocessed");

        // Index for cleanup of old processed messages
        builder.HasIndex(o => o.ProcessedOnUtc)
            .HasFilter("processed_on_utc IS NOT NULL")
            .HasDatabaseName("ix_outbox_message_processed");
    }
}

---

模板：DbContext 集成

// src/{name}.infrastructure/ApplicationDbContext.cs
using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using {name}.domain.abstractions;
using {name}.infrastructure.outbox;

namespace {name}.infrastructure;

public sealed class ApplicationDbContext : DbContext, IUnitOfWork
{
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
        WriteIndented = false
    };

    public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options)
        : base(options)
    {
    }

    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.ApplyConfigurationsFromAssembly(typeof(ApplicationDbContext).Assembly);
        base.OnModelCreating(modelBuilder);
    }

    public override async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
    {
        // ═══════════════════════════════════════════════════════════════
        // CRITICAL: Add domain events to outbox BEFORE SaveChanges
        // This ensures atomic persistence - events saved in same transaction
        // ═══════════════════════════════════════════════════════════════
        ConvertDomainEventsToOutboxMessages();

        return await base.SaveChangesAsync(cancellationToken);
    }

    private void ConvertDomainEventsToOutboxMessages()
    {
        // Get all entities with domain events
        var entitiesWithEvents = ChangeTracker
            .Entries<Entity>()
            .Where(e => e.Entity.GetDomainEvents().Any())
            .Select(e => e.Entity)
            .ToList();

        // Extract all domain events
        var domainEvents = entitiesWithEvents
            .SelectMany(e => e.GetDomainEvents())
            .ToList();

        // Clear events from entities (they're now in outbox)
        foreach (var entity in entitiesWithEvents)
        {
            entity.ClearDomainEvents();
        }

        // Convert to outbox messages
        foreach (var domainEvent in domainEvents)
        {
            var outboxMessage = new OutboxMessage
            {
                Id = Guid.NewGuid(),
                Type = domainEvent.GetType().AssemblyQualifiedName!,
                Content = JsonSerializer.Serialize(
                    domainEvent,
                    domainEvent.GetType(),
                    JsonOptions),
                OccurredOnUtc = DateTime.UtcNow
            };

            OutboxMessages.Add(outboxMessage);
        }
    }
}

---

模板：Outbox 处理器作业（Quartz）

// src/{name}.infrastructure/Outbox/ProcessOutboxMessagesJob.cs
using System.Text.Json;
using MediatR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Quartz;
using {name}.domain.abstractions;

namespace {name}.infrastructure.outbox;

/// <summary>
/// Background job that processes outbox messages
/// Uses Quartz for scheduling with configurable interval
/// </summary>
[DisallowConcurrentExecution]  // Prevent parallel execution
public sealed class ProcessOutboxMessagesJob : IJob
{
    private const int BatchSize = 20;
    private const int MaxRetries = 3;

    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase
    };

    private readonly ApplicationDbContext _dbContext;
    private readonly IPublisher _publisher;
    private readonly ILogger<ProcessOutboxMessagesJob> _logger;

    public ProcessOutboxMessagesJob(
        ApplicationDbContext dbContext,
        IPublisher publisher,
        ILogger<ProcessOutboxMessagesJob> logger)
    {
        _dbContext = dbContext;
        _publisher = publisher;
        _logger = logger;
    }

    public async Task Execute(IJobExecutionContext context)
    {
        _logger.LogDebug("Starting outbox message processing...");

        var messages = await GetUnprocessedMessages(context.CancellationToken);

        if (!messages.Any())
        {
            _logger.LogDebug("No outbox messages to process");
            return;
        }

        _logger.LogInformation(
            "Processing {Count} outbox messages",
            messages.Count);

        foreach (var message in messages)
        {
            await ProcessMessage(message, context.CancellationToken);
        }

        await _dbContext.SaveChangesAsync(context.CancellationToken);

        _logger.LogInformation("Completed outbox message processing");
    }

    private async Task<List<OutboxMessage>> GetUnprocessedMessages(
        CancellationToken cancellationToken)
    {
        return await _dbContext.OutboxMessages
            .Where(m => m.ProcessedOnUtc == null)
            .Where(m => m.RetryCount < MaxRetries)
            .OrderBy(m => m.OccurredOnUtc)
            .Take(BatchSize)
            .ToListAsync(cancellationToken);
    }

    private async Task ProcessMessage(
        OutboxMessage message,
        CancellationToken cancellationToken)
    {
        try
        {
            _logger.LogDebug(
                "Processing outbox message {MessageId} of type {Type}",
                message.Id,
                message.Type);

            // Resolve the event type
            var eventType = Type.GetType(message.Type);

            if (eventType is null)
            {
                _logger.LogError(
                    "Could not resolve type {Type} for message {MessageId}",
                    message.Type,
                    message.Id);

                message.Error = $"Could not resolve type: {message.Type}";
                message.ProcessedOnUtc = DateTime.UtcNow;
                return;
            }

            // Deserialize the event
            var domainEvent = JsonSerializer.Deserialize(
                message.Content,
                eventType,
                JsonOptions) as IDomainEvent;

            if (domainEvent is null)
            {
                _logger.LogError(
                    "Could not deserialize message {MessageId}",
                    message.Id);

                message.Error = "Could not deserialize message content";
                message.ProcessedOnUtc = DateTime.UtcNow;
                return;
            }

            // Publish to MediatR handlers
            await _publisher.Publish(domainEvent, cancellationToken);

            // Mark as processed
            message.ProcessedOnUtc = DateTime.UtcNow;
            message.Error = null;

            _logger.LogInformation(
                "Successfully processed outbox message {MessageId}",
                message.Id);
        }
        catch (Exception ex)
        {
            _logger.LogError(
                ex,
                "Error processing outbox message {MessageId}. Retry count: {RetryCount}",
                message.Id,
                message.RetryCount);

            message.RetryCount++;
            message.Error = ex.ToString();

            // Mark as processed if max retries exceeded
            if (message.RetryCount >= MaxRetries)
            {
                message.ProcessedOnUtc = DateTime.UtcNow;
                
                _logger.LogError(
                    "Outbox message {MessageId} exceeded max retries and has been marked as failed",
                    message.Id);
            }
        }
    }
}

---

模板：作业配置

// src/{name}.infrastructure/Outbox/ProcessOutboxMessagesJobSetup.cs
using Microsoft.Extensions.Options;
using Quartz;

namespace {name}.infrastructure.outbox;

internal sealed class ProcessOutboxMessagesJobSetup 
    : IConfigureOptions<QuartzOptions>
{
    public void Configure(QuartzOptions options)
    {
        var jobKey = JobKey.Create(nameof(ProcessOutboxMessagesJob));

        options
            .AddJob<ProcessOutboxMessagesJob>(jobBuilder =>
                jobBuilder.WithIdentity(jobKey))
            .AddTrigger(triggerBuilder =>
                triggerBuilder
                    .ForJob(jobKey)
                    .WithSimpleSchedule(schedule =>
                        schedule
                            .WithIntervalInSeconds(10)  // Poll every 10 seconds
                            .RepeatForever()));
    }
}

---

模板：幂等事件处理器包装器

// src/{name}.infrastructure/Outbox/IdempotentDomainEventHandler.cs
using MediatR;
using Microsoft.EntityFrameworkCore;
using {name}.domain.abstractions;

namespace {name}.infrastructure.outbox;

/// <summary>
/// Wrapper that ensures domain events are processed only once
/// Uses a separate tracking table to detect duplicates
/// </summary>
public abstract class IdempotentDomainEventHandler<TEvent> 
    : INotificationHandler<TEvent>
    where TEvent : IDomainEvent
{
    private readonly ApplicationDbContext _dbContext;

    protected IdempotentDomainEventHandler(ApplicationDbContext dbContext)
    {
        _dbContext = dbContext;
    }

    public async Task Handle(TEvent notification, CancellationToken cancellationToken)
    {
        var handlerName = GetType().Name;
        var eventId = notification.Id;

        // Check if already processed
        var alreadyProcessed = await _dbContext
            .Set<OutboxMessageConsumer>()
            .AnyAsync(
                c => c.EventId == eventId && c.HandlerName == handlerName,
                cancellationToken);

        if (alreadyProcessed)
        {
            return;  // Skip duplicate processing
        }

        // Process the event
        await HandleAsync(notification, cancellationToken);

        // Mark as processed
        _dbContext.Set<OutboxMessageConsumer>().Add(new OutboxMessageConsumer
        {
            Id = Guid.NewGuid(),
            EventId = eventId,
            HandlerName = handlerName,
            ProcessedOnUtc = DateTime.UtcNow
        });

        await _dbContext.SaveChangesAsync(cancellationToken);
    }

    protected abstract Task HandleAsync(TEvent notification, CancellationToken cancellationToken);
}

/// <summary>
/// Tracks which handlers have processed which events
/// </summary>
public sealed class OutboxMessageConsumer
{
    public Guid Id { get; set; }
    public Guid EventId { get; set; }
    public string HandlerName { get; set; } = string.Empty;
    public DateTime ProcessedOnUtc { get; set; }
}

---

模板：清理作业

// src/{name}.infrastructure/Outbox/CleanupOutboxMessagesJob.cs
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Quartz;

namespace {name}.infrastructure.outbox;

/// <summary>
/// Cleans up old processed outbox messages
/// Runs daily to prevent table bloat
/// </summary>
[DisallowConcurrentExecution]
public sealed class CleanupOutboxMessagesJob : IJob
{
    private const int RetentionDays = 7;
    private const int BatchSize = 1000;

    private readonly ApplicationDbContext _dbContext;
    private readonly ILogger<CleanupOutboxMessagesJob> _logger;

    public CleanupOutboxMessagesJob(
        ApplicationDbContext dbContext,
        ILogger<CleanupOutboxMessagesJob> logger)
    {
        _dbContext = dbContext;
        _logger = logger;
    }

    public async Task Execute(IJobExecutionContext context)
    {
        var cutoffDate = DateTime.UtcNow.AddDays(-RetentionDays);

        _logger.LogInformation(
            "Cleaning up outbox messages processed before {CutoffDate}",
            cutoffDate);

        var totalDeleted = 0;
        int deletedInBatch;

        do
        {
            deletedInBatch = await _dbContext.OutboxMessages
                .Where(m => m.ProcessedOnUtc != null)
                .Where(m => m.ProcessedOnUtc < cutoffDate)
                .Take(BatchSize)
                .ExecuteDeleteAsync(context.CancellationToken);

            totalDeleted += deletedInBatch;

        } while (deletedInBatch == BatchSize);

        _logger.LogInformation(
            "Cleaned up {Count} old outbox messages",
            totalDeleted);
    }
}

---

模板：注册

// src/{name}.infrastructure/DependencyInjection.cs
private static void AddBackgroundJobs(
    IServiceCollection services,
    IConfiguration configuration)
{
    services.AddQuartz(configure =>
    {
        // Use persistent job store for production
        configure.UsePersistentStore(options =>
        {
            options.UsePostgres(configuration.GetConnectionString("Database")!);
            options.UseJsonSerializer();
        });
    });

    services.AddQuartzHostedService(options =>
    {
        options.WaitForJobsToComplete = true;
    });

    // Register job configurations
    services.ConfigureOptions<ProcessOutboxMessagesJobSetup>();
    services.ConfigureOptions<CleanupOutboxMessagesJobSetup>();
}

---

数据库迁移

-- Create outbox_message table
CREATE TABLE outbox_message (
    id UUID PRIMARY KEY,
    type VARCHAR(500) NOT NULL,
    content JSONB NOT NULL,
    occurred_on_utc TIMESTAMP NOT NULL,
    processed_on_utc TIMESTAMP NULL,
    error TEXT NULL,
    retry_count INTEGER NOT NULL DEFAULT 0
);

-- Index for unprocessed messages (most important)
CREATE INDEX ix_outbox_message_unprocessed 
ON outbox_message (occurred_on_utc) 
WHERE processed_on_utc IS NULL;

-- Index for cleanup of old messages
CREATE INDEX ix_outbox_message_processed 
ON outbox_message (processed_on_utc) 
WHERE processed_on_utc IS NOT NULL;

-- Optional: Consumer tracking table for idempotency
CREATE TABLE outbox_message_consumer (
    id UUID PRIMARY KEY,
    event_id UUID NOT NULL,
    handler_name VARCHAR(500) NOT NULL,
    processed_on_utc TIMESTAMP NOT NULL
);

CREATE UNIQUE INDEX ix_outbox_consumer_event_handler 
ON outbox_message_consumer (event_id, handler_name);

---

关键规则

1. 同一事务 - 在一个事务中随聚合一起保存事件
2. 幂等处理器 - 必须能够处理重复投递
3. 不保证顺序 - 事件的处理顺序可能不同
4. 带退避的重试 - 不要对失败事件反复发起请求
5. 清理旧消息 - 防止表膨胀
6. 监控失败 - 超过最大重试次数时发出警报
7. 类型序列化 - 使用 AssemblyQualifiedName 进行反序列化
8. JSON 序列化 - 序列化和反序列化时使用一致的选项
9. 批量处理 - 不要一次只处理一条
10. 禁用并发执行 - 防止重复处理

---

应避免的反模式

// ❌ WRONG: Publishing events directly (not reliable)
await _publisher.Publish(new UserCreatedEvent(user.Id));
await _unitOfWork.SaveChangesAsync();  // Event lost if save fails!

// ✅ CORRECT: Events converted to outbox in SaveChanges
user.RaiseDomainEvent(new UserCreatedEvent(user.Id));
await _unitOfWork.SaveChangesAsync();  // Events saved atomically

// ❌ WRONG: Non-idempotent handler
public async Task Handle(UserCreatedEvent e, CancellationToken ct)
{
    await _emailService.SendWelcomeEmail(e.UserId);  // Sent twice on retry!
}

// ✅ CORRECT: Idempotent handler
public async Task Handle(UserCreatedEvent e, CancellationToken ct)
{
    if (await _emailLog.ExistsAsync(e.UserId, "welcome"))
        return;  // Already sent
    
    await _emailService.SendWelcomeEmail(e.UserId);
    await _emailLog.RecordAsync(e.UserId, "welcome");
}

// ❌ WRONG: Processing one message at a time
foreach (var message in allMessages)  // Could be millions!
{
    await ProcessMessage(message);
}

// ✅ CORRECT: Batch with limit
var messages = await _dbContext.OutboxMessages
    .Where(m => m.ProcessedOnUtc == null)
    .Take(20)  // Batch size
    .ToListAsync();

---

相关技能

• dotnet-domain-events-generator - 放入 Outbox 的领域事件
• dotnet-quartz-background-jobs - 后台作业调度
• dotnet-clean-architecture - 基础设施层设置