【EF Core】 EF Core并发控制：乐观锁与悲观锁的应用

文章目录

前言
一、并发的风险
二、EF Core中的并发控制方式
- 2.1 开放式并发（乐观锁）
- - 2.1.1 应用程序管理的属性并发令牌
  - 2.1.2 数据库生成的并发令牌
- 2.2 悲观锁
总结

前言

实际的生产环境中，我们经常能遇到数据库由多个应用程序同时使用。每个程序作为一个实例都会对数据进行操作。其中如果不同的实例同时对同一个数据进行了修改，这就会遇到我们常见的一个问题——并发。

这里讨论的并发控制，也就是数据库管理系统中用于处理多个事务同时访问和修改共享数据时可能产生问题的机制。其主要目标是确保数据的一致性、完整性和隔离性，同时提高系统的性能和吞吐量。

一、并发的风险

前面我们了解到并发情况下，多个操作同时修改共享资源，容易导致数据不一致或逻辑错误。

比如银行转账，用户 A 转账给用户 B，用户C也转账给用户B。假设用户 B 初始余额为 100 元，A 转 50 元，C 转 30 元。若两个转账操作并发执行，可能出现以下问题：

两个事务同时读取用户 B 的余额为 100 元。
事务 A 将余额更新为 150 元并提交。
事务 C 将余额更新为 130 元并提交，覆盖事务 A 的更新，最终余额变为 130 元（而非正确的 180 元）。

这就是多个操作同时修改共享资源时，可能导致数据不一致的问题。是一个典型的丢失更新的例子

除了并发会导致丢失更新外，还有以下几种情况：

脏读（Dirty Read）：一个事务读取了另一个未提交事务修改的数据。
不可重复读（Non-Repeatable Read）：同一事务中多次读取同一数据却得到不同结果，因为其他事务已提交了修改。
幻读（Phantom Read）：一个事务按相同条件多次查询，却返回了不同的记录集，原因是其他事务插入或删除了数据。
丢失更新（Lost Update）：两个事务同时修改同一数据，后提交的事务覆盖了先提交事务的修改，导致部分修改丢失。

并发编程中处理并发冲突的两种不同策略，悲观锁与乐观锁。

悲观锁：假设冲突一定会发生，因此在操作数据前先获取锁，确保同一时间只有自己能修改数据，其他事务需等待锁释放。【比如银行转账】
乐观锁：假设冲突很少发生，不预先加锁，而是在提交更新时检查数据是否被其他事务修改过

二、EF Core中的并发控制方式

EFCore开发团队实现了乐观并发并集成在了EF Core里，这也是 EF Core 处理并发最常用的方式。乐观并发不会在数据库层面把数据进行锁定，但如果数据在查询后发生更改，该数据修改后会在保存时失败。
在 EF Core 中，乐观并发的实现是通过给属性设置一个并发令牌。该并发令牌会在查询实体时进行加载和跟踪，和数据的其他属性一样。然后，在 SaveChanges() 期间执行更新或删除操作时，数据库中的并发令牌值将与 EF Core 读取的原始值进行比较。
当然我们也可以监测多个属性值，把一整条数据的任何变化作为监测的对象，是一条数据的并发令牌。比起前者单个属性来说，行的并发令牌需要显示设置属性字段。这个我们后续一一讨论。

2.1 开放式并发（乐观锁）

前面我们了解到EFCore开发团队实现了乐观并发并集成在了EF Core里，设置一个并发令牌，保存的时候通过DbUpdateConcurrencyException来捕获异常，从而判断值是否已经被修改。对于单个属性和多个属性的检测，EFCore分别提供ConcurrencyToken和RowVersion两者方式来生成令牌。

ConcurrencyToken是应用程序管理的并发令牌
RowVersion是本机数据库生成的并发令牌
下面分别介绍二者

2.1.1 应用程序管理的属性并发令牌

设想对站点信息的修改的一个场景，两个实例修改同一个站点的Gis号。

属性值设定，指定的属性后面添加IsConcurrencyToken

builder.Property(s => s.GisNo).IsConcurrencyToken();

