From 61f76b7b1977104156e65eec23835d843fb0ef93 Mon Sep 17 00:00:00 2001 From: xuejiaming <326308290@qq.com> Date: Tue, 18 Oct 2022 09:02:54 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DMigrate=E5=9C=A8=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E4=B8=8A=E4=B8=8B=E6=96=87=E4=B8=ADblock=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- nuget-publish.bat | 2 +- .../Controllers/WeatherForecastController.cs | 2 + .../MyShardingExtension.cs | 243 ++++++++++++++++++ .../Controllers/WeatherForecastController.cs | 2 +- src/ShardingCore/EFCores/ShardingMigrator.cs | 4 +- src/ShardingCore/Extensions/TaskExtension.cs | 6 + .../Helpers/DynamicShardingHelper.cs | 6 +- 7 files changed, 258 insertions(+), 7 deletions(-) create mode 100644 samples/Sample.AutoCreateIfPresent/MyShardingExtension.cs diff --git a/nuget-publish.bat b/nuget-publish.bat index 768f233a..4d611277 100644 --- a/nuget-publish.bat +++ b/nuget-publish.bat @@ -1,6 +1,6 @@ :start ::定义版本 -set SHARDINGCORE=6.7.0.5 +set SHARDINGCORE=6.7.0.6 ::删除所有bin与obj下的文件 @echo off diff --git a/samples/Sample.AutoCreateIfPresent/Controllers/WeatherForecastController.cs b/samples/Sample.AutoCreateIfPresent/Controllers/WeatherForecastController.cs index 5f1784ec..ddb9a893 100644 --- a/samples/Sample.AutoCreateIfPresent/Controllers/WeatherForecastController.cs +++ b/samples/Sample.AutoCreateIfPresent/Controllers/WeatherForecastController.cs @@ -52,6 +52,8 @@ public class WeatherForecastController : ControllerBase } public async Task Query1() { + var s = Guid.NewGuid().ToString(); + var anyAsync = await _defaultDbContext.Set().AnyAsync(o=>o.Area==s); var list = await _defaultDbContext.Set().ToListAsync(); return Ok(list); } diff --git a/samples/Sample.AutoCreateIfPresent/MyShardingExtension.cs b/samples/Sample.AutoCreateIfPresent/MyShardingExtension.cs new file mode 100644 index 00000000..832e8101 --- /dev/null +++ b/samples/Sample.AutoCreateIfPresent/MyShardingExtension.cs @@ -0,0 +1,243 @@ +using System.Linq.Expressions; +using Microsoft.EntityFrameworkCore; +using ShardingCore.Core.VirtualRoutes.DataSourceRoutes.RouteRuleEngine; +using ShardingCore.Core.VirtualRoutes.TableRoutes; +using ShardingCore.Core.VirtualRoutes.TableRoutes.RouteTails.Abstractions; +using ShardingCore.Exceptions; +using ShardingCore.Extensions; +using ShardingCore.Sharding.Abstractions; + +namespace Sample.AutoCreateIfPresent; + +public static class MyShardingExtension +{ + + /// + /// 根据对象集合解析 + /// + /// + /// + /// + /// + /// + public static Dictionary>> BulkShardingEnumerable(this TShardingDbContext shardingDbContext, + IEnumerable entities) where TShardingDbContext : DbContext, IShardingDbContext where TEntity : class + { + if (entities.IsEmpty()) + return new Dictionary>>(); + var shardingRuntimeContext = shardingDbContext.GetShardingRuntimeContext(); + var entityType = typeof(TEntity); + var routeTailFactory = shardingRuntimeContext.GetRouteTailFactory(); + var virtualDataSource = shardingDbContext.GetVirtualDataSource(); + var dataSourceRouteManager = shardingRuntimeContext.GetDataSourceRouteManager(); + var tableRouteManager =shardingRuntimeContext.GetTableRouteManager(); + var entityMetadataManager = shardingRuntimeContext.GetEntityMetadataManager(); + var dataSourceNames = new Dictionary>>(); + var entitiesArray = entities as TEntity[] ?? entities.ToArray(); + var isShardingDataSource = entityMetadataManager.IsShardingDataSource(entityType); + var isShardingTable = entityMetadataManager.IsShardingTable(entityType); + if (!isShardingDataSource && !isShardingTable) + return new Dictionary>>() + { + { + virtualDataSource.DefaultDataSourceName, + new Dictionary>() + { + { + shardingDbContext.CreateGenericDbContext(entitiesArray[0]), + entitiesArray + } + } + } + }; + if (!isShardingDataSource) + { + var bulkDicEntries = new Dictionary>(); + dataSourceNames.Add(virtualDataSource.DefaultDataSourceName, bulkDicEntries); + + var tableRoute = tableRouteManager.GetRoute(entityType); + var allTails = tableRoute.GetTails().ToHashSet(); + foreach (var entity in entitiesArray) + { + BulkShardingTableEnumerable(shardingDbContext, virtualDataSource.DefaultDataSourceName, bulkDicEntries, + routeTailFactory, tableRoute, allTails, entity); + } + } + else + { + var virtualDataSourceRoute = dataSourceRouteManager.GetRoute(entityType); + var allDataSourceNames = virtualDataSourceRoute.GetAllDataSourceNames().ToHashSet(); + + var entityMetadata = entityMetadataManager.TryGet(entityType); + IVirtualTableRoute tableRoute = null; + ISet allTails = null; + if (isShardingTable) + { + tableRoute = tableRouteManager.GetRoute(entityType); + allTails = tableRoute.GetTails().ToHashSet(); + } + foreach (var entity in entitiesArray) + { + var shardingDataSourceValue = entity.GetPropertyValue(entityMetadata.ShardingDataSourceProperty.Name); + if (shardingDataSourceValue == null) + throw new ShardingCoreInvalidOperationException($" etities has null value of sharding data source value"); + var shardingDataSourceName = virtualDataSourceRoute.ShardingKeyToDataSourceName(shardingDataSourceValue); + if (!allDataSourceNames.Contains(shardingDataSourceName)) + throw new ShardingCoreException( + $" data source name :[{shardingDataSourceName}] all data source names:[{string.Join(",", allDataSourceNames)}]"); + if (!dataSourceNames.TryGetValue(shardingDataSourceName, out var bulkDicEntries)) + { + bulkDicEntries = new Dictionary>(); + dataSourceNames.Add(shardingDataSourceName, bulkDicEntries); + } + + if (isShardingTable) + { + BulkShardingTableEnumerable(shardingDbContext, shardingDataSourceName, bulkDicEntries, + routeTailFactory, tableRoute, allTails, entity); + } + else + BulkNoShardingTableEnumerable(shardingDbContext, shardingDataSourceName, bulkDicEntries, + routeTailFactory, entity); + } + } + + return dataSourceNames.ToDictionary(o => o.Key, + o => o.Value.Select(o => o.Value).ToDictionary(v => v.InnerDbContext, v => v.InnerEntities.Select(t => t))); + } + + private static void BulkShardingTableEnumerable(TShardingDbContext shardingDbContext, string dataSourceName, Dictionary> dataSourceBulkDicEntries, + IRouteTailFactory routeTailFactory,IVirtualTableRoute tableRoute, ISet allTails, TEntity entity) + where TShardingDbContext : DbContext, IShardingDbContext + where TEntity : class + { + var entityType = typeof(TEntity); + + var shardingKey = entity.GetPropertyValue(tableRoute.EntityMetadata.ShardingTableProperty.Name); + var tail = tableRoute.ShardingKeyToTail(shardingKey); + if (!allTails.Contains(tail)) + { + //不在alltails说明需要新增那么调用routewithvalue就会处理对应tail + var tableRouteUnit = tableRoute.RouteWithValue(new DataSourceRouteResult(new HashSet(new[] { dataSourceName })), + shardingKey); + if (tableRouteUnit.Tail != tail) + { + throw new ShardingCoreException( + $"sharding key route not match entity:{entityType.FullName},sharding key:{shardingKey},sharding tail:{tail}"); + } + allTails.Add(tail); + } + + var routeTail = routeTailFactory.Create(tail); + var routeTailIdentity = routeTail.GetRouteTailIdentity(); + if (!dataSourceBulkDicEntries.TryGetValue(routeTailIdentity, out var bulkDicEntry)) + { + var dbContext = shardingDbContext.GetShareDbContext(dataSourceName, routeTail); + bulkDicEntry = new BulkDicEntry(dbContext, new LinkedList()); + dataSourceBulkDicEntries.Add(routeTailIdentity, bulkDicEntry); + } + + bulkDicEntry.InnerEntities.AddLast(entity); + } + private static void BulkNoShardingTableEnumerable(TShardingDbContext shardingDbContext, string dataSourceName, Dictionary> dataSourceBulkDicEntries, IRouteTailFactory routeTailFactory, TEntity entity) + where TShardingDbContext : DbContext, IShardingDbContext + where TEntity : class + { + var routeTail = routeTailFactory.Create(string.Empty); + var routeTailIdentity = routeTail.GetRouteTailIdentity(); + if (!dataSourceBulkDicEntries.TryGetValue(routeTailIdentity, out var bulkDicEntry)) + { + var dbContext = shardingDbContext.GetShareDbContext(dataSourceName, routeTail); + bulkDicEntry = new BulkDicEntry(dbContext, new LinkedList()); + dataSourceBulkDicEntries.Add(routeTailIdentity, bulkDicEntry); + } + + bulkDicEntry.InnerEntities.AddLast(entity); + } + internal class BulkDicEntry + { + public BulkDicEntry(DbContext innerDbContext, LinkedList innerEntities) + { + InnerDbContext = innerDbContext; + InnerEntities = innerEntities; + } + + public DbContext InnerDbContext { get; } + public LinkedList InnerEntities { get; } + } + + public static Dictionary> BulkShardingTableEnumerable(this TShardingDbContext shardingDbContext, + IEnumerable entities) where TShardingDbContext : DbContext, IShardingDbContext + where TEntity : class + { + var shardingRuntimeContext = shardingDbContext.GetShardingRuntimeContext(); + var entityMetadataManager = shardingRuntimeContext.GetEntityMetadataManager(); + if (entityMetadataManager.IsShardingDataSource(typeof(TEntity))) + throw new ShardingCoreInvalidOperationException(typeof(TEntity).FullName); + //if (!entityMetadataManager.IsShardingTable(typeof(TEntity))) + // throw new ShardingCoreInvalidOperationException(typeof(TEntity).FullName); + if (entities.IsEmpty()) + return new Dictionary>(); + return shardingDbContext.BulkShardingEnumerable(entities).First().Value; + } + /// + /// 根据条件表达式解析 + /// + /// + /// + /// + /// + /// + public static IDictionary> BulkShardingExpression(this TShardingDbContext shardingDbContext, Expression> where) where TEntity : class + where TShardingDbContext : DbContext, IShardingDbContext + { + var shardingRuntimeContext = shardingDbContext.GetShardingRuntimeContext(); + var routeTailFactory = shardingRuntimeContext.GetRouteTailFactory(); + var dataSourceRouteManager = shardingRuntimeContext.GetDataSourceRouteManager(); + var tableRouteManager = shardingRuntimeContext.GetTableRouteManager();// (IVirtualTableManager)ShardingContainer.GetService(typeof(IVirtualTableManager<>).GetGenericType0(shardingDbContext.GetType())); + var entityMetadataManager = shardingRuntimeContext.GetEntityMetadataManager();// (IEntityMetadataManager)ShardingContainer.GetService(typeof(IEntityMetadataManager<>).GetGenericType0(shardingDbContext.GetType())); + + var dataSourceNames = dataSourceRouteManager.GetDataSourceNames(where); + var result = new Dictionary>(); + var entityType = typeof(TEntity); + + foreach (var dataSourceName in dataSourceNames) + { + if (!result.TryGetValue(dataSourceName, out var dbContexts)) + { + dbContexts = new LinkedList(); + result.Add(dataSourceName, dbContexts); + } + if (entityMetadataManager.IsShardingTable(entityType)) + { + var physicTables = tableRouteManager.RouteTo(entityType,new DataSourceRouteResult(dataSourceName),new ShardingTableRouteConfig(predicate: @where)); + if (physicTables.IsEmpty()) + throw new ShardingCoreException($"{where.ShardingPrint()} cant found any physic table"); + + var dbs = physicTables.Select(o => shardingDbContext.GetShareDbContext(dataSourceName, routeTailFactory.Create(o.Tail))).ToList(); + foreach (var dbContext in dbs) + { + dbContexts.AddLast(dbContext); + } + } + else + { + var dbContext = shardingDbContext.GetShareDbContext(dataSourceName, routeTailFactory.Create(string.Empty)); + dbContexts.AddLast(dbContext); + } + + } + + return result.ToDictionary(o => o.Key, o => (IEnumerable)o.Value); + } + + public static IEnumerable BulkShardingTableExpression(this TShardingDbContext shardingDbContext, Expression> where) where TEntity : class + where TShardingDbContext : DbContext, IShardingDbContext + { + var shardingRuntimeContext = shardingDbContext.GetShardingRuntimeContext(); + var entityMetadataManager = shardingRuntimeContext.GetEntityMetadataManager();// (IEntityMetadataManager)ShardingContainer.GetService(typeof(IEntityMetadataManager<>).GetGenericType0(shardingDbContext.GetType())); + if (entityMetadataManager.IsShardingDataSource(typeof(TEntity))) + throw new ShardingCoreInvalidOperationException(typeof(TEntity).FullName); + return shardingDbContext.BulkShardingExpression(where).First().Value; + } +} \ No newline at end of file diff --git a/samples/Sample.MySql/Controllers/WeatherForecastController.cs b/samples/Sample.MySql/Controllers/WeatherForecastController.cs index 194bd4c3..7ba7acc0 100644 --- a/samples/Sample.MySql/Controllers/WeatherForecastController.cs +++ b/samples/Sample.MySql/Controllers/WeatherForecastController.cs @@ -105,7 +105,7 @@ namespace Sample.MySql.Controllers // var firstOrDefault = _defaultTableDbContext.Set().FromSqlRaw($"select * from {nameof(SysUserMod)}").FirstOrDefault(); - var sysUserMods1 = _defaultTableDbContext.Set() + var sysUserMods1 = _defaultTableDbContext.Set().UseConnectionMode(1) .Select(o => new ssss(){ Id = o.Id, C = _abc.Select.Count(x => x.Id == o.Id) }).ToList(); var sysUserMods2 = _defaultTableDbContext.Set() .Select(o => new ssss(){ Id = o.Id, C = GetAll().Count(x => x.Id == o.Id) }).ToList(); diff --git a/src/ShardingCore/EFCores/ShardingMigrator.cs b/src/ShardingCore/EFCores/ShardingMigrator.cs index 9c4642f1..727748ea 100644 --- a/src/ShardingCore/EFCores/ShardingMigrator.cs +++ b/src/ShardingCore/EFCores/ShardingMigrator.cs @@ -70,7 +70,7 @@ namespace ShardingCore.EFCores #endif public override void Migrate(string targetMigration = null) { - this.MigrateAsync(targetMigration).WaitAndUnwrapException(); + this.MigrateAsync(targetMigration).WaitAndUnwrapException(false); // base.Migrate(targetMigration); } @@ -78,7 +78,7 @@ namespace ShardingCore.EFCores { var virtualDataSource = _shardingRuntimeContext.GetVirtualDataSource(); var allDataSourceNames = virtualDataSource.GetAllDataSourceNames(); - await DynamicShardingHelper.DynamicMigrateWithDataSourcesAsync(_shardingRuntimeContext, allDataSourceNames, null,targetMigration,cancellationToken); + await DynamicShardingHelper.DynamicMigrateWithDataSourcesAsync(_shardingRuntimeContext, allDataSourceNames, null,targetMigration,cancellationToken).ConfigureAwait(false); } #if NET6_0 || NET5_0 || NETSTANDARD2_1 diff --git a/src/ShardingCore/Extensions/TaskExtension.cs b/src/ShardingCore/Extensions/TaskExtension.cs index a3936048..ae5dc377 100644 --- a/src/ShardingCore/Extensions/TaskExtension.cs +++ b/src/ShardingCore/Extensions/TaskExtension.cs @@ -40,6 +40,12 @@ namespace ShardingCore.Extensions throw new ArgumentNullException(nameof(task)); task.GetAwaiter().GetResult(); } + public static void WaitAndUnwrapException(this Task task,bool continueOnCapturedContext) + { + if (task == null) + throw new ArgumentNullException(nameof(task)); + task.ConfigureAwait(continueOnCapturedContext).GetAwaiter().GetResult(); + } /// /// Waits for the task to complete, unwrapping any exceptions. diff --git a/src/ShardingCore/Helpers/DynamicShardingHelper.cs b/src/ShardingCore/Helpers/DynamicShardingHelper.cs index a9e422b4..f2c08b88 100644 --- a/src/ShardingCore/Helpers/DynamicShardingHelper.cs +++ b/src/ShardingCore/Helpers/DynamicShardingHelper.cs @@ -76,13 +76,13 @@ namespace ShardingCore.Helpers foreach (var migrationUnits in partitionMigrationUnits) { var migrateUnits = migrationUnits.Select(o =>new MigrateUnit(shellDbContext,o)).ToList(); - await ExecuteMigrateUnitsAsync(shardingRuntimeContext,migrateUnits,targetMigration,cancellationToken); + await ExecuteMigrateUnitsAsync(shardingRuntimeContext,migrateUnits,targetMigration,cancellationToken).ConfigureAwait((false)); } //包含默认默认的单独最后一次处理 if (allDataSourceNames.Contains(defaultDataSourceName)) { - await ExecuteMigrateUnitsAsync(shardingRuntimeContext,new List(){new MigrateUnit(shellDbContext,defaultDataSourceName)},targetMigration,cancellationToken); + await ExecuteMigrateUnitsAsync(shardingRuntimeContext,new List(){new MigrateUnit(shellDbContext,defaultDataSourceName)},targetMigration,cancellationToken).ConfigureAwait(false); } } } @@ -120,7 +120,7 @@ namespace ShardingCore.Helpers },cancellationToken); }).ToArray(); - await TaskHelper.WhenAllFastFail(migrateTasks); + await TaskHelper.WhenAllFastFail(migrateTasks).ConfigureAwait(false); } public static DbContextOptions CreateShellDbContextOptions(IShardingRuntimeContext shardingRuntimeContext,Type dbContextType,string dataSourceName)