我的 Saga 结构是正确的解决方案吗？(NServiceBus)

2026-10-24支付小周

结论

Saga 可以协调“营业日结束后生成银行文件”这类长时间流程，但不适合直接保存当天收到的所有消息。

建议采用以下设计：

Saga 只保存流程状态，如营业日期、截止时间、文件状态和统计数量。
每笔付款或取消请求都作为独立业务记录持久化。
营业结束 timeout 到达后，查询尚未导出的记录并生成银行文件。
使用消息 ID 或业务操作 ID 实现幂等处理，避免消息重试导致重复付款。

如果把不断增长的 List<T> 放入 Saga，系统会依赖具体 Saga persister 的集合映射能力。Saga 数据也会频繁增大，成为并发写入的热点。

当前数据类的问题

示例中的成员是字段，不是属性：

public virtual List<PaymentRequestedMessage> PaymentRequestedMessages;

C# 字段不能声明为 virtual。如果当前持久化方案支持集合，至少应改成属性：

public virtual IList<PaymentRequestedMessage> PaymentRequestedMessages
{
    get;
    set;
}

不过，语法正确并不代表 NServiceBus 的 Saga persister 一定能持久化该属性。不同版本和不同 persister 对复杂对象、集合、子实体及映射配置的支持并不相同。关系型持久化方案通常无法将任意消息对象集合自动映射到 Saga 表中。

即使能够映射，也不建议把完整消息保存在 Saga 中。Saga 每处理一条消息，都可能需要读取并更新整个集合。交易量增加后，记录会越来越大，并发冲突和超时的风险也会增加。

消息处理流程

1. 按营业日关联 Saga

当天收到的第一条付款或取消消息启动 Saga，后续消息通过 BusinessDate 关联到同一实例。

概念上的关联关系如下：

protected override void ConfigureHowToFindSaga(
    SagaPropertyMapper<DailyBankFileSagaData> mapper)
{
    mapper.MapSaga(saga => saga.BusinessDate)
        .ToMessage<PaymentRequested>(message => message.BusinessDate);

    mapper.MapSaga(saga => saga.BusinessDate)
        .ToMessage<PaymentCancelled>(message => message.BusinessDate);
}

具体关联 API 会随 NServiceBus 版本而变化，但设计上必须有稳定的批次标识。单独使用日期只适用于“一天一个批次”的场景。如果系统涉及多个银行、商户、币种或结算账户，应使用组合批次键，例如：

BankId + MerchantId + Currency + BusinessDate

最好把这些字段组合成单一的 BatchId，同时用于 Saga 关联和数据库查询。

2. 首次启动时安排 timeout

public async Task Handle(
    PaymentRequested message,
    IMessageHandlerContext context)
{
    await StorePaymentInstruction(message);

    Data.PaymentCount++;

    if (!Data.TimeoutScheduled)
    {
        Data.TimeoutScheduled = true;
        Data.CutoffTimeUtc = CalculateCutoffTimeUtc(message.BusinessDate);

        await RequestTimeout(
            context,
            Data.CutoffTimeUtc,
            new CloseBusinessDay
            {
                BusinessDate = message.BusinessDate
            });
    }
}

这里的 RequestTimeout 只是示意写法。旧版 NServiceBus 可能使用 Saga 上的 RequestTimeout<T>()，新版可能通过 IMessageHandlerContext 调用，具体应以项目使用的 NServiceBus 版本为准。

取消消息也按相同方式持久化：

public async Task Handle(
    PaymentCancelled message,
    IMessageHandlerContext context)
{
    await StoreCancellationInstruction(message);

    Data.CancellationCount++;

    if (!Data.TimeoutScheduled)
    {
        Data.TimeoutScheduled = true;
        Data.CutoffTimeUtc = CalculateCutoffTimeUtc(message.BusinessDate);

        await RequestTimeout(
            context,
            Data.CutoffTimeUtc,
            new CloseBusinessDay
            {
                BusinessDate = message.BusinessDate
            });
    }
}

3. timeout 到达后启动文件生成

