EntityFramework Core并发迁移解决方案

场景

目前一个项目中数据持久化采用EF Core + MySQL，使用CodeFirst模式开发，并且对数据进行了分库，按照目前颗粒度分完之后，大概有一两百个库，每个库的数据都是相互隔离的。
借鉴了Github上一个开源的仓库 arch/UnitOfWork 实现UnitOfWork，核心操作就是每个api请求的时候带上库名，在执行CRUD之前先将DbContext切换到目标数据库，我们在切换数据库的时候加了一些操作，如检查数据库是否已创建、检查连接是否可用、判断是否需要 表结构迁移 等

/// <summary> /// 切换数据库 这要求数据库在同一台机器上 注意:这只适用于MySQL。 /// </summary> /// <param name="database">目标数据库</param> public void ChangeDatabase(string database) {     // 检查连接     ......      // 检查数据库是否创建     ......      var connection = _context.Database.GetDbConnection();     if (connection.State.HasFlag(ConnectionState.Open))     {         connection.ChangeDatabase(database);     }     else     {         var connectionString = Regex.Replace(connection.ConnectionString.Replace(" ", ""), @"(?<=[Dd]atabase=)w+(?=;)", database, RegexOptions.Singleline);         connection.ConnectionString = connectionString;     }      // 判断是否需要执行表结构迁移     if(_context..Database.GetPendingMigrations().Any())     {         //自定义的迁移的一些逻辑         _context.Database.Migrate(_context);     } }        

但是当多个操作同时对一个库进行Migrate的时候，就会出现问题，比如“新增一张表”的操作已经被第一个迁移执行过了，第二个执行的迁移并不知道已经执行过了Migrate，就会报错表已存在。
于是考虑在执行Migrate的时候，加入一个锁的机制，对当前数据库执行Migrate之前先获取锁，然后再来决定接下来的操作。由于这边有的服务无法访问Redis，这里使用数据库来实现锁的机制，当然用Redis来实现更好，加入锁的机制只是一种解决问题的思路。

利用数据库实现迁移锁

1. 新增 MigrationLocks 表来实现迁移锁

• 锁的操作不依赖DbContext实例
• 在执行Migrate之前，尝试获取一个锁，在获取锁之前，如果表不存在则创建

  CREATE TABLE IF NOT EXISTS MigrationLocks (     LockName VARCHAR(255) PRIMARY KEY,     LockedAt DATETIME NOT NULL ); 

• 成功往表中插入一条记录，视为获取锁成功，主键为需要迁移的库的名称

  INSERT INTO MigrationLocks (LockName, LockedAt) VALUES (@database, NOW()); 

• 迁移完成后，删除这条记录，视为释放锁成功；

  DELETE FROM MigrationLocks WHERE LockName = @database; 

• 为防止 “死锁” 发生，每次尝试获取锁之前，会对锁的状态进行检查，释放超过5分钟的锁（正常来说，上一个迁移的执行时间不会超过5分钟）。

  SELECT COUNT(*) FROM MigrationLocks WHERE LockName = @database AND LockedAt > NOW() - INTERVAL 5 MINUTE; 

2. 封装一下MigrateLock的实现

