完成全部查询的链接模式支持,并且针对所有代码进行了优化和单元测试发布x.3.1.67

Author: xuejmnet

nuget-publish.bat

@@ -1,9 +1,9 @@
 :start
 ::定义版本
-set EFCORE2=2.3.1.66
-set EFCORE3=3.3.1.66
-set EFCORE5=5.3.1.66
-set EFCORE6=6.3.1.66
+set EFCORE2=2.3.1.67
+set EFCORE3=3.3.1.67
+set EFCORE5=5.3.1.67
+set EFCORE6=6.3.1.67
 
 ::删除所有bin与obj下的文件
 @echo off

src/ShardingCore/Core/ConnectionModeEnum.cs

@@ -8,16 +8,16 @@ namespace ShardingCore.Core
 {
     /// <summary>
     /// 链接模式,可以由用户自行指定，使用内存限制,和连接数限制或者系统自行选择最优
-    /// STREAM_MERGE的意思是最小化内存使用率，就是非一次性获取所有数据然后采用流式聚合
-    /// IN_MEMORY_MERGE的意思是最小化连接并发数，就是单次查询并发连接数为设置的连接数<see cref="IShardingConfigOption.MaxQueryConnectionsLimit"/>。因为有限制，所以无法一直挂起连接，必须全部获取到内存后进行内存聚合
+    /// MEMORY_STRICTLY的意思是最小化内存使用率，就是非一次性获取所有数据然后采用流式聚合
+    /// CONNECTION_STRICTLY的意思是最小化连接并发数，就是单次查询并发连接数为设置的连接数<see cref="IShardingConfigOption.MaxQueryConnectionsLimit"/>。因为有限制，所以无法一直挂起连接，必须全部获取到内存后进行内存聚合
     /// 系统自行选择会根据用户的配置采取最小化连接数，但是如果遇到分页则会根据分页策略采取内存限制，因为skip过大会导致内存爆炸
     /// </summary>
     public enum ConnectionModeEnum
     {
-        //流式聚合 同时会有多个链接
-        STREAM_MERGE,
-        //内存聚合 连接数会有限制
-        IN_MEMORY_MERGE,
+        //内存限制模式最小化内存聚合 流式聚合 同时会有多个链接
+        MEMORY_STRICTLY,
+        //连接数限制模式最小化并发连接数 内存聚合 连接数会有限制
+        CONNECTION_STRICTLY,
         //系统自动选择内存还是流式聚合
         SYSTEM_AUTO
     }

src/ShardingCore/Extensions/ConnectionModeExtension.cs

@@ -19,7 +19,7 @@ public static async Task<T> ReleaseConnectionAsync<T>(this Task<T> executeTask,
             }
             finally
             {
-                if (connectionMode == ConnectionModeEnum.IN_MEMORY_MERGE)
+                if (connectionMode == ConnectionModeEnum.CONNECTION_STRICTLY)
                 {
 #if !EFCORE2
                     await dbContext.DisposeAsync();

src/ShardingCore/Sharding/MergeEngines/Abstractions/AbstractBaseMergeEngine.cs

@@ -9,7 +9,9 @@
 using System.Threading;
 using System.Threading.Tasks;
 using ShardingCore.Core;
+using ShardingCore.Core.VirtualRoutes.TableRoutes.RoutingRuleEngine;
 using ShardingCore.Sharding.MergeEngines.Common;
+using ShardingCore.Sharding.MergeEngines.Common.Abstractions;
 using ShardingCore.Sharding.StreamMergeEngines;
 
 namespace ShardingCore.Sharding.MergeEngines.Abstractions
@@ -66,62 +68,16 @@ internal abstract class AbstractBaseMergeEngine<TEntity>
 
         //}
 
-        public Task<LinkedList<TResult2>>[] GetDataSourceGroupAndExecutorGroup<TResult,TResult2>(Func<SqlExecutorUnit,Task<TResult2>> sqlExecutorUnitExecuteAsync,CancellationToken cancellationToken=new CancellationToken())
+        public Task<LinkedList<TResult>>[] GetDataSourceGroupAndExecutorGroup<TResult>(IEnumerable<ISqlRouteUnit> sqlRouteUnits,Func<SqlExecutorUnit,Task<TResult>> sqlExecutorUnitExecuteAsync,CancellationToken cancellationToken=new CancellationToken())
         {
-            var streamMergeContext = GetStreamMergeContext();
-            var maxQueryConnectionsLimit = streamMergeContext.GetMaxQueryConnectionsLimit();
-
-            var waitTaskQueue = streamMergeContext.DataSourceRouteResult.IntersectDataSources.SelectMany(
-                dataSourceName =>
-                {
-                    return streamMergeContext.TableRouteResults.Select(routeResult =>
-                        new SqlRouteUnit(dataSourceName, routeResult));
-                }).GroupBy(o => o.DataSourceName).Select(sqlGroups =>
-            {
-                var sqlCount = sqlGroups.Count();
-                //根据用户配置单次查询期望并发数
-                int exceptCount =
-                    Math.Max(
-                        0 == sqlCount % maxQueryConnectionsLimit
-                            ? sqlCount / maxQueryConnectionsLimit
-                            : sqlCount / maxQueryConnectionsLimit + 1, 1);
-                //计算应该使用那种链接模式
-                ConnectionModeEnum connectionMode = CalcConnectionMode(streamMergeContext.GetConnectionMode(),
-                    streamMergeContext.GetUseMemoryLimitWhileSkip(), maxQueryConnectionsLimit, sqlCount,
-                    streamMergeContext.Skip);
-                var sqlExecutorUnitPartitions = sqlGroups
-                    .Select((o, i) => new { Obj = o, index = i % exceptCount }).GroupBy(o => o.index)
-                    .Select(o => o.Select(g => new SqlExecutorUnit(connectionMode, g.Obj)).ToList()).ToList();
-                return sqlExecutorUnitPartitions.Select(o => new SqlExecutorGroup<SqlExecutorUnit>(o)).ToList();
-            }).Select(executorGroups =>
+            var waitTaskQueue = AggregateQueryByDataSourceName(sqlRouteUnits).Select(GetSqlExecutorGroups).Select(executorGroups =>
             {
                 return Task.Run(async () =>
                 {
-                    LinkedList<TResult2> result = new LinkedList<TResult2>();
+                    LinkedList<TResult> result = new LinkedList<TResult>();
                     foreach (var executorGroup in executorGroups)
                     {
-                        var executorGroupParallelExecuteTasks = executorGroup.Groups.Select(executor =>
-                        {
-                            return Task.Run(async () =>
-                            {
-                                return await sqlExecutorUnitExecuteAsync(executor);
-                                //var dataSourceName = executor.RouteUnit.DataSourceName;
-                                //var routeResult = executor.RouteUnit.TableRouteResult;
-
-                                //var asyncExecuteQueryable =
-                                //    CreateAsyncExecuteQueryable<TResult>(dataSourceName, routeResult);
-
-
-                                //var queryResult = await efQuery(asyncExecuteQueryable);
-
-                                //return new RouteQueryResult<TResult>(dataSourceName, routeResult, queryResult);
-                                //return await AsyncParallelResultExecute(asyncExecuteQueryable, dataSourceName,
-                                //    routeResult, efQuery,
-                                //    cancellationToken);
-
-                            }, cancellationToken);
-                        }).ToArray();
-                        var routeQueryResults = (await Task.WhenAll(executorGroupParallelExecuteTasks)).ToList();
+                        var routeQueryResults = await ExecuteAsync<TResult>(executorGroup.Groups, sqlExecutorUnitExecuteAsync,cancellationToken);
                         foreach (var routeQueryResult in routeQueryResults)
                         {
                             result.AddLast(routeQueryResult);
@@ -134,25 +90,78 @@ public Task<LinkedList<TResult2>>[] GetDataSourceGroupAndExecutorGroup<TResult,T
             return waitTaskQueue;
         }
 
+        protected virtual IEnumerable<ISqlRouteUnit> GetDefaultSqlRouteUnits()
+        {
+
+            var streamMergeContext = GetStreamMergeContext();
+
+            return streamMergeContext.DataSourceRouteResult.IntersectDataSources.SelectMany(
+                dataSourceName =>
+                {
+                    return streamMergeContext.TableRouteResults.Select(routeResult =>
+                        new SqlRouteUnit(dataSourceName, routeResult));
+                });
+        }
+        protected virtual IEnumerable<IGrouping<string, ISqlRouteUnit>> AggregateQueryByDataSourceName(IEnumerable<ISqlRouteUnit> sqlRouteUnits)
+        {
+            return sqlRouteUnits.GroupBy(o => o.DataSourceName);
+        }
 
+        protected List<SqlExecutorGroup<SqlExecutorUnit>> GetSqlExecutorGroups(IGrouping<string, ISqlRouteUnit> sqlGroups)
+        {
+            var streamMergeContext = GetStreamMergeContext();
+            var maxQueryConnectionsLimit = streamMergeContext.GetMaxQueryConnectionsLimit();
+            var sqlCount = sqlGroups.Count();
+            //根据用户配置单次查询期望并发数
+            int exceptCount =
+                Math.Max(
+                    0 == sqlCount % maxQueryConnectionsLimit
+                        ? sqlCount / maxQueryConnectionsLimit
+                        : sqlCount / maxQueryConnectionsLimit + 1, 1);
+            //计算应该使用那种链接模式
+            ConnectionModeEnum connectionMode = streamMergeContext.GetConnectionMode(sqlCount);
+            var sqlExecutorUnitPartitions = sqlGroups
+                .Select((o, i) => new { Obj = o, index = i % exceptCount }).GroupBy(o => o.index)
+                .Select(o => o.Select(g => new SqlExecutorUnit(connectionMode, g.Obj)).ToList()).ToList();
+            return sqlExecutorUnitPartitions.Select(o => new SqlExecutorGroup<SqlExecutorUnit>(o)).ToList();
+        }
 
-        protected ConnectionModeEnum CalcConnectionMode(ConnectionModeEnum currentConnectionMode, int useMemoryLimitWhileSkip, int maxQueryConnectionsLimit, int sqlCount,int? skip)
+        protected async Task<LinkedList<TResult>> ExecuteAsync<TResult>(List<SqlExecutorUnit> sqlExecutorUnits, Func<SqlExecutorUnit, Task<TResult>> sqlExecutorUnitExecuteAsync, CancellationToken cancellationToken = new CancellationToken())
         {
-            switch (currentConnectionMode)
+            if (sqlExecutorUnits.Count <= 0)
+            {
+                return new LinkedList<TResult>();
+            }
+            else
             {
-                case ConnectionModeEnum.STREAM_MERGE:
-                case ConnectionModeEnum.IN_MEMORY_MERGE: return currentConnectionMode;
-                default:
+                var result=new LinkedList<TResult>();
+                Task<TResult>[] tasks=null;
+                if (sqlExecutorUnits.Count > 1)
                 {
-                    if (skip.HasValue && skip.Value > useMemoryLimitWhileSkip)
+                    tasks  = sqlExecutorUnits.Skip(1).Select(sqlExecutorUnit =>
                     {
-                        return ConnectionModeEnum.STREAM_MERGE;
-                    }
-                    return maxQueryConnectionsLimit < sqlCount
-                        ? ConnectionModeEnum.IN_MEMORY_MERGE
-                        : ConnectionModeEnum.STREAM_MERGE; ;
+                        return Task.Run(async () =>
+                        {
+                            return await sqlExecutorUnitExecuteAsync(sqlExecutorUnit);
+
+                        }, cancellationToken);
+                    }).ToArray();
+                }
+                else
+                {
+                    tasks = Array.Empty<Task<TResult>>();
+                }
+                var firstResult = await sqlExecutorUnitExecuteAsync(sqlExecutorUnits[0]);
+                result.AddLast(firstResult);
+                var otherResults = await Task.WhenAll(tasks);
+                foreach (var otherResult in otherResults)
+                {
+                    result.AddLast(otherResult);
                 }
+
+                return result;
             }
+
         }
     }
 }

src/ShardingCore/Sharding/MergeEngines/Abstractions/InMemoryMerge/AbstractInMemoryAsyncMergeEngine.cs

@@ -85,7 +85,8 @@ public AbstractInMemoryAsyncMergeEngine(MethodCallExpression methodCallExpressio
 
         public async Task<List<RouteQueryResult<TResult>>> ExecuteAsync<TResult>(Func<IQueryable, Task<TResult>> efQuery, CancellationToken cancellationToken = new CancellationToken())
         {
-            var waitExecuteQueue = GetDataSourceGroupAndExecutorGroup<TResult, RouteQueryResult<TResult>>(
+            var defaultSqlRouteUnits = GetDefaultSqlRouteUnits();
+            var waitExecuteQueue = GetDataSourceGroupAndExecutorGroup<RouteQueryResult<TResult>>(defaultSqlRouteUnits,
                    async sqlExecutorUnit =>
                     {
                         var connectionMode = _mergeContext.RealConnectionMode(sqlExecutorUnit.ConnectionMode);

src/ShardingCore/Sharding/MergeEngines/Abstractions/StreamMerge/AbstractEnumeratorStreamMergeEngine.cs

@@ -114,12 +114,12 @@ public async Task<IStreamMergeAsyncEnumerator<TEntity>> AsyncParallelEnumerator(
             if (async)
             {
                 var asyncEnumerator = await GetAsyncEnumerator0(queryable);
-                return connectionMode==ConnectionModeEnum.STREAM_MERGE? new StreamMergeAsyncEnumerator<TEntity>(asyncEnumerator):new InMemoryStreamMergeAsyncEnumerator<TEntity>(asyncEnumerator);
+                return connectionMode==ConnectionModeEnum.MEMORY_STRICTLY? new StreamMergeAsyncEnumerator<TEntity>(asyncEnumerator):new InMemoryStreamMergeAsyncEnumerator<TEntity>(asyncEnumerator);
             }
             else
             {
                 var enumerator = GetEnumerator0(queryable);
-                return connectionMode == ConnectionModeEnum.STREAM_MERGE ? new StreamMergeAsyncEnumerator<TEntity>(enumerator):new InMemoryStreamMergeAsyncEnumerator<TEntity>(enumerator);
+                return connectionMode == ConnectionModeEnum.MEMORY_STRICTLY ? new StreamMergeAsyncEnumerator<TEntity>(enumerator):new InMemoryStreamMergeAsyncEnumerator<TEntity>(enumerator);
             }
         }
 

src/ShardingCore/Sharding/MergeEngines/Common/Abstractions/ISqlRouteUnit.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using ShardingCore.Core.VirtualRoutes.TableRoutes.RoutingRuleEngine;
+
+namespace ShardingCore.Sharding.MergeEngines.Common.Abstractions
+{
+    internal interface ISqlRouteUnit
+    {
+        string DataSourceName { get; }
+        TableRouteResult TableRouteResult { get; }
+    }
+}

src/ShardingCore/Sharding/MergeEngines/Common/SqlExecutorGroup.cs

@@ -6,7 +6,7 @@
 
 namespace ShardingCore.Sharding.MergeEngines.Common
 {
-    public sealed class SqlExecutorGroup<T>
+    internal sealed class SqlExecutorGroup<T>
     {
         public SqlExecutorGroup(List<T> groups)
         {

src/ShardingCore/Sharding/MergeEngines/Common/SqlExecutorUnit.cs

@@ -4,19 +4,20 @@
 using System.Text;
 using System.Threading.Tasks;
 using ShardingCore.Core;
+using ShardingCore.Sharding.MergeEngines.Common.Abstractions;
 
 namespace ShardingCore.Sharding.MergeEngines.Common
 {
-    public class SqlExecutorUnit
+    internal class SqlExecutorUnit
     {
 
-        public SqlExecutorUnit(ConnectionModeEnum connectionMode, SqlRouteUnit routeUnit)
+        public SqlExecutorUnit(ConnectionModeEnum connectionMode, ISqlRouteUnit routeUnit)
         {
             ConnectionMode = connectionMode;
             RouteUnit = routeUnit;
         }
 
-        public SqlRouteUnit RouteUnit { get; }
+        public ISqlRouteUnit RouteUnit { get; }
         public ConnectionModeEnum ConnectionMode { get; }
     }
 }

src/ShardingCore/Sharding/MergeEngines/Common/SqlRouteUnit.cs

@@ -4,11 +4,13 @@
 using System.Text;
 using System.Threading.Tasks;
 using ShardingCore.Core.VirtualRoutes.TableRoutes.RoutingRuleEngine;
+using ShardingCore.Sharding.MergeEngines.Common.Abstractions;
 
 namespace ShardingCore.Sharding.MergeEngines.Common
 {
-    public sealed class SqlRouteUnit
+    internal  class SqlRouteUnit: ISqlRouteUnit
     {
+
         public SqlRouteUnit(string dataSourceName, TableRouteResult tableRouteResult)
         {
             DataSourceName = dataSourceName;