通过DbUpdateConcurrencyException捕获属性值并发修改异常。其中var databaseValues = await entry.GetDatabaseValuesAsync()可用于获取当前实体的数据库值。

catch (DbUpdateConcurrencyException ex)
{
    foreach (var entry in ex.Entries)
    {
        if (entry.Entity is Station)
        {
           
            //获取当前实体的数据库值
            var databaseValues = await entry.GetDatabaseValuesAsync();
            string newGisNo = databaseValues.GetValue<string>(nameof(Station.GisNo));
            Console.WriteLine($"并发冲突，有新{newGisNo}更新");
        }
        else
        {
            throw new NotSupportedException(
                "Don't know how to handle concurrency conflicts for "
                + entry.Metadata.Name);
        }
    }
}

完整代码

public class GisChange
{
    public async Task ExecuteAsync()
    {
        Console.WriteLine("===============开始修改GIS===================");
        Console.WriteLine("请输入GIS:");
        string gisNo = Console.ReadLine();
        Console.WriteLine("请确认是否继续：");
        Console.WriteLine("Y/N");
        string input = Console.ReadLine();
        if (input != "Y")
        {
            await ChangeAsync(gisNo, null);
        }
        else
        {
            await ChangeAsync(gisNo, () =>
            {
                Thread.Sleep(5000);
            });
        }
    }

    private async Task ChangeAsync(string gisNo,Action sleep)
    {
        using (ApplicationDbContext dbContext = new ApplicationDbContext())
        {
            bool saved = false;
            while (!saved)
            {
                try
                {
                    Station? station = await dbContext.Stations.FindAsync(1);
                    station.GisNo = gisNo;
                    if(sleep  != null) {
                        sleep();
                    }
                    await dbContext.SaveChangesAsync();
                    saved = true;
                    Console.WriteLine("修改成功");
                }
                catch (DbUpdateConcurrencyException ex)
                {
                    foreach (var entry in ex.Entries)
                    {
                        if (entry.Entity is Station)
                        {
                           
                            //获取当前实体的数据库值
                            var databaseValues = await entry.GetDatabaseValuesAsync();
                            string newGisNo = databaseValues.GetValue<string>(nameof(Station.GisNo));
                            Console.WriteLine($"并发冲突，有新{newGisNo}更新");
                        }
                        else
                        {
                            throw new NotSupportedException(
                                "Don't know how to handle concurrency conflicts for "
                                + entry.Metadata.Name);
                        }
                    }
                }
            }
        }
    }
}

接下来我们启动两个控制台程序，几乎同时更新Gis编号。
监测到并发冲突
在这里插入图片描述

但是如果两次修改的是同一个值，在给属性颁发令牌的这种方式里，EF Core会忽略掉并发冲突
在这里插入图片描述

2.1.2 数据库生成的并发令牌

SQLServer数据库可以用一个byte[]类型的属性做并发令牌属性，然后使用IsRowVersion()把这个属性设置为RowVersion类型，这样这个属性对应的数据库列就会被设置为RowVersion类型。对于RowVersion类型的列，在每次插入或更新行时，数据库会自动为这一行的ROWVERSION类型的列其生成新值。
在SQLServer中，timeStamp和rowVersion是同一种类型的不同别名而已

并不是所有数据库都支持自动生成rowVersion，因此每当持久保留更改时，都必须在应用程序中分配此属性。也可也实现对于行数据的并发监测

设想银行转账这个场景，并且有a,b,c三个账户，Id分别为1，2，3。转账这个流程是一个典型的原子性，b向a转账100，b的账户扣除100，a的账户增加100。要么都成功，要么都失败。

添加byte[] 类型的RowVer列，并且在OnModelCreating方法里注明改属性是.IsRowVersion()

public byte[] RowVer { get; set; }

builder.Property(ba => ba.RowVer).IsRowVersion();

通过DbUpdateConcurrencyException捕获并发冲突

try
{
    await dbContext.SaveChangesAsync();
    return true;
}
catch (DbUpdateConcurrencyException ex)
{
    Console.WriteLine("并发冲突，正在重新加载数据...");
    // 处理并发冲突：重新加载当前数据，重新计算
    ex.Entries.Single().Reload();
    // 基于最新数据重新累加
    fromAccount.Balance -= amount;
    toAccount.Balance += amount;
    await dbContext.SaveChangesAsync();
    return true;
}

