dotnet-messaging-patterns

dotnet-messaging-patterns

适用于 .NET 事件驱动架构的持久化消息传递模式。涵盖发布/订阅、竞争消费者、死信队列、Saga/流程管理器编排，以及使用 Azure Service Bus、RabbitMQ 和 MassTransit 实现的投递保证策略。

不在范围内： 后台服务生命周期和 IHostedService 注册——请参阅 [skill:dotnet-background-services]。弹性管道和重试策略——请参阅 [skill:dotnet-resilience]。JSON/二进制序列化配置——请参阅 [skill:dotnet-serialization]。使用 Channel<T> 的进程内生产者/消费者队列——请参阅 [skill:dotnet-channels]。

交叉参考：[skill:dotnet-background-services] 用于托管消息消费者，[skill:dotnet-resilience] 用于消息处理程序的容错，[skill:dotnet-serialization] 用于消息信封序列化，[skill:dotnet-channels] 用于进程内排队模式。

---

消息传递基础

消息类型

类型	用途	示例
命令	请求执行某项操作（一个接收者）	PlaceOrder、ShipPackage
事件	通知某事已发生（多个接收者）	OrderPlaced、PaymentReceived
文档	在系统之间传输数据	CustomerProfile、ProductCatalog

命令会发送到特定队列；事件会发布到主题/交换机，并投递给所有订阅者。这一区别决定了应选择点对点还是发布/订阅拓扑。

投递保证

保证	行为	实现方式
至多一次	发出后不再处理；消息可能丢失	不确认，不重试
至少一次	消息会持续重试，直到得到确认；可能出现重复	处理后确认 + 失败时重试
恰好一次	每条消息只处理一次	至少一次 + 幂等消费者

至少一次投递配合幂等消费者是持久化消息传递的标准做法。真正的恰好一次投递需要分布式事务（大多数代理不支持），或由消费者端进行去重。

---

发布/订阅

Azure Service Bus 主题

// Publisher -- send event to a topic
await using var client = new ServiceBusClient(connectionString);
await using var sender = client.CreateSender("order-events");

var message = new ServiceBusMessage(
    JsonSerializer.SerializeToUtf8Bytes(new OrderPlaced(orderId, total)))
{
    Subject = nameof(OrderPlaced),
    ContentType = "application/json",
    MessageId = Guid.NewGuid().ToString()
};

await sender.SendMessageAsync(message, cancellationToken);

// Subscriber -- process events from a subscription
await using var processor = client.CreateProcessor(
    topicName: "order-events",
    subscriptionName: "billing-service",
    new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = 10,
        AutoCompleteMessages = false
    });

processor.ProcessMessageAsync += async args =>
{
    var body = args.Message.Body.ToObjectFromJson<OrderPlaced>();
    await HandleOrderPlacedAsync(body);
    await args.CompleteMessageAsync(args.Message);
};

processor.ProcessErrorAsync += args =>
{
    logger.LogError(args.Exception, "Error processing message");
    return Task.CompletedTask;
};

await processor.StartProcessingAsync(cancellationToken);

关键包：

<PackageReference Include="Azure.Messaging.ServiceBus" Version="7.*" />

RabbitMQ Fanout 交换机

// Publisher -- declare exchange and publish
var factory = new ConnectionFactory { HostName = "localhost" };
await using var connection = await factory.CreateConnectionAsync();
await using var channel = await connection.CreateChannelAsync();

await channel.ExchangeDeclareAsync(
    exchange: "order-events",
    type: ExchangeType.Fanout,
    durable: true);

var body = JsonSerializer.SerializeToUtf8Bytes(
    new OrderPlaced(orderId, total));

await channel.BasicPublishAsync(
    exchange: "order-events",
    routingKey: string.Empty,
    body: body);

关键包：

<PackageReference Include="RabbitMQ.Client" Version="7.*" />

MassTransit 发布

MassTransit 对消息代理进行抽象，为 Azure Service Bus、RabbitMQ、Amazon SQS 和内存传输提供统一 API。

// Registration
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });
        cfg.ConfigureEndpoints(context);
    });
});

// Publisher
public sealed class OrderService(IPublishEndpoint publishEndpoint)
{
    public async Task PlaceOrderAsync(
        Guid orderId, decimal total, CancellationToken ct)
    {
        // Process order...
        await publishEndpoint.Publish(
            new OrderPlaced(orderId, total), ct);
    }
}

// Consumer
public sealed class OrderPlacedConsumer(
    ILogger<OrderPlacedConsumer> logger)
    : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        logger.LogInformation(
            "Processing order {OrderId}", context.Message.OrderId);
        await ProcessAsync(context.Message);
    }
}

// Message contract (use records in a shared contracts assembly)
public record OrderPlaced(Guid OrderId, decimal Total);

