如何在.NET中处理基于Task并行调用的多SQL Server实例异步分布式事务异常回滚问题

嘿，这个场景正好是分布式事务里典型的「部分成功、部分失败」问题，用**两阶段提交（2PC）**来处理完全适配，结合.NET的Task并行机制，我给你梳理一个可落地的具体实现方案：

2PC的核心就是把整个事务拆成两个阶段：

1. 准备阶段：让所有参与的服务/Task先执行「可撤销的预操作」，并反馈自己是否能完成最终提交；
2. 提交/回滚阶段：如果所有参与者都准备成功，就统一触发提交；只要有任何一个参与者失败，就让所有已经准备成功的参与者执行回滚。

对应到你的场景：Task1（应用DB）、Task2（集中式DB）、Task3（迁移DB）都需要支持「准备-提交-回滚」三个步骤，然后由一个协调器来统一调度这两个阶段。

1. 给每个数据库操作封装标准的三阶段方法 #

首先要为每个Task对应的数据库操作定义统一的接口，确保它们都能执行准备、提交、回滚动作：

// 定义分布式操作的标准接口
public interface IDistributedDbOperation
{
    // 准备：执行可撤销的预写入操作，返回是否准备成功
    Task<bool> PrepareAsync();
    // 提交：确认预操作，将数据永久写入
    Task CommitAsync();
    // 回滚：撤销预操作，恢复到初始状态
    Task RollbackAsync();
}

以Task1的应用DB操作为例，实现这个接口：

public class ApplicationDbOperation : IDistributedDbOperation
{
    private bool _isPrepared = false;
    // 保存预备操作的上下文（比如事务句柄、临时数据标识）
    private SqlTransaction _pendingTransaction;
    private SqlConnection _dbConnection;

    public async Task<bool> PrepareAsync()
    {
        Console.WriteLine("开始准备应用DB的写入操作...");
        try
        {
            _dbConnection = new SqlConnection("你的应用DB连接字符串");
            await _dbConnection.OpenAsync();
            // 开启一个未提交的事务，作为预备操作
            _pendingTransaction = _dbConnection.BeginTransaction();
            
            // 执行实际的写入逻辑（但不提交事务）
            var cmd = new SqlCommand("INSERT INTO ...", _dbConnection, _pendingTransaction);
            await cmd.ExecuteNonQueryAsync();
            
            _isPrepared = true;
            return true;
        }
        catch (Exception ex)
        {
            Console.WriteLine($"应用DB准备失败：{ex.Message}");
            CleanupResources();
            return false;
        }
    }

    public async Task CommitAsync()
    {
        if (!_isPrepared || _pendingTransaction == null) return;
        Console.WriteLine("提交应用DB的写入操作...");
        try
        {
            _pendingTransaction.Commit();
            _isPrepared = false;
        }
        finally
        {
            CleanupResources();
        }
    }

    public async Task RollbackAsync()
    {
        if (!_isPrepared || _pendingTransaction == null) return;
        Console.WriteLine("回滚应用DB的写入操作...");
        try
        {
            _pendingTransaction.Rollback();
            _isPrepared = false;
        }
        finally
        {
            CleanupResources();
        }
    }

    private void CleanupResources()
    {
        _pendingTransaction?.Dispose();
        _dbConnection?.Close();
        _dbConnection?.Dispose();
    }
}

同样的，你可以为Task2（集中式DB）和Task3（迁移DB）实现IDistributedDbOperation接口，逻辑类似——都是在准备阶段开启未提交事务执行写入，提交时确认事务，回滚时撤销事务。

2. 实现2PC协调器，统一调度并行任务 #

接下来写一个协调器类，负责触发准备阶段、收集结果，然后决定是提交还是回滚：

public class TwoPhaseCommitCoordinator
{
    private readonly List<IDistributedDbOperation> _dbOperations;

    public TwoPhaseCommitCoordinator(params IDistributedDbOperation[] operations)
    {
        _dbOperations = operations.ToList();
    }

    public async Task<bool> ExecuteTransactionAsync()
    {
        // 第一阶段：并行执行所有准备操作
        var prepareTasks = _dbOperations.Select(op => op.PrepareAsync()).ToList();
        var prepareResults = await Task.WhenAll(prepareTasks);

        // 检查所有准备是否都成功
        if (prepareResults.All(result => result))
        {
            // 第二阶段：全部准备成功，并行提交所有操作
            var commitTasks = _dbOperations.Select(op => op.CommitAsync()).ToList();
            await Task.WhenAll(commitTasks);
            Console.WriteLine("所有数据库操作提交成功！");
            return true;
        }
        else
        {
            // 第二阶段：有操作准备失败，回滚所有已成功准备的操作
            var rollbackTasks = _dbOperations
                .Where((op, index) => prepareResults[index]) // 只回滚准备成功的操作
                .Select(op => op.RollbackAsync())
                .ToList();
            await Task.WhenAll(rollbackTasks);
            Console.WriteLine("部分操作失败，已回滚所有成功准备的操作！");
            return false;
        }
    }
}

3. 调用示例 #

最后在业务代码里创建三个操作实例，交给协调器执行即可：

public static async Task Main(string[] args)
{
    // 初始化三个数据库操作实例
    var appDbOp = new ApplicationDbOperation();
    var centralDbOp = new CentralizedDbOperation();
    var migrationDbOp = new MigrationDbOperation();

    // 创建协调器并执行2PC事务
    var coordinator = new TwoPhaseCommitCoordinator(appDbOp, centralDbOp, migrationDbOp);
    await coordinator.ExecuteTransactionAsync();
}

• 幂等性保障：准备、提交、回滚方法必须是幂等的——比如重复调用回滚不会导致额外错误，因为网络波动可能让协调器重试操作。
• 容错与日志：一定要记录每个阶段的操作状态（比如哪个操作准备成功/失败），如果协调器在提交阶段崩溃，后续可以通过日志进行补偿操作（比如手动提交或回滚）。
• 替代方案：System.Transactions 如果你的数据库支持分布式事务（比如SQL Server配合MSDTC），也可以用TransactionScope结合异步Task（需要启用TransactionScopeAsyncFlowOption.Enabled），但这种方式依赖MSDTC的配置，跨环境部署可能受限，手动实现2PC则更灵活可控。

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