PostgreSQL+EF Core 8事务并发锁问题求解决方案

场景示例 #

存在三层实体类A、B、C，层级关系为A包含多个B，B包含多个C，实体定义如下：

public class A
{
    public Guid Id { get; set; }
    public List<B> Bs { get; set; } = [];
    public string Name { get; set; }
}

public class B
{
    public Guid Id { get; set; }
    public Guid AId { get; set; }
    public A A { get; set; }
    public List<C> Cs { get; set; } = [];
}

public class C
{
    public Guid Id { get; set; }
    public Guid BId { get; set; }
    public B B { get; set; }
    public string Name { get; set; }
}

业务需求 #

修改A的请求需满足以下条件：

• A必须包含3个B实例
• 每个B实例都包含1个名称以小写字母开头的C实例

当前实现（PostgreSQL + EF Core 8） #

当前实现逻辑：

MainDbContext db1 = ...;
A a = db1.A.First(...);

await using var t1 = await db1.Database.BeginTransactionAsync(IsolationLevel.ReadUncommitted);
a.Name = "Anew";

await db1.SaveChangesAsync();

if (db1.As.Any(m => m.Id == aId && m.Bs.Count() == 3 && m.Bs.All(n => n.Cs.Count == 1 && n.Cs.First().Name == "c")))
    await t1.CommitAsync();
else
    await t1.RollbackAsync();

问题描述 #

事务提交/回滚前存在时间间隙，期间其他进程可能对B或C执行删除、添加或修改操作，导致条件判断失效，做出错误的提交/回滚决策。

尝试过的方案 #

曾尝试使用Serializable隔离级，但该级别要求所有请求都使用才能生效，无法满足当前场景需求。

完整测试代码 #

using System.Data;
using System.Data.Common;
using System.Text.RegularExpressions;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
builder.Services.AddDbContext<MainDbContext>(m => m.UseNpgsql(builder.Configuration.GetConnectionString("DB")));
var app = builder.Build();
var db1 = app.Services.CreateScope().ServiceProvider.GetRequiredService<MainDbContext>();
var db2 = app.Services.CreateScope().ServiceProvider.GetRequiredService<MainDbContext>();
var db3 = app.Services.CreateScope().ServiceProvider.GetRequiredService<MainDbContext>();
await db3.Database.EnsureCreatedAsync();
await db3.Database.EnsureDeletedAsync();
await db3.Database.MigrateAsync();
var aId = Guid.NewGuid();
db3.As.Add(new A()
{
    Id = aId,
    Name = "A",
    Bs =
    [
        new B()
        {
            Cs =
            [
                new C()
                {
                    Name = "c"
                }
            ]
        },
        new B()
        {
            Cs =
            [
                new C()
                {
                    Name = "c"
                }
            ]
        },
        new B()
        {
            Cs =
            [
                new C()
                {
                    Name = "c"
                }
            ]
        }
    ]

});
await db3.SaveChangesAsync();

if (Equals(db1, db2))
    throw new InvalidOperationException();
await using var t1 = await db1.Database.BeginTransactionAsync(IsolationLevel.ReadUncommitted);
await using var t2 = await db2.Database.BeginTransactionAsync(IsolationLevel.ReadUncommitted);
var aFromDb1 = await db1.As.Include(m => m.Bs).FirstAsync();
var aFromDb2 = await db2.As.Include(m => m.Bs).FirstAsync();

aFromDb1.Name = "Anew";
await db1.SaveChangesAsync();

db2.Cs.Where(m=> m.BId == db2.Bs.First().Id).ExecuteDelete();
await t2.CommitAsync();

if (db1.As.Any(m => m.Id == aId && m.Bs.Count() == 3 && m.Bs.All(n => n.Cs.Count == 1 && n.Cs.First().Name == "c")))
    await t1.CommitAsync();
else
    await t1.RollbackAsync();

return;


public class MainDbContext : DbContext
{
    public DbSet<A> As { get; set; }
    public DbSet<B> Bs { get; set; }
    public DbSet<C> Cs { get; set; }

    public MainDbContext(DbContextOptions<MainDbContext> dbContextOptions) : base(dbContextOptions)
    {
    }
}

