前言
本文主要是讲解EF Core3.0+ 通过拦截器实现读写分离与SQL日志记录
注意拦截器只有EF Core3.0+ 支持,2.1请考虑上下文工厂的形式实现.
说点题外话..
一晃又大半年没更新技术博客..唉,去年一年发生了太多事情..博主真的 一言难尽..
有兴趣的可以去看看:记录一下,也许是转折,也许是结束,也许是新希望的一年
正文
1.通过拦截器实现读写分离
先讲一下本文实现的方式吧
SQL 通过数据库本身的功能 实现主从备份 大概原理如图:
EF Core在查询的时候通过DbCommandInterceptor 拦截器(PS:这个功能在EF6.0+中也实现了)来拦截对数据库的访问,从而切换主从数据库
下面直接上代码吧
首先我们创建一个类 继承DbCommandInterceptor:
public class DbMasterSlaveCommandInterceptor : DbCommandInterceptor
{
private string _masterConnectionString;
private string _slaveConnectionString; public DbMasterSlaveCommandInterceptor(string masterConnectionString, string slaveConnectionString)
{
_masterConnectionString = masterConnectionString;
_slaveConnectionString = slaveConnectionString;
}
}
通过构造函数传递主库连接地址与从库地址(可有多个 通过"|"分割)
添加一个随机分配从表读取连接的方法(PS:这里只是demo所以很简陋的随机,如果正式要用,应包含权重判断,定时心跳从库连接情况,请自行修改):
/// <summary>
/// 通过随机数分配获取多个从库
/// </summary>
/// <returns></returns>
private string GetSlaveConnectionString()
{
var readArr = _slaveConnectionString.Split(new char[] { '|' }, StringSplitOptions.RemoveEmptyEntries);
var resultConn = string.Empty;
if (readArr != null && readArr.Any())
{
resultConn = readArr[Convert.ToInt32(Math.Floor((double)new Random().Next(0, readArr.Length)))];
}
return resultConn;
}
添加判断是否主从操作连接方法:
private void UpdateToSlave(DbCommand command)
{
//判断是否配置了主从分离
if (!string.IsNullOrWhiteSpace(GetSlaveConnectionString()))//如果配置了读写分离，就进入判断
{
//判断是否为插入语句(EF 插入语句会通过Reader执行并查询主键),否则进入
if (command.CommandText.ToLower().StartsWith("insert", StringComparison.InvariantCultureIgnoreCase) == false)
{
// 判断当前会话是否处于分布式事务中
bool isDistributedTran = Transaction.Current != null &&
Transaction.Current.TransactionInformation.Status !=
TransactionStatus.Committed;
//判断该 context 是否处于普通数据库事务中
bool isDbTran = command.Transaction != null;
//如果不处于事务中,则执行从服务器查询
if (!isDbTran && !isDistributedTran)
{
command.Connection.Close();
command.Connection.ConnectionString = GetSlaveConnectionString();
command.Connection.Open(); } }
}
}
重载DbCommandInterceptor当中的拦截方法,代码如下:
//如果是写入,则正常执行
public override InterceptionResult<int> NonQueryExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<int> result)
{
return base.NonQueryExecuting(command, eventData, result);
}
public override Task<InterceptionResult<int>> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<int> result, CancellationToken cancellationToken = default)
{
return base.NonQueryExecutingAsync(command, eventData, result, cancellationToken);
} public override InterceptionResult<DbDataReader> ReaderExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
{
this.UpdateToSlave(command);
return base.ReaderExecuting(command, eventData, result);
} public override Task<InterceptionResult<DbDataReader>> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result, CancellationToken cancellationToken = default)
{
this.UpdateToSlave(command);
return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
} public override InterceptionResult<object> ScalarExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<object> result)
{
this.UpdateToSlave(command);
return base.ScalarExecuting(command, eventData, result);
} public override Task<InterceptionResult<object>> ScalarExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<object> result, CancellationToken cancellationToken = default)
{
this.UpdateToSlave(command);
return base.ScalarExecutingAsync(command, eventData, result, cancellationToken);
}
最后在EF core的上下文中注入拦截器(PS:我这里使用的Autofac模块注入):
builder.Register(
c =>
{
var optionsBuilder = new DbContextOptionsBuilder<TestEFContext>();
//注入拦截器
optionsBuilder.AddInterceptors(new DbMasterSlaveCommandInterceptor(WriteConnect, ReadConnect));
//MaxBatchSize 处理批量操作BUG
optionsBuilder.UseMysql(WriteConnect, b=>b.MaxBatchSize(1));
return optionsBuilder.Options;
}
).As<DbContextOptions<TestEFContex>>().SingleInstance();
这样就实现了通过拦截器实现读写分离.
2.通过拦截器实现SQL日志记录
同理,我们可以通过拦截器实现EF Core SQL语句的记录与调试
首先我们创建一个新的拦截器DBlogCommandInterceptor 如下:
public class DBlogCommandInterceptor : DbCommandInterceptor
{
//创建一个队列记录SQL执行时间
static readonly ConcurrentDictionary<DbCommand, DateTime> MStartTime = new ConcurrentDictionary<DbCommand, DateTime>();
private ILogger<DBlogCommandInterceptor> _logger { get; set; }
//通过构造函数注入日志
public DBlogCommandInterceptor(ILogger<DBlogCommandInterceptor> Logger)
{
_logger = Logger;
}
}
创建2个私有的方法,一个记录执行开始时间,一个记录SQL
//记录SQL开始执行的时间
private void OnStart(DbCommand command)
{
MStartTime.TryAdd(command, DateTime.Now);
}
//通过_logger输出日志
private void Log(DbCommand command)
{ DateTime startTime;
TimeSpan duration;
//得到此command的开始时间
MStartTime.TryRemove(command, out startTime);
if (startTime != default(DateTime))
{
duration = DateTime.Now - startTime;
}
else
{
duration = TimeSpan.Zero;
}
var parameters = new StringBuilder();
//循环获取执行语句的参数值
foreach (DbParameter param in command.Parameters)
{
parameters.AppendLine(param.ParameterName + " " + param.DbType + " = " + param.Value);
}
_logger.LogInformation("{starttime}开始执行SQL语句:{sql},参数:{canshu},执行时间{readtime}",
startTime.ToString(), command.CommandText, parameters.ToString(), duration.TotalSeconds); }
最后重载拦截器的方法:
public override InterceptionResult<int> NonQueryExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<int> result)
{
OnStart(command);
return base.NonQueryExecuting(command, eventData, result);
}
public override Task<InterceptionResult<int>> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<int> result, CancellationToken cancellationToken = default)
{
OnStart(command);
return base.NonQueryExecutingAsync(command, eventData, result, cancellationToken);
}
public override int NonQueryExecuted(DbCommand command, CommandExecutedEventData eventData, int result)
{
Log(command);
return base.NonQueryExecuted(command, eventData, result);
}
public override Task<int> NonQueryExecutedAsync(DbCommand command, CommandExecutedEventData eventData, int result, CancellationToken cancellationToken = default)
{
Log(command);
return base.NonQueryExecutedAsync(command, eventData, result, cancellationToken);
} public override InterceptionResult<object> ScalarExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<object> result)
{
OnStart(command);
return base.ScalarExecuting(command, eventData, result);
}
public override Task<InterceptionResult<object>> ScalarExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<object> result, CancellationToken cancellationToken = default)
{
OnStart(command);
return base.ScalarExecutingAsync(command, eventData, result, cancellationToken);
}
public override object ScalarExecuted(DbCommand command, CommandExecutedEventData eventData, object result)
{
Log(command);
return base.ScalarExecuted(command, eventData, result);
}
public override Task<object> ScalarExecutedAsync(DbCommand command, CommandExecutedEventData eventData, object result, CancellationToken cancellationToken = default)
{
Log(command);
return base.ScalarExecutedAsync(command, eventData, result, cancellationToken);
} public override InterceptionResult<DbDataReader> ReaderExecuting(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
{
OnStart(command);
return base.ReaderExecuting(command, eventData, result);
}
public override Task<InterceptionResult<DbDataReader>> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result, CancellationToken cancellationToken = default)
{
OnStart(command);
return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
} public override Task<DbDataReader> ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default)
{
Log(command);
return base.ReaderExecutedAsync(command, eventData, result, cancellationToken);
}
public override DbDataReader ReaderExecuted(DbCommand command, CommandExecutedEventData eventData, DbDataReader result)
{
Log(command);
return base.ReaderExecuted(command, eventData, result);
}
这样,我们就实现了通过拦截器实现SQL日志记录~效果如下:
调试SQL语句就方便了很多~