.NET 分表分庫動態化處理

介紹

本期主角:ShardingCore?一款ef-core下高性能、輕量級針對分表分庫讀寫分離的解決方案，具有零依賴、零學習成本、零業務代碼入侵

我不是efcore怎么辦

這邊肯定有小伙伴要問有沒有不是efcore的,我這邊很確信的和你講有并且適應所有的ADO.NET包括sqlhelper
ShardingConnector?一款基于ado.net下的高性能分表分庫解決方案目前已有demo案例，這個框架你可以認為是.Net版本的ShardingSphere但是目前僅實現了ShardingSphere-JDBC,后續我將會實現ShardingSphere-Proxy希望各位.Neter多多關注

背景

最近有個小伙伴來問我,分表下他有一批數據,這個數據是白天可能會相對比較頻繁數據錄入,但是到了晚上可能基本上就沒有對應的數據了,因為看到了我的框架,本來想以按小時來實現分表但是這么以來可能會導致一天有24張表,表多的情況下還導致了數據分布不均勻,這是一個很嚴重的問題因為可能以24小時制會讓8-17這幾張白天的表數據很多,但是晚上和凌晨的表基本沒有數據,沒有數據其實意味著這些表其實不需要去查詢,基于這個情況想來問我應該如何實現這個自定義的路由。

聽了他的需求,其實我這邊又進行了一次確認,針對這個場景更多的其實是這個小伙伴需要的是按需分片，實時建表，來保證需要的數據進行合理的插入，那么我們應該如何在ShardingCore下實現這么一個需求呢，廢話不多說直接開始吧~~~

創建項目

本次需求我們以mysql作為測試數據庫，然后使用efcore6作為數據庫驅動orm來實現怎么處理才能達到這個效果的分表分庫（本次只涉及分表）。

新建一個項目

添加依賴

//請安裝最新版本第一個版本號6代表efcore的版本號
Install-Package ShardingCore -Version 6.4.3.4Install-Package Microsoft.EntityFrameworkCore.SqlServer -Version 6.0.1

新建一個對象表,配置對應的數據庫映射關系并且關聯到dbcontext

//創建數據庫對象public class OrderByHour{public string Id { get; set; }public DateTime CreateTime { get; set; }public string Name { get; set; }}
//映射對象結構到數據庫public class OrderByHourMap:IEntityTypeConfiguration<OrderByHour>{public void Configure(EntityTypeBuilder<OrderByHour> builder){builder.HasKey(o => o.Id);builder.Property(o => o.Id).IsRequired().HasMaxLength(50);builder.Property(o => o.Name).IsRequired().HasMaxLength(128);builder.ToTable(nameof(OrderByHour));}}
//創建dbcontext為efcore所用上下文public class DefaultDbContext:AbstractShardingDbContext,IShardingTableDbContext{public DefaultDbContext(DbContextOptions<DefaultDbContext> options) : base(options){}protected override void OnModelCreating(ModelBuilder modelBuilder){base.OnModelCreating(modelBuilder);modelBuilder.ApplyConfiguration(new OrderByHourMap());}public IRouteTail RouteTail { get; set; }}

到這邊其實只需要啟動時候依賴注入

services.AddDbContext<DefaultDbContext>(o=>o.UseMySql(xxxx));

那么efcore就可以運行了,這么一看其實并沒有很復雜而且IEntityTypeConfiguration也不是必須的,efcore允許使用attribute來實現
當然DefaultDbContext:AbstractShardingDbContext,IShardingTableDbContext這一部分在原生efcore中應該是DefaultDbContext:DbContext

創建分片路由

首先我們來看一下ShardingCore針對分片路由的自定義情況的分析，通過文檔我們可以了解到，如果想要使用自定義路由那么你只需要自己新建一個路由并且繼承實現AbstractShardingOperatorVirtualTableRoute,當然這是分表的，分庫是AbstractShardingOperatorVirtualDataSourceRoute.

接下來我們新建一個路由并且實現分表操作。

public class orderByHourRoute : AbstractShardingOperatorVirtualTableRoute<OrderByHour, DateTime>{public override string ShardingKeyToTail(object shardingKey){throw new NotImplementedException();}public override List<string> GetAllTails(){throw new NotImplementedException();}public override void Configure(EntityMetadataTableBuilder<OrderByHour> builder){throw new NotImplementedException();}public override Expression<Func<string, bool>> GetRouteToFilter(DateTime shardingKey, ShardingOperatorEnum shardingOperator){throw new NotImplementedException();}}

接下來我們依次來實現并且說明各個接口。

ShardingKeyToTail:將你的對象轉成數據庫的后綴尾巴,比如你是按月分片，那么你的分片值大概率是datetime，那么只需要datetime.ToString("yyyyMM")就可以獲取到分片后綴
GetAllTails:返回集合,集合是數據庫現有的當前表的所有后綴,僅程序啟動時被調用,這個接口就是需要你返回當前數據庫中當前表在系統里面有多少張表，然后返回這些表的后綴
Configure:配置當前對象按什么字段分片
GetRouteToFilter:因為ShardingCore內存有當前所有表的后綴,假設后綴為list集合，返回的Expression<Func<string, bool>>在經過And、Or后的組合進行Compile(),然后對list.Where(expression.Compile()).ToList()就可以返回對應的本次查詢的后綴信息

