.netcore实现一个读写分离的数据库访问中间件
在实际业务系统中，当单个数据库不能承载负载压力的时候，一般我们采用数据库读写分离的方式来分担数据库负载。主库承担写以及事务操作，从库承担读操作。
为了支持多种数据库我们先定义一个数据类型字典。key为连接字符串，value为数据库类型：
/// <summary> /// 数据库方言集合 /// </summary> private readonly Dictionary<string, DatabaseDialectEnum> DialectDictionary = new Dictionary<string, DatabaseDialectEnum> { ["sqlconnection"] = DatabaseDialectEnum.MSSQL, ["sqlceconnection"] = DatabaseDialectEnum.SQLCE, ["npgsqlconnection"] = DatabaseDialectEnum.POSTGRES, ["sqliteconnection"] = DatabaseDialectEnum.SQLLITE, ["mysqlconnection"] = DatabaseDialectEnum.MYSQL, ["fbconnection"] = DatabaseDialectEnum.FIREBASE };
这样我们切换不同的数据库只需要配置数据库连接字符串即可。
以mssql为例，配置数据库连接字符串
"ConnectionString": { "sqlconnection": "Data Source=.;Initial Catalog=Db;User ID=sa;Password=**;Enlist=false;Max Pool SIZE=500;Min Pool SIZE=50;MultipleActiveResultSets=True", "sqlconnection_slaver_1": "Data Source=.;Initial Catalog=Db;User ID=sa;Password=**;Enlist=false;Max Pool SIZE=500;Min Pool SIZE=50;MultipleActiveResultSets=True", "sqlconnection_slaver_2": "Data Source=.;Initial Catalog=Db;User ID=sa;Password=**;Enlist=false;Max Pool SIZE=500;Min Pool SIZE=50;MultipleActiveResultSets=True" }
Key: sqlconnection为主库（master）连接字符串，Key: sqlconnection_slaver_1和sqlconnection_slaver_2为两个从库（slaver）连接字符串。多个从库(slaver)可以实现随机访问。也可以采用其他算法来负载均衡。 根据字符串连接配置我们得到 主库 连接串，和从库连接串集合。同时根据连接串的key 确定数据库种类。代码如下：
/// <summary> /// 主数据库连接串 /// </summary> private string MasterConnectionString { get; set; } /// <summary> /// 从数据库连接串集合 /// </summary> private List<string> SlaverConnectionStrings { get; set; } = new List<string>(); public ConnectionFactory(IConfiguration configuration, ILoggerFactory loggerFactory) { _logger = loggerFactory.CreateLogger<ConnectionFactory>(); var connectionKeys = configuration.GetSection("ConnectionString").GetChildren().Select(s => s.Key).ToArray(); foreach (var connKey in connectionKeys) { var connSplit = connKey.Split('_'); if (connSplit.Length == 1) { MasterConnectionString = configuration[$"ConnectionString:{connKey}"]; //根据连接字符串约定，确定数据库类型 DatabaseDialect = DialectDictionary[connKey]; } else { SlaverConnectionStrings.Add(configuration[$"ConnectionString:{connKey}"]); } } }
/// <summary> /// 数据库类型 /// </summary> public DatabaseDialectEnum DatabaseDialect { get; private set; }
获取主库连接
private IDbConnection GetMasterConnection() { return GetConnection(MasterConnectionString); }
获取从库连接，这里采用随机算法，如果没有配置从库，这里会返回主库连接。
private IDbConnection GetSlaverConnection() { int sc = SlaverConnectionStrings.Count(); if (sc > 0) { Random random = new Random(); int index = random.Next(0, sc); return GetConnection(SlaverConnectionStrings[index]); } else { _logger.LogInformation("没有设置从库，将建立主库连接"); return GetMasterConnection(); } }
private IDbConnection GetConnection(string connectionString) => DatabaseDialect switch { DatabaseDialectEnum.MSSQL =>new ProfiledDbConnection(new SqlConnection(connectionString),MiniProfiler.Current), DatabaseDialectEnum.MYSQL => new ProfiledDbConnection(new MySqlConnection(connectionString), MiniProfiler.Current), _ => throw new NotImplementedException() };
注：这里采用MiniProfiler来监控数据库连接性能，所以 返回的connection用ProfiledDbConnection进行了包装。
主从数据源类型如下：
public enum DataSourceEnum { MASTER, SLAVE }
本ConnectionFactory为单例模式，存在多线程访问的情况，所以数据源设置为ThreadLocal<DataSourceEnum>,线程内共享。
private static ThreadLocal<DataSourceEnum> threadLocal = new ThreadLocal<DataSourceEnum>(); /// <summary> /// 当前线程数据源 /// </summary> /// <param name="sourceEnum"></param> public DataSourceEnum DataSource { set { threadLocal.Value = value; } get { return threadLocal.Value; } }
下面正式获取IDbConnection
public IDbConnection GetDbConnection() { if (DataSource == DataSourceEnum.MASTER) { return GetMasterConnection(); } else { return GetSlaverConnection(); } }
使用：
根据文章开头所描述的实际操作来进行主从库访问。
private IDbConnection GetDbConnection(DataSourceEnum dataSource) { ConnectionFactory.DataSource = dataSource; return ConnectionFactory.GetDbConnection(); }
using var connection = GetDbConnection(DataSourceEnum.MASTER); connection.Execute(sql, param, CurrentTransaction, null, commandType)
using var connection = GetDbConnection(DataSourceEnum.SLAVE); connection.Get<T>(id, CurrentTransaction, CommandTimeout)
奉上全部代码
public class ConnectionFactory : IConnectionFactory { private readonly ILogger _logger; private static ThreadLocal<DataSourceEnum> threadLocal = new ThreadLocal<DataSourceEnum>(); static ConnectionFactory() { //设置dapper的tableName取值 SqlMapperExtensions.TableNameMapper = (type) => type.Name; } /// <summary> /// 当前线程数据源 /// </summary> /// <param name="sourceEnum"></param> public DataSourceEnum DataSource { set { threadLocal.Value = value; } get { return threadLocal.Value; } } /// <summary> /// 主数据库连接串 /// </summary> private string MasterConnectionString { get; set; } /// <summary> /// 从数据库连接串集合 /// </summary> private List<string> SlaverConnectionStrings { get; set; } = new List<string>(); public ConnectionFactory(IConfiguration configuration, ILoggerFactory loggerFactory) { _logger = loggerFactory.CreateLogger<ConnectionFactory>(); var connectionKeys = configuration.GetSection("ConnectionString").GetChildren().Select(s => s.Key).ToArray(); foreach (var connKey in connectionKeys) { var connSplit = connKey.Split('_'); if (connSplit.Length == 1) { MasterConnectionString = configuration[$"ConnectionString:{connKey}"]; DatabaseDialect = DialectDictionary[connKey]; } else { SlaverConnectionStrings.Add(configuration[$"ConnectionString:{connKey}"]); } } } /// <summary> /// 数据库方言集合 /// </summary> private readonly Dictionary<string, DatabaseDialectEnum> DialectDictionary = new Dictionary<string, DatabaseDialectEnum> { ["sqlconnection"] = DatabaseDialectEnum.MSSQL, ["sqlceconnection"] = DatabaseDialectEnum.SQLCE, ["npgsqlconnection"] = DatabaseDialectEnum.POSTGRES, ["sqliteconnection"] = DatabaseDialectEnum.SQLLITE, ["mysqlconnection"] = DatabaseDialectEnum.MYSQL, ["fbconnection"] = DatabaseDialectEnum.FIREBASE }; /// <summary> /// 数据库方言 /// </summary> public DatabaseDialectEnum DatabaseDialect { get; private set; } private IDbConnection GetConnection(string connectionString) => DatabaseDialect switch { DatabaseDialectEnum.MSSQL =>new ProfiledDbConnection(new SqlConnection(connectionString),MiniProfiler.Current), DatabaseDialectEnum.MYSQL => new ProfiledDbConnection(new MySqlConnection(connectionString), MiniProfiler.Current), _ => throw new NotImplementedException() }; public IDbConnection GetDbConnection() { if (DataSource == DataSourceEnum.MASTER) { return GetMasterConnection(); } else { return GetSlaverConnection(); } } private IDbConnection GetMasterConnection() { return GetConnection(MasterConnectionString); } private IDbConnection GetSlaverConnection() { int sc = SlaverConnectionStrings.Count(); if (sc > 0) { Random random = new Random(); int index = random.Next(0, sc); return GetConnection(SlaverConnectionStrings[index]); } else { _logger.LogInformation("没有设置从库，将从建立主库连接"); return GetMasterConnection(); } } } public enum DataSourceEnum { MASTER, SLAVE }
赞 (0)