/// <summary> /// 迁移锁 /// </summary> public interface IMigrateLock {     /// <summary>     /// 尝试获取锁     /// </summary>     /// <param name="connection"></param>     /// <returns></returns>     bool TryAcquireLock(IDbConnection connection);      /// <summary>     /// 尝试获取锁     /// </summary>     /// <param name="connection"></param>     /// <returns></returns>     Task<bool> TryAcquireLockAsync(IDbConnection connection);      /// <summary>     /// 释放锁     /// </summary>     void ReleaseLock(IDbConnection connection);      /// <summary>     /// 释放锁     /// </summary>     /// <returns></returns>     Task ReleaseLockAsync(IDbConnection connection); }  /// <summary> /// 迁移锁 /// </summary> public class MigrateLock : IMigrateLock {     private readonly ILogger<MigrateLock> _logger;      public MigrateLock(ILogger<MigrateLock> logger)     {         _logger = logger;     }      private const string CreateTableSql = @"         CREATE TABLE IF NOT EXISTS MigrationLocks (             LockName VARCHAR(255) PRIMARY KEY,             LockedAt DATETIME NOT NULL         );";      private const string CheckLockedSql = "SELECT COUNT(*) FROM MigrationLocks WHERE LockName = @database AND LockedAt > NOW() - INTERVAL 5 MINUTE;";      private const string AcquireLockSql = "INSERT INTO MigrationLocks (LockName, LockedAt) VALUES (@database, NOW());";      private const string ReleaseLockSql = "DELETE FROM MigrationLocks WHERE LockName = @database;";      /// <summary>     /// 尝试获取锁     /// </summary>     /// <param name="connection"></param>     /// <returns></returns>     public bool TryAcquireLock(IDbConnection connection)     {         try           {             CheckLocked(connection);              var result = connection.Execute(AcquireLockSql, new { database = connection.Database });             if (result == 1)             {                 _logger.LogInformation("Lock acquired: {LockName}", connection.Database);                  return true;             }              _logger.LogWarning("Failed to acquire lock: {LockName}", connection.Database);              return false;         }         catch (Exception ex)         {             if (ex.Message.StartsWith("Duplicate"))             {                 _logger.LogWarning("Failed acquiring lock due to duplicate entry: {LockName}", connection.Database);             }             else             {                 _logger.LogError(ex, "Error acquiring lock: {LockName}", connection.Database);             }              return false;         }     }      /// <summary>     /// 尝试获取锁     /// </summary>     /// <param name="connection"></param>     /// <returns></returns>     public async Task<bool> TryAcquireLockAsync(IDbConnection connection)     {         try         {             await CheckLockedAsync(connection);              var result = await connection.ExecuteAsync(AcquireLockSql, new { database = connection.Database });             if (result == 1)             {                 _logger.LogInformation("Lock acquired: {LockName}", connection.Database);                  return true;             }              _logger.LogWarning("Failed to acquire lock: {LockName}", connection.Database);              return false;         }         catch (Exception ex)         {             if (ex.Message.StartsWith("Duplicate"))             {                 _logger.LogWarning("Failed acquiring lock due to duplicate entry: {LockName}", connection.Database);             }             else             {                 _logger.LogError(ex, "Error acquiring lock: {LockName}", connection.Database);             }              return false;         }     }      /// <summary>     /// 释放锁     /// </summary>     public void ReleaseLock(IDbConnection connection)     {         try         {             connection.ExecuteAsync(ReleaseLockSql, new { database = connection.Database });             _logger.LogInformation("Lock released: {LockName}", connection.Database);         }         catch (Exception ex)         {             _logger.LogError(ex, "Error releasing lock: {LockName}", connection.Database);         }     }      /// <summary>     /// 释放锁     /// </summary>     public async Task ReleaseLockAsync(IDbConnection connection)     {         try         {             await connection.ExecuteAsync(ReleaseLockSql, new { database = connection.Database });             _logger.LogInformation("Lock released: {LockName}", connection.Database);         }         catch (Exception ex)         {             _logger.LogError(ex, "Error releasing lock: {LockName}", connection.Database);         }     }      /// <summary>     /// 检查锁     /// </summary>     private void CheckLocked(IDbConnection connection)     {         connection.Execute(CreateTableSql);           var databaseParam = new         {             database = connection.Database         };          var lockExists = connection.QueryFirstOrDefault<int>(CheckLockedSql, databaseParam);         if (lockExists <= 0)         {             return;         }          _logger.LogWarning("Lock exists and is older than 5 minutes. Releasing old lock.");         connection.Execute(ReleaseLockSql, databaseParam);     }      /// <summary>     /// 检查锁     /// </summary>     private async Task CheckLockedAsync(IDbConnection connection)     {         await connection.ExecuteAsync(CreateTableSql);                   var databaseParam = new         {             database = connection.Database         };          var lockExists = await connection.QueryFirstOrDefaultAsync<int>(CheckLockedSql, databaseParam);         if (lockExists <= 0)         {             return;         }          _logger.LogWarning("Lock exists and is older than 5 minutes. Releasing old lock.");         await connection.ExecuteAsync(ReleaseLockSql, databaseParam);     } } 