public class A
{
    public Guid Id { get; set; }
    public List<B> Bs { get; set; } = [];
    
    public string Name { get; set; }
}

public class B
{
    public Guid Id { get; set; }
    public Guid AId { get; set; }
    public A A { get; set; }

    public List<C> Cs { get; set; } = [];
}

public class C
{
    public Guid Id { get; set; }

    public Guid BId { get; set; }
    public B B { get; set; }

    public string Name { get; set; }
}

解决方案 #

方案1：使用乐观锁（版本号） #

在实体A、B、C中添加版本号字段，EF Core会自动处理并发冲突检测：

public class A
{
    public Guid Id { get; set; }
    public List<B> Bs { get; set; } = [];
    public string Name { get; set; }
    [Timestamp]
    public byte[] Version { get; set; }
}

// B和C类同样添加[Timestamp]属性的Version字段

修改逻辑时，先加载包含关联数据的A，验证条件后再执行更新，若期间有其他进程修改相关数据，SaveChangesAsync会抛出DbUpdateConcurrencyException，此时捕获异常并回滚：

await using var t1 = await db1.Database.BeginTransactionAsync();
var a = await db1.As.Include(m => m.Bs).ThenInclude(n => n.Cs)
                  .FirstAsync(m => m.Id == aId);

// 先验证条件
if (a.Bs.Count != 3 || !a.Bs.All(n => n.Cs.Count == 1 && char.IsLower(n.Cs.First().Name[0])))
{
    await t1.RollbackAsync();
    return;
}

a.Name = "Anew";
try
{
    await db1.SaveChangesAsync();
    await t1.CommitAsync();
}
catch (DbUpdateConcurrencyException)
{
    await t1.RollbackAsync();
    // 处理并发冲突，比如重试或返回错误
}

方案2：在事务中锁定相关资源 #

使用PostgreSQL的SELECT ... FOR UPDATE语句锁定A及其关联的B、C数据，确保事务期间其他进程无法修改这些数据。EF Core中可通过FromSqlRaw实现：

await using var t1 = await db1.Database.BeginTransactionAsync(IsolationLevel.RepeatableRead);

// 锁定A、关联的B和C
var a = await db1.As.FromSqlRaw(@"
    SELECT * FROM ""As"" 
    WHERE ""Id"" = {0} 
    FOR UPDATE
", aId)
.Include(m => m.Bs)
.ThenInclude(n => n.Cs)
.FirstAsync();

// 验证条件
if (a.Bs.Count != 3 || !a.Bs.All(n => n.Cs.Count == 1 && char.IsLower(n.Cs.First().Name[0])))
{
    await t1.RollbackAsync();
    return;
}

a.Name = "Anew";
await db1.SaveChangesAsync();
await t1.CommitAsync();

FOR UPDATE会锁定查询到的行，直到事务结束，其他进程修改这些行时会被阻塞，直到当前事务提交或回滚。

方案3：将更新与条件判断合并为原子操作 #

直接编写SQL语句，将条件判断和更新合并为一个原子操作，避免中间间隙：

await using var t1 = await db1.Database.BeginTransactionAsync();

var affectedRows = await db1.Database.ExecuteSqlAsync(@"
    UPDATE ""As"" 
    SET ""Name"" = 'Anew'
    WHERE ""Id"" = {0}
    AND (
        SELECT COUNT(*) FROM ""Bs"" WHERE ""AId"" = {0}
    ) = 3
    AND NOT EXISTS (
        SELECT 1 FROM ""Bs"" 
        WHERE ""AId"" = {0}
        AND (
            SELECT COUNT(*) FROM ""Cs"" WHERE ""BId"" = ""Bs"".""Id""
        ) != 1
        OR (
            SELECT ""Name"" FROM ""Cs"" WHERE ""BId"" = ""Bs"".""Id"" LIMIT 1
        ) !~ '^[a-z]'
    )
", aId);

if (affectedRows > 0)
{
    await t1.CommitAsync();
}
else
{
    await t1.RollbackAsync();
}

这个SQL会在同一个原子操作中完成条件验证和更新，不存在中间间隙，只有当所有条件满足时才会更新A的名称。

内容的提问来源于stack exchange，提问作者Иван Иван