- VisualStudio2022插件的安装及使用-编程手把手系列文章
- pprof-在现网场景怎么用
- C#实现的下拉多选框,下拉多选树,多级节点
- 【学习笔记】基础数据结构:猫树
目前一个项目中数据持久化采用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来实现更好,加入锁的机制只是一种解决问题的思路.
MigrationLocks
表来实现迁移锁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;
SELECT COUNT(*) FROM MigrationLocks WHERE LockName = @database AND LockedAt > NOW() - INTERVAL 5 MINUTE;
/// <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);
}
}
/// <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.");
}
}
最后此篇关于EntityFrameworkCore并发迁移解决方案的文章就讲到这里了,如果你想了解更多关于EntityFrameworkCore并发迁移解决方案的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
这是一个使用dotnet-ef迁移工具生成MySQL数据库的简单案例。 我有 Pomelo.EntityFrameworkCore.MySql 作为 MySQL DB 连接器和类属性: public
在使用EFCore加密SQLite数据库时,我遇到了这个GitHub issue 请参阅8月1日bricelam的评论。该问题说明了如何使用efcore加密SQLite数据库。 它指的是Microso
我有一个使用EF Core的ASP.NET Core 3.1 Web API应用程序。这是ConfigureServices类的Startup方法中的我的配置: services.AddDbConte
我很长一段时间都遇到麻烦。让我们想象一下这个例子: public class Coordinate { public int id {get;set;} public int x
我正在尝试将 mysql 与 Microsoft.EntityFrameworkCore 2.1-rc-final 和 MySql.Data.EntityFrameworkCore 8.0.11 一起
一个带有自动递增 Key 列的简单类 public class SomeClass { [Key] public long SomeClassId { get; set;} } 通常
我最近决定在我使用 ASP.NET Core MVC 开发的网站上实现 ASP.NET Identity 功能。 让我们快速浏览一下主题中的表和类: public class User : Ident
我有以下两个实体: public class Company { public Guid Id { get; set; } public string Name { get; set;
在学习 EntityFrameworkCore 上的其中一门 PluralSight 类(class)时,我从视频中看到了不同的结果。即,当调用 DbSet.AddRange(item 1, item
对于集成测试,我使用 EntityFrameworkCore SQLite 内存数据库并根据 Microsoft 文档创建其架构,但是当我尝试播种数据时,会引发异常该表不存在。 DbContext.D
我正在尝试 EntityFrameworkCore .我查看了文档,但找不到一种方法来轻松更新与另一个实体相关的复杂实体。 这是一个简单的例子。我有 2 个类 - 公司和员工。 public clas
加载相关数据(类(class)的类(class)页面)时出现以下错误。 错误 MySqlException: Unknown column 'c.Coursepages.CoursepageId' i
有没有办法使用 EntityFrameworkCore 将一个点转换为另一个 SRID? 之前我使用了 ST_Transform(ST_GeomFromText(@coord,4326),32661)
我正在尝试使用以下代码将我的 Web API 连接到 MySql 数据库: public class Startup { public void ConfigureServices
我正在使用 XUNIT 在点网核心应用程序中进行测试。 我需要测试一项服务,该服务在我的数据上下文中对 DbSet 进行内部异步查询。 I've seen here异步模拟 DbSet 是可能的。 我
我知道这是以前问过的那种问题,但我找不到有效的答案。 那么,是否有可能改变迁移的存储位置?我正在使用 EF Core。 仅供引用:我知道如果您移动第一个生成的迁移,接下来的迁移将放置在同一位置。 最佳
我正在尝试使用此命令安装 Microsoft.EntityFrameworkCore.Tools Install-Package Microsoft.EntityFrameworkCore.Tools
我尝试为 Visual Studio 2017 执行 Razor ContosoUniversity(最初在 ContsoObserve 中重命名)并出现错误我的 .csproj 是
我尝试为 Visual Studio 2017 执行 Razor ContosoUniversity(最初在 ContsoObserve 中重命名)并出现错误我的 .csproj 是
你好,我遇到了 Entity Framework 的问题,告诉我这些对象在我的数据库中不存在,而我已经使用 .ToTable 方法映射了名称。下面是我如何将上下文注入(inject)我的 Contro
我是一名优秀的程序员,十分优秀!