迁移Access大表至SQL Server时遇TransactionScope嵌套事务报错

场景说明 #

需将Microsoft Access中包含30万条记录的超大表迁移至SQL Server，小批量（约1000条）导入时代码运行正常，但全量导入时执行await context.SaveChangesAsync()出现错误：TransactionScope "A root ambient transaction was completed before the nested transaction"。即使设置了超长超时、将SaveChanges放入循环，仍在接近完成时触发相同异常。

原实现代码 #

public async Task Sync()
{
    using TransactionScope trans = new(TransactionScopeOption.Required, new TransactionOptions()
    {
        IsolationLevel = IsolationLevel.ReadCommitted,
        Timeout = TimeSpan.FromMinutes(60),
    }, TransactionScopeAsyncFlowOption.Enabled);
    var options = new DbContextOptionsBuilder<FlushpanelContext>()
        .UseSqlServer(Environment.GetEnvironmentVariable("CS") ?? throw new Exception("CS envar not defined"))
        .Options;
    await using var context = new FlushpanelContext(options
    );
    context.Database.SetCommandTimeout(TimeSpan.FromHours(1));
    await context.Database.OpenConnectionAsync();
    try
    {
        await using OleDbConnection mydb1 = new($@"Provider=Microsoft.ACE.OLEDB.12.0;Data Source={its2000Path};");
        await mydb1.OpenAsync();

        var partsService = new AccessPartsService(context, its2000Cn, logger);
        await partsService.SyncPartsAsync();
        trans.Complete();
    }
    finally
    {
        await context.Database.CloseConnectionAsync();
    }
}

public async Task SyncPartsAsync()
{
    // Get all of the parts from access
    var accessParts = await accessCn.QueryAsync<Access.Models.Part>("SELECT * FROM parts WHERE Imported = 0");
    logger.WriteLine($"{accessParts.Count()} parts to be imported in access", Color.DarkBlue);
    var partTypes = await context.PartType
        .Where(x => x.AccessTableId != null && x.AccessId != null)
        .ToDictionaryAsync(x => (x.AccessTableId!.Value, x.AccessId!.Value));
    var tablePartTypes = await context.PartType.Where(x => x.AccessTableId == null && x.AccessId == null)
        .ToDictionaryAsync(x => x.Title);
    await ProcessBatch(accessParts, alreadyExistingSqlParts, manufacturers, partTypes, tablePartTypes);
    logger.WriteLine("Saving changes", Color.Purple);
    await context.SaveChangesAsync();
    await accessCn.ExecuteAsync("UPDATE parts SET Imported = 1 WHERE Imported = 0");
    //await transaction.CommitAsync();
}

private async Task ProcessBatch(IEnumerable<Access.Models.Part> accessParts, Dictionary<string, int> alreadyExistingSqlParts, FrozenDictionary<string, int> manufacturers,
    Dictionary<(int accessTableId, int accessId), PartType> partTypes, Dictionary<string, PartType> tablePartTypes)
{
    // This will run for about 300,000 parts
    foreach (var accessPart in accessParts)
    {
            var converted = accessPart.ToPart(manufacturers);
            context.Part.Add(converted);
            logger.WriteLine($"Added part {accessPart.Part_ID}", Color.DarkGreen);
    }
}

问题根源分析 #

1. 分布式事务资源耗尽：TransactionScope会启动分布式事务，同时管理Access和SQL Server两个连接，30万条数据的超长处理时间会触发系统级分布式事务超时（即使显式设置60分钟，系统默认DTC超时可能更短）。
2. EF上下文内存过载：一次性将30万条实体加入上下文追踪，会占用大量内存，引发GC频繁回收、性能骤降，间接导致事务异常。
3. 跨连接事务的不稳定交互：OleDb连接与SQL Server连接在分布式事务中长时间保持打开状态，容易触发事务管理器的资源回收机制。

解决方案 #

1. 替换TransactionScope为EF原生事务 #

放弃分布式事务，改用EF Core原生数据库事务，减少跨连接事务的管理开销：

public async Task Sync()
{
    var options = new DbContextOptionsBuilder<FlushpanelContext>()
        .UseSqlServer(Environment.GetEnvironmentVariable("CS") ?? throw new Exception("CS envar not defined"))
        .Options;
    await using var context = new FlushpanelContext(options);
    context.Database.SetCommandTimeout(TimeSpan.FromHours(1));
    
    await using OleDbConnection mydb1 = new($@"Provider=Microsoft.ACE.OLEDB.12.0;Data Source={its2000Path};");
    await mydb1.OpenAsync();

    var partsService = new AccessPartsService(context, mydb1, logger);
    // 使用EF原生事务
    using var transaction = await context.Database.BeginTransactionAsync();
    try
    {
        await partsService.SyncPartsAsync();
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}

2. 分批次处理数据，避免内存过载 #

不要一次性加载全量数据，分批次读取、处理、保存，同时清空上下文追踪释放内存：

public async Task SyncPartsAsync()
{
    int batchSize = 1000; // 每次处理1000条，可根据服务器性能调整
    int offset = 0;
    bool hasMoreRecords = true;
    logger.WriteLine("开始分批次导入零件数据", Color.DarkBlue);

    // 预加载字典数据，避免循环重复查询
    var partTypes = await context.PartType
        .Where(x => x.AccessTableId != null && x.AccessId != null)
        .ToDictionaryAsync(x => (x.AccessTableId!.Value, x.AccessId!.Value));
    var tablePartTypes = await context.PartType.Where(x => x.AccessTableId == null && x.AccessId == null)
        .ToDictionaryAsync(x => x.Title);

    do
    {
        // 分批次读取Access数据，避免一次性加载30万条
        var accessParts = await accessCn.QueryAsync<Access.Models.Part>(
            "SELECT * FROM parts WHERE Imported = 0 ORDER BY Part_ID LIMIT @Offset, @BatchSize",
            new { Offset = offset, BatchSize = batchSize });
        
        hasMoreRecords = accessParts.Any();
        if (!hasMoreRecords) break;

        logger.WriteLine($"当前批次导入 {accessParts.Count()} 条零件", Color.DarkBlue);
        await ProcessBatch(accessParts, alreadyExistingSqlParts, manufacturers, partTypes, tablePartTypes);
        await context.SaveChangesAsync();
        // 清空上下文追踪，释放内存
        context.ChangeTracker.Clear();

        // 标记当前批次为已导入，避免重复处理
        await accessCn.ExecuteAsync(
            "UPDATE parts SET Imported = 1 WHERE Part_ID IN @PartIds",
            new { PartIds = accessParts.Select(p => p.Part_ID).ToList() });

        offset += batchSize;
    } while (hasMoreRecords);

    logger.WriteLine("所有零件导入完成", Color.Purple);
}

3. 优化Access端性能 #

• 给Access的parts表的Imported和Part_ID字段建立索引，提升查询和更新速度。
• 避免使用SELECT *，只读取迁移所需的字段，减少数据传输量。

4. 可选：调整系统DTC超时（不推荐） #

如果必须使用TransactionScope，需调整Windows分布式事务协调器（DTC）的超时设置：

• 打开组件服务 → 计算机 → 我的电脑 → 分布式事务协调器 → 本地DTC
• 右键属性 → 事务超时，设置更长时间（如180分钟）
• 重启DTC服务

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