如何在LINQ to Entities中避免多线程读取SQL数据库重复记录

问题根因 #

ReadCommitted隔离级别的核心约束仅为禁止事务读取其他事务未提交的变更，并不会对普通读取操作加排他锁。你当前的逻辑存在「读取-修改」的时间差窗口：线程1先读取了IsRead=false的文件记录，此时还未执行SaveChanges提交修改，线程2读取时这些记录的IsRead仍为false，也不存在未提交的变更，完全符合ReadCommitted的读取规则，因此必然会出现重复读取的问题。

解决方案 #

下面提供三种可落地的实现方式，按推荐优先级排序：

方案1：原子更新直接返回记录（最高效） #

利用数据库的原子更新特性，单条SQL完成「标记已读+返回待处理记录」的操作，全程无并发窗口，不需要额外控制事务隔离级别：
你可以直接在EF中执行带OUTPUT子句的SQL（以SQL Server为例）：

using (var context = new Context())
{
    // 原子更新，直接返回被标记的记录，不会出现并发冲突
    var processedFiles = context.File
        .FromSqlRaw(@"UPDATE File 
                      SET IsRead = 1 
                      OUTPUT inserted.* 
                      WHERE IsRead = 0")
        .ToList();
    
    // 事务提交后再处理文件，不要把文件IO放到事务内
    foreach (var file in processedFiles)
    {
        // 执行文件处理逻辑
    }
}

方案2：悲观并发加锁控制 #

如果需要保留先查询再处理的逻辑，可以通过查询加更新锁+跳过已锁定记录的方式避免重复读取：

using (var context = new Context())
{
    var transOptions = new TransactionOptions
    {
        IsolationLevel = IsolationLevel.ReadCommitted
    };
    using (var tran = new TransactionScope(TransactionScopeOption.Required, transOptions))
    {
        try
        {
            // UPDLOCK：给查询到的记录加更新锁，其他事务无法修改/加更新锁
            // READPAST：跳过已经被其他事务加锁的记录，不会阻塞等待
            var pendingFiles = context.File
                .FromSqlRaw("SELECT * FROM File WITH (UPDLOCK, READPAST) WHERE IsRead = 0")
                .ToList();

            foreach (var file in pendingFiles)
            {
                file.IsRead = true;
                context.Entry(file).State = EntityState.Modified;
            }
            // 批量提交，不要放到循环内逐次提交，缩短事务窗口
            context.SaveChanges();
            tran.Complete();
        }
        finally
        {
            // 事务结束后再处理文件
        }
    }
}

方案3：乐观并发控制 #

给File表添加并发令牌字段，更新时校验版本，过滤已被其他线程处理的记录：

1. 给File实体添加时间戳字段：

public class File
{
    // 其他字段省略
    [Timestamp]
    public byte[] RowVersion { get; set; }
}

2. 处理逻辑中捕获并发冲突，跳过已被处理的记录：

using (var context = new Context())
{
    var pendingFiles = context.File.Where(f => !f.IsRead).ToList();
    foreach (var file in pendingFiles)
    {
        try
        {
            file.IsRead = true;
            context.SaveChanges();
            // 处理当前文件
        }
        catch (DbUpdateConcurrencyException)
        {
            // 并发冲突说明该文件已被其他线程处理，直接跳过
            context.Entry(file).Reload();
            continue;
        }
    }
}

注意事项 #

• 禁止在数据库事务内部执行文件IO操作，文件IO耗时不稳定会拉长事务持有锁的时间，严重影响数据库性能
• 建议新增ProcessStatus（待处理/处理中/处理成功/处理失败）和LastUpdateTime字段，避免线程处理过程中崩溃导致任务丢失，可定期扫描超时的处理中任务重新执行
• 优先使用批量提交代替循环内逐次提交，减少事务交互次数，缩短锁持有时间

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