3. 封装一下MigrateExecutor的实现

/// <summary> /// 数据库迁移执行器 /// </summary> public interface IMigrateExcutor {     /// <summary>     /// 执行迁移     /// </summary>     /// <param name="dbContext"></param>     void Migrate(DbContext dbContext);      /// <summary>     /// 执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <returns></returns>     Task MigrateAsync(DbContext dbContext);      /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     void ConcurrentMigrate(DbContext dbContext, bool wait = true);           /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     /// <returns></returns>     Task ConcurrentMigrateAsync(DbContext dbContext, bool wait = true);      /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="connection"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     void ConcurrentMigrate(DbContext dbContext, IDbConnection connection, bool wait = true);      /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="connection"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     Task ConcurrentMigrateAsync(DbContext dbContext, IDbConnection connection, bool wait = true); }  /// <summary> /// 数据库迁移执行器 /// </summary> public class MigrateExcutor : IMigrateExcutor {     private readonly IMigrateLock _migrateLock;     private readonly ILogger<MigrateExcutor> _logger;      public MigrateExcutor(         IMigrateLock migrateLock,         ILogger<MigrateExcutor> logger)     {         _migrateLock = migrateLock;         _logger = logger;     }      /// <summary>     /// 执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <returns></returns>     public void Migrate(DbContext dbContext)     {         try         {             if (dbContext.Database.GetPendingMigrations().Any())             {                 dbContext.Database.Migrate();             }         }         catch (Exception e)         {             _logger.LogError(e, "Migration failed");              HandleError(dbContext, e);         }     }      /// <summary>     /// 执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <returns></returns>     public async Task MigrateAsync(DbContext dbContext)     {         try         {             if ((await dbContext.Database.GetPendingMigrationsAsync()).Any())             {                 await dbContext.Database.MigrateAsync();             }         }         catch (Exception e)         {             _logger.LogError(e, "Migration failed");              await HandleErrorAsync(dbContext, e);         }     }      /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     /// <returns></returns>     public void ConcurrentMigrate(DbContext dbContext, bool wait = true)     {         if (!dbContext.Database.GetPendingMigrations().Any())         {             return;         }          using var connection = MySqlConnectionHelper.CreateConnection(dbContext.Database.GetDbConnection().Database);          ConcurrentMigrate(dbContext, connection, wait);     }      /// <summary>     /// 并发场景执行迁移     /// </summary>     /// <param name="dbContext"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     /// <returns></returns>     public async Task ConcurrentMigrateAsync(DbContext dbContext, bool wait = true)     {         if ((await dbContext.Database.GetPendingMigrationsAsync()).Any())         {             return;         }          await using var connection = await MySqlConnectionHelper.CreateConnectionAsync(dbContext.Database.GetDbConnection().Database);          await ConcurrentMigrateAsync(dbContext, connection, wait);     }      /// <summary>     /// 并发场景执行迁移（供数据同步相关服务使用，”迁移锁“ 使用传入的 <see cref="IDbConnection"/> 对象来完成）     /// </summary>     /// <param name="dbContext"></param>     /// <param name="connection"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     public void ConcurrentMigrate(DbContext dbContext, IDbConnection connection, bool wait = true)     {         if (!dbContext.Database.GetPendingMigrations().Any())         {             return;         }          while (true)         {             if (_migrateLock.TryAcquireLock(connection))             {                 try                 {                     Migrate(dbContext);                      break;                 }                 finally                 {                     _migrateLock.ReleaseLock(connection);                 }             }              if (wait)             {                 _logger.LogWarning("Migration is locked, wait for 2 seconds");                 Thread.Sleep(20000);                  continue;             }              _logger.LogInformation("Migration is locked, skip");         }     }      /// <summary>     /// 并发场景执行迁移（供数据同步相关服务使用，”迁移锁“ 使用传入的 <see cref="IDbConnection"/> 对象来完成）     /// </summary>     /// <param name="dbContext"></param>     /// <param name="connection"></param>     /// <param name="wait">是否等待至正在进行中的迁移完成</param>     public async Task ConcurrentMigrateAsync(DbContext dbContext, IDbConnection connection, bool wait = true)     {         if ((await dbContext.Database.GetPendingMigrationsAsync()).Any())         {             return;         }          while (true)         {             if (await _migrateLock.TryAcquireLockAsync(connection))             {                 try                 {                     await MigrateAsync(dbContext);                     break;                 }                 finally                 {                     await _migrateLock.ReleaseLockAsync(connection);                 }             }              if (wait)             {                 _logger.LogWarning("Migration is locked, wait for 2 seconds");                 Thread.Sleep(20000);                  continue;             }              _logger.LogInformation("Migration is locked, skip");              break;         }     }      private void HandleError(DbContext dbContext, Exception e)     {         var needChangeList = dbContext.Database.GetPendingMigrations().ToList();         var allChangeList = dbContext.Database.GetMigrations().ToList();         var hasChangeList = dbContext.Database.GetAppliedMigrations().ToList();          if (needChangeList.Count + hasChangeList.Count > allChangeList.Count)         {             int errIndex = allChangeList.Count - needChangeList.Count;              if (hasChangeList.Count - 1 == errIndex && hasChangeList[errIndex] != needChangeList[0])             {                 int index = needChangeList[0].IndexOf("_", StringComparison.Ordinal);                 string errSuffix = needChangeList[0].Substring(index, needChangeList[0].Length - index);                 if (hasChangeList[errIndex].EndsWith(errSuffix))                 {                     dbContext.Database.ExecuteSqlRaw($"Update __EFMigrationsHistory set MigrationId = '{needChangeList[0]}' where MigrationId = '{hasChangeList[errIndex]}'");                     dbContext.Database.Migrate();                 }                 else                 {                     throw e;                 }             }             else             {                 throw e;             }         }         else         {             throw e;         }          _logger.LogInformation("Migration failed, but success on second try.");     }      private async Task HandleErrorAsync(DbContext dbContext, Exception e)     {         var needChangeList = (await dbContext.Database.GetPendingMigrationsAsync()).ToList();         var allChangeList = dbContext.Database.GetMigrations().ToList();         var hasChangeList = (await dbContext.Database.GetAppliedMigrationsAsync()).ToList();          if (needChangeList.Count + hasChangeList.Count > allChangeList.Count)         {             int errIndex = allChangeList.Count - needChangeList.Count;              if (hasChangeList.Count - 1 == errIndex && hasChangeList[errIndex] != needChangeList[0])             {                 int index = needChangeList[0].IndexOf("_", StringComparison.Ordinal);                 string errSuffix = needChangeList[0].Substring(index, needChangeList[0].Length - index);                 if (hasChangeList[errIndex].EndsWith(errSuffix))                 {                     await dbContext.Database.ExecuteSqlRawAsync($"Update __EFMigrationsHistory set MigrationId = '{needChangeList[0]}' where MigrationId = '{hasChangeList[errIndex]}'");                     await dbContext.Database.MigrateAsync();                 }                 else                 {                     throw e;                 }             }             else             {                 throw e;             }         }         else         {             throw e;         }          _logger.LogInformation("Migration failed, but success on second try.");     } }