如何使用Npgsql配合EF Core实现读创事务规避并发竞态问题

你遇到的是典型的读-计算-写间隙产生的并发竞态问题：两个并发请求同时读到同一条最新用户记录，基于同一份数据计算出两条新记录插入，最终出现数据错乱、顺序冲突。普通无锁事务默认的READ COMMITTED隔离级别无法阻止这类问题，以下是两种生产可用、性能可控的实现方式，不需要加全表锁。

方案1：行级排他锁 + 事务（优先推荐，性能损耗极低） #

不需要锁整表，只需要在事务内对「目标用户对应的锚点行」加排他行锁即可：所有要给该用户新增记录的请求会在加锁步骤串行排队，不同用户的操作完全不冲突，性能影响可以忽略。
Npgsql provider 原生支持EF Core的ForUpdate()语法，不需要手写原生SQL，参考实现：

// 事务隔离级别用RepeatableRead即可，不需要更高的Serializable级别
using var transaction = await _dbContext.Database.BeginTransactionAsync(IsolationLevel.RepeatableRead);
try
{
    // 第一步：加锁，优先锁用户主表中对应用户的行——哪怕用户还没有业务记录，这行也存在，不会出现锁空的问题
    await _dbContext.Users
        .Where(u => u.Id == targetUserId)
        .ForUpdate() // 加排他行锁，其他同用户的请求执行到这里会阻塞，直到当前事务提交/回滚
        .FirstAsync();

    // 拿到锁之后再读取该用户最新的业务记录，此时读操作不会被其他同用户请求干扰
    var latestBizRecord = await _dbContext.UserBizData
        .Where(d => d.UserId == targetUserId)
        .OrderByDescending(d => d.Id) // 用自增主键/创建时间/业务版本号排序均可，保证取到最新行
        .FirstOrDefaultAsync();

    // 基于最新记录计算新行的值
    var newRecord = ComputeNewRecord(latestBizRecord, targetUserId);
    _dbContext.UserBizData.Add(newRecord);
    await _dbContext.SaveChangesAsync();

    await transaction.CommitAsync();
}
catch
{
    await transaction.RollbackAsync();
    throw;
}

注意事项 #

• 事务范围内不要放无关的长IO操作（比如调用第三方接口、读大文件），避免锁持有时间过长阻塞同用户的其他请求
• 如果没有独立的用户主表，可以新建一张轻量的用户序列锚点表，每个用户存一行记录专门用来加锁即可，不要直接锁最新业务行——用户首条记录插入时没有历史业务行，会出现锁空的问题。

方案2：唯一约束 + 冲突重试（无锁方案，适合低并发场景） #

不需要显式加锁，靠数据库唯一约束兜底拦截并发冲突，冲突后重试整个流程即可：

• 首先给业务表加用户维度的序号字段Sequence，标记该条记录是用户的第几条数据，配置联合唯一索引(UserId, Sequence)
• 每次读取最新记录后，新行的Sequence取最新记录值+1
• 插入时如果两个并发请求拿到了相同的Sequence，数据库会抛出唯一键冲突异常，捕获异常后重试整个读-算-写流程即可

参考实现：

// 首先在EF Core的模型配置中加唯一约束
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<UserBizData>()
        .HasIndex(d => new { d.UserId, d.Sequence })
        .IsUnique();
}

// 业务逻辑配置冲突重试策略，这里用Polly做重试，也可以自己写循环捕获
var retryPolicy = Policy
    .Handle<DbUpdateException>(ex => ex.InnerException is PostgresException pgEx 
        && pgEx.SqlState == PostgresErrorCodes.UniqueViolation)
    .WaitAndRetryAsync(3, _ => TimeSpan.FromMilliseconds(Random.Shared.Next(10,50))); // 加随机延迟避免活锁

await retryPolicy.ExecuteAsync(async () =>
{
    var latestRecord = await _dbContext.UserBizData
        .Where(d => d.UserId == targetUserId)
        .OrderByDescending(d => d.Sequence)
        .FirstAsync();
    var newRecord = new UserBizData
    {
        UserId = targetUserId,
        Sequence = latestRecord.Sequence + 1,
        // 其他计算字段赋值
    };
    _dbContext.UserBizData.Add(newRecord);
    await _dbContext.SaveChangesAsync();
});

注意事项 #

• 该方案无读阻塞，但单用户并发量超过10QPS时，冲突重试的概率会明显升高，反而会拉低整体性能
• 如果新行的计算逻辑很重（比如涉及大量计算、多次IO查数据），重试成本会很高，不适合用这个方案

不推荐的实现方式 #

• 全表锁：粒度过大，会阻塞所有用户的写入操作，性能极差，完全没有必要
• 只开事务不加锁/唯一约束：默认隔离级别下无法阻止两个事务同时读到同一条最新记录，必然会出现并发问题
• 应用层静态锁/分布式锁：静态锁无法覆盖多实例部署的场景，分布式锁的额外开销远高于数据库行锁，属于舍近求远。

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