廢話不多說針對這個條件我們直接開始操作完成路由的實現

路由的編寫

1.ShardingKeyToTail:因為我們是按小時分表所以格式化值后綴我們采用日期格式化

//因為分片建是DateTime類型所以直接強轉public override string ShardingKeyToTail(object shardingKey){var dateTime = (DateTime)shardingKey;return ShardingKeyFormat(dateTime);}private string ShardingKeyFormat(DateTime dateTime){var tail = $"{dateTime:yyyyMMddHH}";return tail;}

2.Configure:分表配置

public override void Configure(EntityMetadataTableBuilder<OrderByHour> builder){builder.ShardingProperty(o => o.CreateTime);}

3.GetRouteToFilter:路由比較，因為是時間字符串的后綴具有和按年，按月等相似的屬性所以我們直接參考默認路由來實現

public override Expression<Func<string, bool>> GetRouteToFilter(DateTime shardingKey, ShardingOperatorEnum shardingOperator){var t = ShardingKeyFormat(shardingKey);switch (shardingOperator){case ShardingOperatorEnum.GreaterThan:case ShardingOperatorEnum.GreaterThanOrEqual:return tail => String.Compare(tail, t, StringComparison.Ordinal) >= 0;case ShardingOperatorEnum.LessThan:{var currentHourBeginTime = new DateTime(shardingKey.Year,shardingKey.Month,shardingKey.Day,shardingKey.Hour,0,0);//處于臨界值 不應該被返回if (currentHourBeginTime == shardingKey)return tail => String.Compare(tail, t, StringComparison.Ordinal) < 0;return tail => String.Compare(tail, t, StringComparison.Ordinal) <= 0;}case ShardingOperatorEnum.LessThanOrEqual:return tail => String.Compare(tail, t, StringComparison.Ordinal) <= 0;case ShardingOperatorEnum.Equal: return tail => tail == t;default:{return tail => true;}}}

4.GetAllTails:比較特殊我們因為并不是連續生成的所以沒辦法使用起始時間然后一直推到當前時間來實現后綴的返回，只能依靠ado.net的能力讀取數據庫然后返回對應的表后綴，當然你也可以使用redis等三方工具來存儲

//1.構造函數注入 IVirtualDataSourceManager<DefaultDbContext> virtualDataSourceManager//2/mysql的ado.net讀取數據庫表（sqlserver和mysql有差異自行百度或者查看ShardingCore的SqlServerTableEnsureManager類）private const string CurrentTableName = nameof(OrderByHour);private const string Tables = "Tables";private const string TABLE_SCHEMA = "TABLE_SCHEMA";private const string TABLE_NAME = "TABLE_NAME";private readonly ConcurrentDictionary<string, object?> _tails = new ConcurrentDictionary<string, object?>();/// <summary>/// 如果你是非mysql數據庫請自行實現這個方法返回當前類在數據庫已經存在的后綴/// 僅啟動時調用/// </summary>/// <returns></returns>public override List<string> GetAllTails(){//啟動尋找有哪些表后綴using (var connection = new MySqlConnection(_virtualDataSourceManager.GetCurrentVirtualDataSource().DefaultConnectionString)){connection.Open();var database = connection.Database;using (var dataTable = connection.GetSchema(Tables)){for (int i = 0; i < dataTable.Rows.Count; i++){var schema = dataTable.Rows[i][TABLE_SCHEMA];if (database.Equals($"{schema}", StringComparison.OrdinalIgnoreCase)){var tableName = dataTable.Rows[i][TABLE_NAME]?.ToString()??string.Empty;if (tableName.StartsWith(CurrentTableName, StringComparison.OrdinalIgnoreCase)){//如果沒有下劃線那么需要CurrentTableName.Length有下劃線就要CurrentTableName.Length+1_tails.TryAdd(tableName.Substring(CurrentTableName.Length),null);}}}}}return _tails.Keys.ToList();}

動態創建添加表

到目前為止我們已經完成了路由的靜態分片的處理，但是還有一點需要處理就是如何在插入值得時候判斷當前有沒有對應的數據庫表是否需要創建等操作

查看AbstractShardingOperatorVirtualTableRoute分表抽象類的父類我們發現當前抽象類有兩個地方會調用路由的獲取判斷方法

DoRouteWithPredicate:使用條件路由也就是where后面的表達式
RouteWithValue:使用值路由也就是我們的新增和修改整個對象的時候會被調用

所以通過上述流程的梳理我們可以知道只需要在RouteWithValue處進行動手腳即可，又因為我們需要動態建表所以我們可以參考默認路由的自動建表的代碼進行參考
AbstractShardingAutoCreateOperatorVirtualTableRoute下的ExecuteAsync

private readonly IVirtualDataSourceManager<DefaultDbContext> _virtualDataSourceManager;private readonly IVirtualTableManager<DefaultDbContext> _virtualTableManager;private readonly IShardingTableCreator<DefaultDbContext> _shardingTableCreator;private readonly ConcurrentDictionary<string, object?> _tails = new ConcurrentDictionary<string, object?>();private readonly object _lock = new object();public OrderByHourRoute(IVirtualDataSourceManager<DefaultDbContext> virtualDataSourceManager,IVirtualTableManager<DefaultDbContext> virtualTableManager, IShardingTableCreator<DefaultDbContext> shardingTableCreator){_virtualDataSourceManager = virtualDataSourceManager;_virtualTableManager = virtualTableManager;_shardingTableCreator = shardingTableCreator;}public override IPhysicTable RouteWithValue(List<IPhysicTable> allPhysicTables, object shardingKey){var shardingKeyToTail = ShardingKeyToTail(shardingKey);if (!_tails.TryGetValue(shardingKeyToTail,out var _)){lock (_lock){if (!_tails.TryGetValue(shardingKeyToTail,out var _)){var virtualTable = _virtualTableManager.GetVirtualTable(typeof(OrderByHour));
//必須先執行AddPhysicTable在進行CreateTable_virtualTableManager.AddPhysicTable(virtualTable, new DefaultPhysicTable(virtualTable, shardingKeyToTail));try{_shardingTableCreator.CreateTable<OrderByHour>(_virtualDataSourceManager.GetCurrentVirtualDataSource().DefaultDataSourceName, shardingKeyToTail);}catch (Exception ex){Console.WriteLine("嘗試添加表失敗" + ex);}_tails.TryAdd(shardingKeyToTail,null);}}}var needRefresh = allPhysicTables.Count != _tails.Count;if (needRefresh){var virtualTable = _virtualTableManager.GetVirtualTable(typeof(OrderByHour));//修復可能導致迭代器遍歷時添加的bugvar keys = _tails.Keys.ToList();foreach (var tail in keys){var hashSet = allPhysicTables.Select(o=>o.Tail).ToHashSet();if (!hashSet.Contains(tail)){var tables = virtualTable.GetAllPhysicTables();var physicTable = tables.FirstOrDefault(o=>o.Tail==tail);if (physicTable!= null){allPhysicTables.Add(physicTable);}}}}var physicTables = allPhysicTables.Where(o => o.Tail== shardingKeyToTail).ToList();if (physicTables.IsEmpty()){throw new ShardingCoreException($"sharding key route not match {EntityMetadata.EntityType} -> [{EntityMetadata.ShardingTableProperty.Name}] ->【{shardingKey}】 all tails ->[{string.Join(",", allPhysicTables.Select(o=>o.FullName))}]");}if (physicTables.Count > 1)throw new ShardingCoreException($"more than one route match table:{string.Join(",", physicTables.Select(o => $"[{o.FullName}]"))}");return physicTables[0];}

通過和父類的比較我們只是在對應的根據值判斷當前系統是否存在xx表如果不存在就在ShardingCore上插入AddPhysicTable然后CreateTable最后_tails.TryAdd(shardingKeyToTail,null);

needRefresh處的代碼需要針對如果當前需要和傳入的全量表進行匹配因為新加的表后綴不在全量表里面所以需要先進行對其的處理然后再進行執行

啟動配置必不可少

ILoggerFactory efLogger = LoggerFactory.Create(builder =>
{builder.AddFilter((category, level) => category == DbLoggerCategory.Database.Command.Name && level == LogLevel.Information).AddConsole();
});var builder = WebApplication.CreateBuilder(args);// Add services to the container.builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
builder.Services.AddShardingDbContext<DefaultDbContext>().AddEntityConfig(o =>{o.ThrowIfQueryRouteNotMatch = false;o.CreateShardingTableOnStart = true;o.EnsureCreatedWithOutShardingTable = true;o.AddShardingTableRoute<OrderByHourRoute>();}).AddConfig(o =>{o.ConfigId = "c1";o.AddDefaultDataSource("ds0", "server=127.0.0.1;port=3306;database=shardingTest;userid=root;password=root;");o.UseShardingQuery((conn, b) =>{b.UseMySql(conn, new MySqlServerVersion(new Version())).UseLoggerFactory(efLogger);});o.UseShardingTransaction((conn, b) =>{b.UseMySql(conn, new MySqlServerVersion(new Version())).UseLoggerFactory(efLogger);});o.ReplaceTableEnsureManager(sp=>new MySqlTableEnsureManager<DefaultDbContext>());}).EnsureConfig();
var app = builder.Build();// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{app.UseSwagger();app.UseSwaggerUI();
}
app.Services.GetRequiredService<IShardingBootstrapper>().Start();
app.UseAuthorization();app.MapControllers();app.Run();