完整代码

public class Trade
{
    public async Task ExecuteAsync()
    {
        Console.Write("请输入您的账户ID：");
        long fromAccountId = Convert.ToInt64(Console.ReadLine());
        Console.Write("请输入您要转账的账户ID：");
        long toAccountId = Convert.ToInt64(Console.ReadLine());
        Console.Write("请输入转账金额：");
        decimal amount = Convert.ToDecimal(Console.ReadLine());
        Console.WriteLine("请确认是否继续：");
        Console.WriteLine("Y/N");
        string input = Console.ReadLine();
        if (input != "Y")
        {
            Console.WriteLine("取消转账");
            return;
        }
        Console.WriteLine($"正在转账 {amount} 元...${DateTime.Now}");
        bool isTransferring = true;
        // 启动后台计时器
        var timerTask = Task.Run(async () =>
        {
            while (isTransferring)
            {
                Console.WriteLine($"处理中: {DateTime.Now}");
                await Task.Delay(1000);
            }
        });
       
        try
        {
            bool success = await TransferAsync(fromAccountId, toAccountId, amount,
                () => Thread.Sleep(5000));

            if (success)
            {
                Console.WriteLine($"转账成功！{DateTime.Now}");
            }
        }
        finally
        {
            // 标记转账完成并等待计时器任务结束
            isTransferring = false;
            await timerTask; // 等待任务自然结束
        }
        Console.ReadLine();
    }

    public async Task<bool> TransferAsync(long fromAccountId, long toAccountId, decimal amount, Action sleep)
    {
        using (ApplicationDbContext dbContext = new ApplicationDbContext())
        {
            var fromAccount = await dbContext.BankAccounts
            .FirstOrDefaultAsync(a => a.AccountID == fromAccountId);

            var toAccount = await dbContext.BankAccounts
            .FirstOrDefaultAsync(a => a.AccountID == toAccountId);

            if (sleep != null)
            {
                sleep();
            }
            // 验证账户是否存在
            if (fromAccount == null || toAccount == null)
            {
                throw new InvalidOperationException("账户不存在");
            }
            // 检查余额是否足够
            if (fromAccount.Balance < amount)
            {
                throw new InvalidOperationException("余额不足");
            }
            // 执行转账
            fromAccount.Balance -= amount;
            toAccount.Balance += amount;

            try
            {
                await dbContext.SaveChangesAsync();
                return true;
            }
            catch (DbUpdateConcurrencyException ex)
            {
                Console.WriteLine("并发冲突，正在重新加载数据...");
                // 处理并发冲突：重新加载当前数据，重新计算
                ex.Entries.Single().Reload();
                // 基于最新数据重新累加
                fromAccount.Balance -= amount;
                toAccount.Balance += amount;
                await dbContext.SaveChangesAsync();
                return true;
            }
        }
            
    }
}

接下来我们启动两个控制台程序，第一个程序通过3号账号向1号账号转账，第二个程序通过2号账号向1号账号转账。两个程序同时执行，我们观察执行结果。
成功捕捉到并发冲突。
在这里插入图片描述
此时数据库的RowVer便是数据库维护的令牌列

在这里插入图片描述

2.2 悲观锁

在EF Core中实现悲观锁主要是借助数据库的锁机制来实现，并显示结合事务。

同样是2.1.2的银行转账场景

首先我们通过EF Core显示开启事务

using (IDbContextTransaction transaction = await dbContext.Database.BeginTransactionAsync(IsolationLevel.Serializable))

并且在转账逻辑完成的时候，提交事务

// 提交事务（释放锁）
await transaction.CommitAsync();

使用 SQL Server 原生语句查询获取账号并锁定账户，使用ROWLOCK, UPDLOCK锁定行数据的更新。

 // 使用 SQL Server 原生语句查询并锁定账户（悲观锁）
 var fromAccount = await dbContext.BankAccounts
     .FromSql($"SELECT * FROM T_BankAccount WITH (ROWLOCK, UPDLOCK) WHERE AccountID = {fromAccountId}")
     .SingleOrDefaultAsync();

 var toAccount = await dbContext.BankAccounts
     .FromSql($"SELECT * FROM T_BankAccount WITH (ROWLOCK, UPDLOCK) WHERE AccountID = {toAccountId}")
     .SingleOrDefaultAsync();

