如何在EF Core中并发替换指定团队的成员数据集？

实体定义 #

DbContext中包含以下实体：

public class TeamEntity 
{ 
    public int Id { get; set; } 
    public string Key { get; set; } 
    public ICollection<TeamMemberEntity> Members { get; set; } 
}
public class TeamMemberEntity 
{ 
    public int Id { get; set; } 
    public string UserKey { get; set; } 
    public int TeamId { get; set; } 
    public TeamEntity Team { get; set; }
}

需求说明 #

实现一个方法，接收teamKey和memberUserKeys参数，将指定团队的现有成员全部替换为memberUserKeys对应的新成员。

并发场景下需保证：当两个请求并行针对同一个teamKey传入不同的memberUserKeys时，最后执行的写入操作生效。示例：

UpdateTeamMembers("key1", [1,2,3]);
UpdateTeamMembers("key1", [4,5,6]);
// 允许结果: 成员为[1,2,3]或[4,5,6]
// 禁止结果: 成员为[1,2,3,4,5,6]

现有实现问题 #

当前实现的并发处理效果不佳，可能出现成员集合合并的错误结果：

await using var transaction = await dbContext.Database.BeginTransactionAsync();
try
{
    var team = await dbContext.Teams
        .AsNoTracking()
        .Where(x => x.Key == teamKey)
        .Select(x => new {
            x.Id
        })
        .FirstOrDefaultAsync(cancellationToken);

    if (team is null)
        throw new NotFoundException("Team was not found");

    var teamId = team.Id;

    // 删除当前所有团队成员
    await dbContext.TeamMembers
        .Where(tm => tm.TeamId == teamId)
        .ExecuteDeleteAsync(cancellationToken);

    // 添加新成员
    if (memberUserKeys.Length > 0)
    {
        await dbContext.TeamMembers.AddRangeAsync(
            memberUserKeys.Select(x => new TeamMemberEntity()
            {
                TeamId = teamId,
                UserKey = x,
            }), cancellationToken);
        await dbContext.SaveChangesAsync(cancellationToken);
    }

    await transaction.CommitAsync(cancellationToken);

    return Unit.Value;
}
catch (Exception ex)
{
    // 回滚逻辑
}

问题根源在于：无锁查询团队后，并发请求可能同时进入删除和添加流程，导致前一个请求的添加操作在第二个请求的删除操作之后执行，最终两个请求的成员都被保留。

优化方案 #

方案1：悲观锁（行级锁） #

通过在查询团队时添加悲观写锁，确保同一时间只有一个请求能修改该团队的成员：

await using var transaction = await dbContext.Database.BeginTransactionAsync();
try
{
    // 锁定团队行，阻止其他并发请求修改相关数据
    var team = await dbContext.Teams
        .Where(x => x.Key == teamKey)
        .Select(x => new { x.Id })
        .LockMode(LockMode.PessimisticWrite) // EF Core 5+支持，不同数据库语法可能有差异
        .FirstOrDefaultAsync(cancellationToken);

    if (team is null)
        throw new NotFoundException("Team was not found");

    var teamId = team.Id;

    // 删除所有现有成员
    await dbContext.TeamMembers
        .Where(tm => tm.TeamId == teamId)
        .ExecuteDeleteAsync(cancellationToken);

    // 添加新成员
    if (memberUserKeys.Any())
    {
        var newMembers = memberUserKeys.Select(x => new TeamMemberEntity
        {
            TeamId = teamId,
            UserKey = x
        });
        await dbContext.TeamMembers.AddRangeAsync(newMembers, cancellationToken);
        await dbContext.SaveChangesAsync(cancellationToken);
    }

    await transaction.CommitAsync(cancellationToken);

    return Unit.Value;
}
catch (Exception ex)
{
    await transaction.RollbackAsync(cancellationToken);
    throw;
}

说明：悲观锁会在查询时锁定团队对应的数据库行，其他请求必须等待当前事务提交后才能继续执行，从根本上避免并发冲突，最终结果必然是最后一个完成的请求的成员集合。

方案2：乐观锁（版本号机制） #

在TeamEntity中添加版本号字段，利用EF Core的并发检测机制处理冲突：

1. 修改实体与上下文配置：

// 更新TeamEntity
public class TeamEntity 
{ 
    public int Id { get; set; } 
    public string Key { get; set; } 
    public ICollection<TeamMemberEntity> Members { get; set; } 
    public int Version { get; set; } // 添加并发版本号
}

// 在DbContext的OnModelCreating中配置
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<TeamEntity>()
        .Property(t => t.Version)
        .IsConcurrencyToken(); // 标记为并发令牌
}

2. 修改替换方法，添加冲突重试逻辑：

public async Task<Unit> UpdateTeamMembers(string teamKey, string[] memberUserKeys, CancellationToken cancellationToken)
{
    while (true)
    {
        try
        {
            await using var dbContext = new YourDbContext(); // 每次重试创建新上下文
            await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken);

            var team = await dbContext.Teams
                .Where(x => x.Key == teamKey)
                .FirstOrDefaultAsync(cancellationToken);

            if (team is null)
                throw new NotFoundException("Team was not found");

            var teamId = team.Id;

            // 删除现有成员
            await dbContext.TeamMembers
                .Where(tm => tm.TeamId == teamId)
                .ExecuteDeleteAsync(cancellationToken);

            // 添加新成员
            if (memberUserKeys.Any())
            {
                var newMembers = memberUserKeys.Select(x => new TeamMemberEntity
                {
                    TeamId = teamId,
                    UserKey = x
                });
                await dbContext.TeamMembers.AddRangeAsync(newMembers, cancellationToken);
            }

            // 更新版本号，触发并发检测
            team.Version++;
            await dbContext.SaveChangesAsync(cancellationToken);

            await transaction.CommitAsync(cancellationToken);
            return Unit.Value;
        }
        catch (DbUpdateConcurrencyException)
        {
            // 并发冲突，自动重试
            continue;
        }
        catch (Exception ex)
        {
            throw;
        }
    }
}

说明：乐观锁不主动锁定资源，而是在保存更改时检查版本号是否与查询时一致。如果并发修改导致版本号变化，会抛出DbUpdateConcurrencyException，此时通过重试确保最终执行的是最后一次请求的操作。适合并发冲突频率较低的场景，性能开销更小。

方案选择建议 #

• 若团队成员修改操作并发频率高，优先选择悲观锁，避免频繁重试带来的开销。
• 若并发频率低，乐观锁是更轻量的方案，不会影响正常请求的响应速度。

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