关键包：

<PackageReference Include="MassTransit" Version="8.*" />

<PackageReference Include="MassTransit.RabbitMQ" Version="8.*" />

<PackageReference Include="MassTransit.Azure.ServiceBus.Core" Version="8.*" />

---

竞争消费者

多个消费者实例并行处理同一队列中的消息。代理会将每条消息投递给一个消费者，从而在多个实例间分配负载。

模式

Queue: order-processing
  ├── Consumer Instance A  (picks message 1)
  ├── Consumer Instance B  (picks message 2)
  └── Consumer Instance C  (picks message 3)

Azure Service Bus——扩展消费者

// Multiple instances reading from the same queue automatically compete.
// MaxConcurrentCalls controls per-instance parallelism.
var processor = client.CreateProcessor("order-processing",
    new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = 20,
        PrefetchCount = 50,
        AutoCompleteMessages = false
    });

MassTransit——并发限制

x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseConcurrentMessageLimit(10);
});

顺序注意事项

竞争消费者会牺牲严格的顺序，以换取更高吞吐量。需要保证顺序时：

• Azure Service Bus：使用会话（RequiresSession = true），保证同一会话 ID（例如每位客户）内的先进先出顺序
• RabbitMQ：每个队列使用单个消费者，或使用一致性哈希交换机按键分区
• MassTransit：配置 UseMessagePartitioner，实现基于键的顺序处理

---

死信队列

死信队列（DLQ）用于捕获重试次数耗尽后仍无法处理的消息，避免毒消息阻塞主队列。

消息进入死信队列的原因

原因	触发条件
超过最大投递尝试次数	消息处理失败 N 次
TTL 过期	消息在队列中的停留时间超过其生存时间
消费者拒绝	消费者明确将消息移入死信队列
队列长度超限	队列溢出策略将消息路由到死信队列

Azure Service Bus DLQ

// Dead-letter a message with reason
await args.DeadLetterMessageAsync(
    args.Message,
    deadLetterReason: "ValidationFailed",
    deadLetterErrorDescription: "Missing required field: CustomerId");

// Read from the dead-letter sub-queue
await using var dlqReceiver = client.CreateReceiver(
    "order-processing",
    new ServiceBusReceiverOptions
    {
        SubQueue = SubQueue.DeadLetter
    });

while (true)
{
    var message = await dlqReceiver.ReceiveMessageAsync(
        TimeSpan.FromSeconds(5), cancellationToken);
    if (message is null) break;

    logger.LogWarning(
        "DLQ message: {Reason} - {Description}",
        message.DeadLetterReason,
        message.DeadLetterErrorDescription);

    // Inspect, fix, and re-submit or discard
    await dlqReceiver.CompleteMessageAsync(message);
}

MassTransit 错误/故障队列

MassTransit 会自动创建 _error 和 _skipped 队列。重试次数耗尽后，失败的消息会连同故障元数据一起移入错误队列。

// Configure retry before dead-lettering
x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseMessageRetry(r => r.Intervals(
        TimeSpan.FromSeconds(1),
        TimeSpan.FromSeconds(5),
        TimeSpan.FromSeconds(15)));
});

DLQ 监控

始终监控 DLQ 深度并设置警报。未经监控的 DLQ 会悄无声息地不断积压，直到数据丢失或过时。

---

Saga / 流程管理器

Saga 用于协调跨服务的多步骤业务流程。每个步骤都会发布事件，以触发下一步；发生故障时，则通过补偿逻辑进行处理。

编排与协调

风格	工作方式	适用场景
协调式	服务独立响应事件；没有中央协调器	简单流程、步骤较少、耦合较松
编排式	Saga / 流程管理器指挥每个步骤	复杂流程、需要补偿、需要可见性

MassTransit 状态机 Saga

// Saga state
public class OrderState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = default!;
    public Guid OrderId { get; set; }
    public decimal Total { get; set; }
    public DateTime? PaymentReceivedAt { get; set; }
}

// State machine definition
public sealed class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State Submitted { get; private set; } = default!;
    public State PaymentPending { get; private set; } = default!;
    public State Completed { get; private set; } = default!;
    public State Faulted { get; private set; } = default!;

    public Event<OrderSubmitted> OrderSubmitted { get; private set; } = default!;
    public Event<PaymentReceived> PaymentReceived { get; private set; } = default!;
    public Event<PaymentFailed> PaymentFailed { get; private set; } = default!;

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => OrderSubmitted,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentReceived,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentFailed,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));

        Initially(
            When(OrderSubmitted)
                .Then(ctx =>
                {
                    ctx.Saga.OrderId = ctx.Message.OrderId;
                    ctx.Saga.Total = ctx.Message.Total;
                })
                .Publish(ctx => new RequestPayment(
                    ctx.Saga.OrderId, ctx.Saga.Total))
                .TransitionTo(PaymentPending));

        During(PaymentPending,
            When(PaymentReceived)
                .Then(ctx =>
                    ctx.Saga.PaymentReceivedAt = DateTime.UtcNow)
                .Publish(ctx => new FulfillOrder(ctx.Saga.OrderId))
                .TransitionTo(Completed),
            When(PaymentFailed)
                .Publish(ctx => new CancelOrder(ctx.Saga.OrderId))
                .TransitionTo(Faulted));
    }
}