4.观察执行转账，启动一个后台进程循环打印处理中，方便观察。通过isTransferring标识符来结束循环。

bool isTransferring = true;
// 启动后台计时器
var timerTask = Task.Run(async () =>
{
    while (isTransferring)
    {
        Console.WriteLine($"处理中: {DateTime.Now}");
        await Task.Delay(1000);
    }
});

完整代码

public class Program
{
    static async Task Main(string[] args)
    {
        Console.Write("请输入您的账户ID：");
        long fromAccountId = Convert.ToInt64(Console.ReadLine());
        Console.Write("请输入您要转账的账户ID：");
        long toAccountId = Convert.ToInt64(Console.ReadLine());
        Console.Write("请输入转账金额：");
        decimal amount = Convert.ToDecimal(Console.ReadLine());
        Console.WriteLine("请确认是否继续：");
        Console.WriteLine("Y/N");
        string input = Console.ReadLine();
        if (input != "Y")
        {
            Console.WriteLine("取消转账");
            return;
        }
        Console.WriteLine($"正在转账 {amount} 元...${DateTime.Now}");
        bool isTransferring = true;
        // 启动后台计时器
        var timerTask = Task.Run(async () =>
        {
            while (isTransferring)
            {
                Console.WriteLine($"处理中: {DateTime.Now}");
                await Task.Delay(1000);
            }
        });
        try
        {
            bool success = await TransferMoneyAsync(fromAccountId, toAccountId, amount,
                () => Thread.Sleep(5000));

            if (success)
            {
                Console.WriteLine($"转账成功！{DateTime.Now}");
            }
        }
        finally
        {
            // 标记转账完成并等待计时器任务结束
            isTransferring = false;
            await timerTask; // 等待任务自然结束
        }
        Console.ReadLine();
    }

    /// <summary>
    /// 执行转账
    /// </summary>
    /// <param name="fromAccountId"></param>
    /// <param name="toAccountId"></param>
    /// <param name="amount"></param>
    /// <param name="sleep"></param>
    /// <returns></returns>
    /// <exception cref="InvalidOperationException"></exception>
    public static async Task<bool> TransferMoneyAsync(long fromAccountId, long toAccountId, decimal amount, Action sleep)
    {
        using (ApplicationDbContext dbContext = new ApplicationDbContext())
        using (IDbContextTransaction transaction = await dbContext.Database.BeginTransactionAsync(IsolationLevel.Serializable))
        {
            try
            {
                // 使用 SQL Server 原生语句查询并锁定账户（悲观锁）
                var fromAccount = await dbContext.BankAccounts
                    .FromSql($"SELECT * FROM T_BankAccount WITH (ROWLOCK, UPDLOCK) WHERE AccountID = {fromAccountId}")
                    .SingleOrDefaultAsync();

                var toAccount = await dbContext.BankAccounts
                    .FromSql($"SELECT * FROM T_BankAccount WITH (ROWLOCK, UPDLOCK) WHERE AccountID = {toAccountId}")
                    .SingleOrDefaultAsync();
                if (sleep != null)
                {
                    sleep();
                }
                // 验证账户是否存在
                if (fromAccount == null || toAccount == null)
                {
                    throw new InvalidOperationException("账户不存在");
                }
                // 检查余额是否足够
                if (fromAccount.Balance < amount)
                {
                    throw new InvalidOperationException("余额不足");
                }
                // 执行转账
                fromAccount.Balance -= amount;
                toAccount.Balance += amount;
                // 更新数据库
                await dbContext.SaveChangesAsync();
                // 提交事务（释放锁）
                await transaction.CommitAsync();
                return true;
            }
            catch (Exception ex)
            {
                // 发生异常时回滚事务
                await transaction.RollbackAsync();
                Console.WriteLine($"转账失败: {ex.Message}");
                return false;
            }
        }
    }
}