public async Task Timeout(
    CloseBusinessDay timeout,
    IMessageHandlerContext context)
{
    if (Data.FileGenerationStarted)
    {
        return;
    }

    Data.FileGenerationStarted = true;

    await context.SendLocal(new GenerateBankFile
    {
        BusinessDate = Data.BusinessDate
    });
}

timeout 最好只发送 GenerateBankFile 命令，不要直接在 Saga handler 中生成文件。文件生成可能涉及大量查询、格式化、磁盘写入或远程存储，处理时间通常比普通消息更长。交给独立 handler 后，重试、监控和扩展都会更方便。

生成命令的 handler 再查询待处理记录：

public async Task Handle(
    GenerateBankFile message,
    IMessageHandlerContext context)
{
    var instructions = await repository.GetPendingInstructions(
        message.BusinessDate);

    var file = bankFileWriter.Create(instructions);

    await bankFileStore.Save(file);

    await repository.MarkAsExported(
        instructions.Select(x => x.Id));

    await context.Publish(new BankFileGenerated
    {
        BusinessDate = message.BusinessDate,
        InstructionCount = instructions.Count
    });
}

Saga 收到完成事件后结束：

public Task Handle(
    BankFileGenerated message,
    IMessageHandlerContext context)
{
    Data.FileGenerated = true;
    MarkAsComplete();

    return Task.CompletedTask;
}

必须处理的边界情况

截止时间附近到达的消息

timeout 和付款消息可能同时到达。不能认为 timeout 开始处理时，所有在截止时间前发送的消息都已经处理完成。不同队列、不同 endpoint 和消息重试都会影响实际到达顺序。

需要明确业务规则，例如：

根据消息中的业务时间确定所属批次。
截止时间之后收到的消息进入下一个营业日。
截止后保留一个宽限窗口。
生成文件时锁定批次，锁定后到达的消息进入补充文件。

批次归属不能只由消息实际进入处理流程的时间决定。

消息重复

NServiceBus 处理消息时可能进行重试，因此 handler 必须具备幂等性。至少应保存以下字段并设置唯一约束：

SourceMessageId

要求更严格的系统还应为业务操作建立唯一键，例如：

PaymentId + OperationType + OperationSequence

并发和批次锁定

生成文件前，应以原子操作将符合条件的记录从 Pending 改为 Exporting，锁定本次文件包含的内容。文件保存成功后，再将状态改为 Exported。

如果生成过程失败，系统需要能够根据批次号安全重试，避免重复生成内容不同的银行文件。

事务一致性

如果业务记录和 Saga 数据位于同一个支持事务的存储中，可以在同一次消息处理事务中提交。如果两者使用不同数据库，或者文件保存在外部存储中，就需要根据项目的 NServiceBus 版本和 transport 配置采用 Outbox、幂等写入或显式状态机。不能默认数据库写入、消息发送和文件保存天然属于同一事务。

Command 和 Event 的语义

“生成银行文件”要求一个明确的接收方执行操作，适合使用 Send 发送 command。

“付款已接受”“付款已取消”和“银行文件已生成”描述的是已经发生的事实，适合使用 Publish 发布 event。通常不应在接收 command 后将其原样再次发布。更清楚的流程是由 command handler 完成业务处理，再发布描述处理结果的 event。

如果确实需要在 Saga 中保存集合

只有当集合很小、具有明确上限，而且当前 Saga persister 明确支持相应映射时，才应考虑这种设计。集合中也应保存精简的值对象或记录 ID，不要保存完整的消息对象：

public class DailyBankFileSagaData : IContainSagaData
{
    public virtual Guid Id { get; set; }
    public virtual string Originator { get; set; }
    public virtual string OriginalMessageId { get; set; }

    public virtual IList<Guid> InstructionIds { get; set; }
}

这种方式仍然取决于 persister 是否支持集合映射以及相关配置。对于每天可能产生大量付款的系统，更合适的做法通常是独立保存每笔指令，让 Saga 只负责管理批次的生命周期。