// Registration -- requires MassTransit.EntityFrameworkCore package for EF persistence
// NuGet: MassTransit.EntityFrameworkCore Version="8.*"
builder.Services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .EntityFrameworkRepository(r =>
        {
            r.ExistingDbContext<SagaDbContext>();
            r.UsePostgres();
        });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });
});

Saga 持久化

存储	包	适用场景
Entity Framework Core	MassTransit.EntityFrameworkCore	已在使用 EF Core；需要事务
MongoDB	MassTransit.MongoDb	面向文档的状态；高吞吐量
Redis	MassTransit.Redis	临时 Saga；低延迟
内存中	内置	仅用于测试——重启后状态会丢失

补偿模式

Saga 步骤失败时，发布补偿命令以撤销之前的步骤：

OrderSubmitted -> RequestPayment -> PaymentReceived -> ReserveInventory
                                                          |
                                                     InventoryFailed
                                                          |
                                                    RefundPayment (compensation)
                                                          |
                                                    CancelOrder (compensation)

---

幂等消费者

至少一次投递意味着消费者可能会多次收到同一条消息。幂等消费者可确保重复处理产生相同结果。

基于数据库的去重

public sealed class IdempotentOrderConsumer(
    AppDbContext db,
    ILogger<IdempotentOrderConsumer> logger)
    : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        var messageId = context.MessageId
            ?? throw new InvalidOperationException("Missing MessageId");

        // Check if already processed
        var exists = await db.ProcessedMessages
            .AnyAsync(m => m.MessageId == messageId);

        if (exists)
        {
            logger.LogInformation(
                "Duplicate message {MessageId}, skipping", messageId);
            return;
        }

        // Process the message
        await ProcessOrderAsync(context.Message);

        // Record as processed
        db.ProcessedMessages.Add(new ProcessedMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow,
            ConsumerType = nameof(IdempotentOrderConsumer)
        });

        await db.SaveChangesAsync();
    }
}

自然幂等性

优先采用天然具备幂等性的操作：

• 插入或更新（INSERT ... ON CONFLICT UPDATE），而不是盲目插入
• 条件更新（UPDATE ... WHERE Status = 'Pending'），而不是无条件更新
• 使用根据消息内容生成的确定性 ID，而不是自动生成

---

消息信封模式

使用包含追踪、版本控制和路由元数据的标准信封包装消息负载。

public sealed record MessageEnvelope<T>(
    string MessageId,
    string MessageType,
    DateTimeOffset Timestamp,
    string CorrelationId,
    string Source,
    int Version, // Schema version for backward-compatible deserialization
    T Payload);

MassTransit 会通过 ConsumeContext（MessageId、CorrelationId、Headers）自动提供此功能。使用原始代理客户端时，请显式实现信封。

---

Agent 注意事项

1. 不要对 Azure Service Bus 使用自动完成——设置 AutoCompleteMessages = false，并在处理成功后调用 CompleteMessageAsync。自动完成会在处理结束前确认消息，发生故障时可能导致数据丢失。
2. 不要忘记处理毒消息——始终配置最大投递次数并监控 DLQ。若未配置这些选项，一条有问题的消息就可能无限期阻塞整个队列。
3. 不要在生产环境中使用内存中的 Saga 持久化——重启后 Saga 状态会丢失，导致业务流程处于未知状态。请使用 Entity Framework、MongoDB 或 Redis 持久化。
4. 不要假定分区之间的消息有序——默认情况下，竞争消费者和主题订阅会以乱序方式投递消息。顺序很重要时，请使用会话或分区。
5. 对于至少一次投递的消费者，不要跳过幂等性处理——代理可能会因超时、网络故障或消费者重启而重新投递消息。每个消费者都必须安全处理重复消息。
6. 不要硬编码连接字符串——请使用环境变量或 Azure Key Vault 引用。本地开发时，请使用用户机密或从源代码管理中排除的 .env 文件。

---

参考资料

• Azure Service Bus 文档
• .NET 的 Azure Service Bus 客户端库
• RabbitMQ .NET 客户端文档
• MassTransit 文档
• MassTransit Saga
• 企业集成模式