EF Core3.0+ 通過攔截器實現讀寫分離與SQL日志記錄

前言

本文主要是講解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語句就方便了很多~