如何在Transaction Scope内同时向Azure Service Bus队列和主题发送消息

根本原因 #

Azure Service Bus 原生不支持单个本地事务跨多个顶层实体（队列、主题、订阅均属于顶层实体），你遇到的Local transactions cannot span multiple top-level entities such as queue or topic报错就是该限制的直接体现，无法通过直接修改事务配置绕过，需要调整实现方案来满足一致性要求。

可选解决方案 #

方案一：使用Service Bus官方跨实体事务（改动量最小） #

Service Bus 提供了*经由实体（Via Entity）*机制支持同一命名空间下的跨实体原子操作，适配你当前的场景：

• 先在同一个Service Bus命名空间下创建一个专用的经由队列，不需要额外开发消费逻辑，仅作为事务中转使用
• 同一个事务范围内所有发往不同队列、主题的消息，都指定同一个经由队列路径，Service Bus会保证这批消息的发送是原子性的，要么全部成功投递到目标实体，要么全部失败
• 只要你的数据库、文件存储资源支持加入Transaction Scope（如SQL Server、支持SMB 3.0+的Azure Files），就可以和这两个发送操作放在同一个事务范围内，实现整体原子性

参考实现代码（C#）：

await using var serviceBusClient = new ServiceBusClient("<你的Service Bus连接字符串>");
var transactionOptions = new TransactionOptions
{
    IsolationLevel = IsolationLevel.ReadCommitted,
    Timeout = TransactionManager.MaximumTimeout
};

using (var scope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled, transactionOptions))
{
    // 发消息到TopicA，指定经由队列
    var topicSender = serviceBusClient.CreateSender("TopicA");
    await topicSender.SendMessageAsync(
        new ServiceBusMessage("<TopicA消息内容>"), 
        new SendMessageOptions { ViaEntityPath = "<你创建的经由队列名称>" });
    
    // 发消息到QueueB，指定同一个经由队列
    var queueSender = serviceBusClient.CreateSender("QueueB");
    await queueSender.SendMessageAsync(
        new ServiceBusMessage("<QueueB消息内容>"), 
        new SendMessageOptions { ViaEntityPath = "<你创建的经由队列名称>" });
    
    // 执行剩余操作：移动共享存储文件、写入数据库
    /* 你的原有业务逻辑代码 */
    
    scope.Complete();
}

注意：该方案仅支持同一Service Bus标准层及以上层级的命名空间内的实体操作，不支持跨命名空间使用。

方案二：Saga模式实现最终一致性（适配所有场景） #

如果你的场景需要跨Service Bus命名空间、或者涉及不支持分布式事务的资源，可以采用补偿式Saga模式实现最终一致性：

• 新增事件状态表，记录每条QueueA消息的处理进度：业务逻辑执行完成、TopicA消息发送完成、QueueB消息发送完成、文件移动完成、数据库写入完成
• 每个步骤执行成功后更新对应状态，若某个步骤执行失败，按倒序执行已完成步骤的补偿操作：
  • 若文件移动失败，删除已发送到TopicA、QueueB的消息，回滚数据库写入
  • 若数据库写入失败，将文件移回原路径，删除已发送的两条Service Bus消息
• 配合Azure Function内置的重试机制处理瞬时错误，超过最大重试次数再触发补偿流程，减少不必要的回滚操作

线程安全保证 #

不管采用哪种方案，只要满足以下要求即可保证线程安全：

• 每个QueueA消息的处理流程对应独立的事务上下文/ Saga实例，不要在多个消息处理逻辑之间共享事务上下文、数据库连接等资源
• 数据库操作使用合适的事务隔离级别，避免脏读、不可重复读问题；文件操作时通过唯一文件名标识、文件锁等方式避免并发操作冲突

内容的提问来源于stack exchange，提问作者Wellington Zanelli