重构Consumer模式

Doc/UpgradeLog/Sers2.1.txt

@@ -149,5 +149,10 @@ Serslot
 	ServiceCenter添加输出信息到txt功能
 
 
+--------------------------------------
+[2021-08-20]
+  [Serslot]修改Async模式的bug。bug版本试图通过调用IHttpApplication.ProcessRequestAsync方法实现异步，然而此方法不会开启异步线程，故会阻塞请求。
+
+
 --------------------------------------
 

Publish/ReleaseFile/docker-image/docker导入导出镜像.md

@@ -1,5 +1,5 @@
 #导出镜像
-tag=2.1.9
+tag=2.1.10-temp
 
 docker save -o /root/image/dotnet.2.1.tar serset/dotnet:2.1
 docker save -o /root/image/sers-cgateway-1.2.0.tar serset/sers-cgateway:1.2.0

Publish/ReleaseFile/docker-image/制作镜像Sers.md

@@ -37,7 +37,7 @@ cd /root/image
 
 
 #构建镜像并推送到 Docker Hub 
-export tag=2.1.9
+export tag=2.1.10-temp
 
 cd sers
 docker buildx build . -t serset/sers:${tag} -t serset/sers --platform=linux/amd64,linux/arm64,linux/arm/v7 --push

dotnet/Gateway/App.Gateway/App.Gateway.csproj

@@ -8,7 +8,7 @@
 	<PropertyGroup>
 		<OutputType>Exe</OutputType>
 		<TargetFramework>netcoreapp2.1</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 		<PackageProjectUrl>https://github.com/serset/Sers</PackageProjectUrl>
 	</PropertyGroup>
 

dotnet/Library/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe-额外命名管道/Sers.CL.Ipc.NamedPipe.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
-    <Version>2.1.9</Version>
+    <Version>2.1.10-temp</Version>
     <Description>https://github.com/serset/Sers</Description>
   </PropertyGroup>
 

dotnet/Library/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Sers.CL.Ipc.NamedPipe.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/Ipc/Sers.CL.Ipc.SharedMemory/Sers.CL.Ipc.SharedMemory.csproj

@@ -2,7 +2,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/Socket/Sers.CL.Socket.Iocp/Sers.CL.Socket.Iocp.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/Socket/Sers.CL.Socket.ThreadWait/Sers.CL.Socket.ThreadWait.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/WebSocket/Sers.CL.WebSocket/Sers.CL.WebSocket.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/Zmq/FullDuplex/Sers.CL.Zmq.FullDuplex/Sers.CL.Zmq.FullDuplex.csproj

@@ -3,7 +3,7 @@
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
 		<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.CL/Zmq/ThreadWait/Sers.CL.ClrZmq.ThreadWait/Sers.CL.ClrZmq.ThreadWait.csproj

@@ -2,7 +2,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.Core/Sers.Core.Temp/Sers.Core.Temp.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
-    <Version>2.1.9</Version>
+    <Version>2.1.10-temp</Version>
   </PropertyGroup>
   
   <PropertyGroup>
@@ -17,10 +17,13 @@
   </PropertyGroup>
   
   <ItemGroup>
+    <Compile Remove="Sers.Core\Util\Consumer\**" />
     <Compile Remove="SersLoader\ApiDesc\Attribute\Valid\delete\**" />
     <Compile Remove="Vit.Core\Extensions\byte\**" />
+    <EmbeddedResource Remove="Sers.Core\Util\Consumer\**" />
     <EmbeddedResource Remove="SersLoader\ApiDesc\Attribute\Valid\delete\**" />
     <EmbeddedResource Remove="Vit.Core\Extensions\byte\**" />
+    <None Remove="Sers.Core\Util\Consumer\**" />
     <None Remove="SersLoader\ApiDesc\Attribute\Valid\delete\**" />
     <None Remove="Vit.Core\Extensions\byte\**" />
   </ItemGroup>
@@ -42,6 +45,7 @@
  
  
   <ItemGroup>
+    <Folder Include="Sers.Core\Util\" />
     <Folder Include="Vit.Core\Extensions\" />
   </ItemGroup>
 

dotnet/Library/Sers/Sers.Core/Sers.Core/CL/MessageOrganize/DefaultOrganize/RequestAdaptor.cs

@@ -99,10 +99,6 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
         public void Start()
         {
             //(x.1) task_DeliveryToOrganize_Processor
-            //task_DeliveryToOrganize_Processor.Stop();
-            task_DeliveryToOrganize_Processor.processor = DeliveryToOrganize_ProcessFrame;
-            task_DeliveryToOrganize_Processor.workThreadCount = workThreadCount;
-            task_DeliveryToOrganize_Processor.name = "CL-RequestAdaptor-dealer";
             task_DeliveryToOrganize_Processor.Start(); 
 
             //(x.2) heartBeat thread
@@ -137,11 +133,16 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
         #region (x.x.4)构造函数       
         public RequestAdaptor(JObject config)
         {
-            workThreadCount = config["workThreadCount"]?.Deserialize<int?>() ?? workThreadCount;
             heartBeatTimeoutMs = config["heartBeatTimeoutMs"]?.Deserialize<int?>() ?? heartBeatTimeoutMs;
             heartBeatRetryCount = config["heartBeatRetryCount"]?.Deserialize<int?>() ?? heartBeatRetryCount;
             heartBeatIntervalMs = config["heartBeatIntervalMs"]?.Deserialize<int?>() ?? heartBeatIntervalMs;
-            requestTimeoutMs = config["requestTimeoutMs"]?.Deserialize<int?>() ?? requestTimeoutMs;
+
+            requestTimeoutMs = config["workThread"]?["timeoutMs"]?.Deserialize<int?>() ?? 300000;
+
+
+            task_DeliveryToOrganize_Processor = ConsumerFactory.CreateConsumer<DeliveryToOrganize_MessageFrame>(config["workThread"] as JObject);
+            task_DeliveryToOrganize_Processor.processor = DeliveryToOrganize_ProcessFrame;
+            task_DeliveryToOrganize_Processor.name = "CL-RequestAdaptor-dealer";
         }
         #endregion
 
@@ -157,21 +158,19 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
 
   
         long reqKeyIndex = CommonHelp.NewFastGuidLong();
-   
+
         #endregion
 
 
-        #region (x.x.2)config
+        #region (x.x.2) config      
 
-        /// <summary>
-        /// 后台处理消息的线程个数（单位个，默认2）
-        /// </summary>
-        int workThreadCount = 2;
 
         /// <summary>
         /// 请求超时时间（单位ms，默认300000）
         /// </summary>
-        public int requestTimeoutMs = 300000;
+        public int requestTimeoutMs;
+
+
 
         /// <summary>
         /// 心跳检测超时时间（单位ms，默认30000）
@@ -201,8 +200,7 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
         #region deliveryToOrganize_MessageFrameQueue  
 
 
-        IConsumer<DeliveryToOrganize_MessageFrame> task_DeliveryToOrganize_Processor = ConsumerFactory.CreateConsumer<DeliveryToOrganize_MessageFrame>();
- 
+        readonly IConsumer<DeliveryToOrganize_MessageFrame> task_DeliveryToOrganize_Processor; 
 
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

dotnet/Library/Sers/Sers.Core/Sers.Core/Module/Api/LocalApi/LocalApiService/LocalApiService.cs

@@ -1,12 +1,10 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Threading;
 using Vit.Extensions;
 using Vit.Core.Module.Log;
 using Sers.Core.Module.Rpc;
 using Vit.Core.Util.ConfigurationManager;
-using Vit.Core.Util.Threading;
 using Sers.Core.CL.CommunicationManage;
 using Sers.Core.CL.MessageOrganize;
 using Sers.Core.Module.Message;
@@ -14,6 +12,7 @@ using Vit.Core.Util.ComponentModel.SsError;
 using Sers.Core.Module.Api.LocalApi.Event;
 using Sers.Core.Util.Consumer;
 using System.Runtime.CompilerServices;
+using Newtonsoft.Json.Linq;
 
 namespace Sers.Core.Module.Api.LocalApi
 {
@@ -23,11 +22,16 @@ namespace Sers.Core.Module.Api.LocalApi
         /// <summary>
         /// 后台服务的线程个数（单位个，默认0,代表不开启服务）(appsettings.json :: Sers.LocalApiService.workThreadCount)
         /// </summary>
-        public int workThreadCount { get; set; }
+        public int workThreadCount { get => workThread.workThreadCount; set => workThread.workThreadCount = value; }
 
         public LocalApiService()
         {
-            workThreadCount = ConfigurationManager.Instance.GetByPath<int?>("Sers.LocalApiService.workThreadCount") ?? 0;
+            workThread = ConsumerFactory.CreateConsumer<RequestInfo>(ConfigurationManager.Instance.GetByPath<JObject>("Sers.LocalApiService.workThread"));
+
+            workThread.name = "LocalApiService";
+            workThread.processor = Consumer_Processor;
+            workThread.OnFinish = Consumer_OnFinish;
+            workThread.OnTimeout = Consumer_OnTimeout;
         }
 
         public void Init()
@@ -43,8 +47,10 @@ namespace Sers.Core.Module.Api.LocalApi
         public IEnumerable<IApiNode> apiNodes => apiNodeMng.apiNodes;
 
 
-        public ApiNodeMng ApiNodeMng => apiNodeMng;  
+        public ApiNodeMng ApiNodeMng => apiNodeMng;
+
 
+        static readonly ApiMessage const_ApiReply_Err_Timeout = new ApiMessage().InitAsApiReplyMessageByError(SsError.Err_HandleTimeout);
 
 
 
@@ -104,7 +110,21 @@ namespace Sers.Core.Module.Api.LocalApi
         #endregion
 
 
-        static readonly ApiMessage const_ApiReply_Err_Timeout = new ApiMessage().InitAsApiReplyMessageByError(SsError.Err_HandleTimeout);
+
+        #region Start Stop
+        public bool Start()
+        {
+            Stop();
+            return Consumer_Start();
+        }
+
+        public void Stop()
+        {
+            Consumer_Stop();
+        }
+        #endregion
+
+
 
 
         #region 后台服务
@@ -112,6 +132,7 @@ namespace Sers.Core.Module.Api.LocalApi
         {
             public IOrganizeConnection  conn;
             public ApiMessage apiRequest;
+            public ApiMessage apiReply;
             public Object sender;
             public Action<object, ApiMessage> callback;
         }
@@ -128,275 +149,105 @@ namespace Sers.Core.Module.Api.LocalApi
             requestInfo.apiRequest = apiRequest;
             requestInfo.callback = callback;
 
-            worker.Publish(requestInfo);  
+            Consumer_Publish(requestInfo);  
         }
+        #endregion
 
 
 
-        #region Start Stop
-
-    
-        /// <summary>
-        /// 后台调用Api的线程 
-        /// </summary>
-        IWorker worker = null;
-
-        public bool Start()
-        {
-            Stop();
-
-            int timeout_ms = ConfigurationManager.Instance.GetByPath<int?>("Sers.LocalApiService.timeout_ms") ?? 0;
+        #region Consumer
+       
 
-            if (timeout_ms > 0)
-            {
-                worker = new Worker_TimeLimit() { timeout_ms = timeout_ms };
-              
-            }
-            else
-            {
-                worker = new Worker();
-            }
+        readonly IConsumer<RequestInfo> workThread;
 
-          
-            return worker.Start(workThreadCount, CallLocalApi);
-        }
 
-        public void Stop()
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void Consumer_Publish(RequestInfo requestInfo)
         {
-            worker?.Stop();
-            worker = null;
+            workThread.Publish(requestInfo);
         }
-        #endregion
-
-
-       
-
-        #endregion
-
-
-
-        #region class Worker
-        interface IWorker { bool Start(int workThreadCount, Func<ApiMessage, ApiMessage> callLocalApi);void Stop(); void Publish(RequestInfo requestInfo); }
 
-        #region Worker
 
-
-        class Worker : IWorker
+        bool Consumer_Start()
         {
-
-            IConsumer<RequestInfo> taskToCallApi = ConsumerFactory.CreateConsumer<RequestInfo>();
-
-
-            Func<ApiMessage, ApiMessage> callLocalApi;
-
-
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            public void Publish(RequestInfo requestInfo)
-            {
-                taskToCallApi.Publish(requestInfo);
-            }
-
-
-            public bool Start(int workThreadCount, Func<ApiMessage, ApiMessage> callLocalApi)
+            if (workThread.IsRunning) return false;
+            try
             {
-                if (taskToCallApi.IsRunning) return false;
-
-                try
+                if (workThreadCount > 0)
                 {
-                    this.callLocalApi = callLocalApi; 
-
-                    taskToCallApi.name = "LocalApiService";
-                    taskToCallApi.workThreadCount = workThreadCount;
-                    taskToCallApi.processor = Processor;
+                    workThread.Start();
+                    Logger.Info("[LocalApiService] Started,workThreadCount:" + workThreadCount);
 
-                    if (workThreadCount > 0)
-                    {
-                        taskToCallApi.Start();
-                        Logger.Info("[LocalApiService] Started,workThreadCount:" + workThreadCount);
-
-                    }
-                    return true;
-                }
-                catch (Exception ex)
-                {
-                    Logger.Error(ex);
-                    Stop();
-                    return false;
                 }
+                return true;
             }
-
-            public void Stop()
+            catch (Exception ex)
             {
-                if (!taskToCallApi.IsRunning) return;
-
-                try
-                {
-                    taskToCallApi.Stop();
-                }
-                catch (Exception ex)
-                {
-                    Logger.Error(ex);
-                }
-                Logger.Info("[LocalApiService] Stoped");
-            }
-
-
-            #region TaskToCallApi
-
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            void Processor(RequestInfo requestInfo)
-            {             
-
-                try
-                {
-                    CommunicationManageServer.CurConn = requestInfo.conn;
-
-                    //处理请求
-                    ApiMessage apiReply = callLocalApi(requestInfo.apiRequest);
-
-                    //调用请求回调
-                    requestInfo.callback(requestInfo.sender, apiReply);
-                }
-                catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
-                {
-                    Logger.Error(ex);
-                }
+                Logger.Error(ex);
+                Consumer_Stop();
+                return false;
             }
-
- 
-            #endregion
         }
-        #endregion
 
-
-
-        #region Worker_TimeLimit
-        class Worker_TimeLimit : IWorker
+        void Consumer_Stop()
         {
+            if (!workThread.IsRunning) return;
 
-            BlockingCollection<RequestInfo> requestQueue = new BlockingCollection<RequestInfo>();
-
-
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            public void Publish(RequestInfo requestInfo)
+            try
             {
-                requestQueue.Add(requestInfo);
+                workThread.Stop();
             }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+            Logger.Info("[LocalApiService] Stoped");
+        }
 
 
- 
-            LongTaskHelp_TimeLimit taskToCallApi = new LongTaskHelp_TimeLimit();
-            Func<ApiMessage, ApiMessage> callLocalApi;
+        #region event
 
-            /// <summary>
-            /// 超时时间。（主动关闭超过此时间的任务,实际任务强制关闭的时间会在1倍超时时间到2倍超时时间内)。单位：ms。
-            /// 脉冲间隔。
-            /// </summary>
-            public int timeout_ms { set { taskToCallApi.timeout_ms = value; } }
-            public bool Start(int workThreadCount, Func<ApiMessage, ApiMessage> callLocalApi)
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void Consumer_Processor(RequestInfo requestInfo)
+        {
+            try
             {
-                try
-                {
-                    this.callLocalApi = callLocalApi;
-
-                    taskToCallApi.threadName = "LocalApiService";
-                    taskToCallApi.threadCount = workThreadCount;                   
+                CommunicationManageServer.CurConn = requestInfo.conn;
 
-                    if (workThreadCount > 0)
-                    {
-                        taskToCallApi.Start(GetWork, DealWork, OnFinish, OnTimeout);
-                        Logger.Info("[LocalApiService] Started,workThreadCount:" + workThreadCount+ ",timeout_ms:" + taskToCallApi.timeout_ms);
-                    }
-                    return true;
-                }
-                catch (Exception ex)
-                {
-                    Logger.Error(ex);
-                    Stop();
-                    return false;
-                }
+                //处理请求
+                requestInfo.apiReply = CallLocalApi(requestInfo.apiRequest);
             }
-
-            public void Stop()
+            catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
             {
-                if (!taskToCallApi.IsRunning) return;
-
-                try
-                {
-                    taskToCallApi.Stop();
-                }
-                catch (Exception ex)
-                {
-                    Logger.Error(ex);
-                }
-                Logger.Info("[LocalApiService] Stoped");
+                Logger.Error(ex);
             }
+        }
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void Consumer_OnFinish(RequestInfo requestInfo)
+        {
+            //调用请求回调
+            requestInfo.callback(requestInfo.sender, requestInfo.apiReply);
+        }
 
-            #region TaskToCallApi
-
-            /*
-             workArg        apiRequest
-             workArg2       apiReply
-             workArg3       sender
-             workArg4       callback
-                 
-                 
-                 */
-
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void Consumer_OnTimeout(RequestInfo requestInfo)
+        {
+            ApiMessage apiReply = new ApiMessage();
+            apiReply.rpcContextData_OriData = const_ApiReply_Err_Timeout.rpcContextData_OriData;
+            apiReply.value_OriData = const_ApiReply_Err_Timeout.value_OriData;
 
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            void GetWork(LongTaskHelp_TimeLimit.Worker w)
-            {
-                //堵塞获取请求
-                var requestInfo = requestQueue.Take();
-                CommunicationManageServer.CurConn = requestInfo.conn;
+            //调用请求回调
+            requestInfo.callback(requestInfo.sender, apiReply);
+        }
 
-                w.workArg = requestInfo.apiRequest;     
-                w.workArg3 = requestInfo.sender;
-                w.workArg4 = requestInfo.callback;
-            
-            }
+        #endregion
 
+        #endregion
 
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            void DealWork(LongTaskHelp_TimeLimit.Worker w)
-            {
-                ApiMessage apiRequest = (ApiMessage)w.workArg;
 
-                //处理请求
-                w.workArg2 = callLocalApi(apiRequest);                  
-            }
 
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            void OnFinish(LongTaskHelp_TimeLimit.Worker w)
-            {
-                ApiMessage apiReply = (ApiMessage)w.workArg2;
-                object sender = w.workArg3;
-                Action<object, ApiMessage> callback = (Action<object, ApiMessage>)w.workArg4;
-                //调用请求回调
-                callback(sender, apiReply);         
-            }
 
-            [MethodImpl(MethodImplOptions.AggressiveInlining)]
-            void OnTimeout(LongTaskHelp_TimeLimit.Worker w)
-            {
-                ApiMessage apiReply =  new ApiMessage();
-                apiReply.rpcContextData_OriData = const_ApiReply_Err_Timeout.rpcContextData_OriData;
-                apiReply.value_OriData = const_ApiReply_Err_Timeout.value_OriData;
- 
-                object sender = w.workArg3;
-                Action<object, ApiMessage> callback = (Action<object, ApiMessage>)w.workArg4;
-
-                //调用请求回调
-                callback(sender, apiReply);               
-            }  
-         
-            #endregion
-        }
-        #endregion
-        
-        #endregion
 
 
 

dotnet/Library/Sers/Sers.Core/Sers.Core/Sers.Core.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/ConsumerFactory.cs

@@ -1,66 +1,28 @@
-using Vit.Core.Util.ConfigurationManager;
+using Newtonsoft.Json.Linq;
 
 namespace Sers.Core.Util.Consumer
 {
     public class ConsumerFactory
     {
-
-        /// <summary>
-        /// 队列模式。可为 BlockingCollection(默认)、 ConsumerCache_BlockingCollection(高性能)
-        /// </summary>
-        public static string ConsumerMode = ConfigurationManager.Instance.GetByPath<string>("Vit.ConsumerMode");
-
-        public static IConsumer<T> CreateConsumer<T>() 
+        public static IConsumer<T> CreateConsumer<T>(JObject config) 
         {
+            if (config == null) config = new JObject();
 
-            //return new ConsumerCache<T, Consumer_BlockingCollection<T>>();
             IConsumer<T> consumer;
-            switch (ConsumerMode)
-            {
-                //case "ActionBlock":
-                //    consumer = new Consumer_ActionBlock<T>();  // 16 16 700万     24 24 900-1000万
-                //    break;
-                //case "BufferBlock":
-                //    consumer = new Consumer_BufferBlock<T>();   //2 36 800-1000万
-                //    break;
+            switch (config["mode"].ToString())
+            {               
                 case "BlockingCollection":
-                    consumer = new Consumer_BlockingCollection<T>();  //16 16 440万          2  2  800万
+                    consumer = new LongTask<T>();  //16 16 440万          2  2  800万
                     break;
-
-
-                //case "Disruptor":
-                //    consumer = new Consumer_Disruptor<T>(); // 16 16 800万
-                //    break;
-                //case "WorkerPool":
-                //    consumer = new Consumer_WorkerPool<T>(); //16 16 800-900万
-                //    break;
-
-                //case "ConsumerCache_ActionBlock":
-                //    consumer = new ConsumerCache<T, Consumer_ActionBlock<T>>(); // 16 16  4000-4200万
-                //    break;
-                //case "ConsumerCache_BufferBlock":
-                //    consumer = new ConsumerCache<T, Consumer_BufferBlock<T>>(); // 16 16  1500-1600万
-                //    break;
                 case "ConsumerCache_BlockingCollection":
-                    consumer = new ConsumerCache<T, Consumer_BlockingCollection<T>>(); //16 16 4200-4500万
+                    consumer = new ConsumerCascade<T, LongTask<T>>(); //16 16 4200-4500万
                     break;
 
-
-
-                //case "ConsumerCache_WorkerPool":
-                //    consumer = new ConsumerCache<T, Consumer_WorkerPool<T>>(); //750万  异常
-                //    break;
-                //case "WorkerPoolCache":
-                //    consumer = new Consumer_WorkerPoolCache<T>();        //940万 异常
-                //    break;
-                //case "WorkerPoolCascade":
-                //    consumer = new Consumer_WorkerPoolCascade<T>();   //1400万 异常
-                //    break;
-
                 default:
-                    consumer = new Consumer_BlockingCollection<T>();
+                    consumer = new LongTask<T>();
                     break;
             }
+            consumer.Init(config);
 
             return consumer;
         }

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/IConsumer.cs

@@ -1,4 +1,5 @@
-using System;
+using Newtonsoft.Json.Linq;
+using System;
 
 namespace Sers.Core.Util.Consumer
 {
@@ -13,6 +14,10 @@ namespace Sers.Core.Util.Consumer
 
         Action<T> processor { get; set; }
 
+        Action<T> OnFinish { get; set; }
+        Action<T> OnTimeout { get; set; }
+
+        void Init(JObject config);
 
 
         void Publish(T t);

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/Mode/AsyncTask.cs

@@ -0,0 +1,77 @@
+using Newtonsoft.Json.Linq;
+using System;
+using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Vit.Core.Util.Threading;
+
+namespace Sers.Core.Util.Consumer.Mode
+{
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    public class Async<T>: IConsumer<T>
+    {
+
+        BlockingCollection<T> queue = new BlockingCollection<T>();
+ 
+
+
+        public string name { get; set; }
+
+
+        public int workThreadCount { get; set; }
+
+
+        /// <summary>
+        ///  超时时间。脉冲间隔。（主动关闭超过此时间的任务,实际任务强制关闭的时间会在1倍超时时间到2倍超时时间内)。单位：ms。(默认300000)
+        /// </summary>
+        public int timeout_ms { get; set; }
+
+        public Action<T> processor { get; set; }
+        public Action<T> OnFinish { get; set; }
+        public Action<T> OnTimeout { get; set; }
+
+
+
+        public bool IsRunning { get; set; }
+
+        public void Init(JObject config)
+        {
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public void Publish(T t)
+        {        
+            Task.Run(() =>
+            {
+                try
+                {
+                    processor(t);
+                }
+                catch (Exception ex)
+                {
+                    //throw;
+                    OnTimeout?.Invoke(t);
+                    return;
+                }
+
+                OnFinish?.Invoke(t);
+
+            }, new CancellationTokenSource(timeout_ms).Token);
+        }
+
+
+        public void Start()
+        {             
+        }
+
+        public void Stop() 
+        {   
+        }
+         
+
+    }
+}

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/ConsumerCache.cs → dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/Mode/ConsumerCascade.cs

@@ -1,4 +1,5 @@
-using System;
+using Newtonsoft.Json.Linq;
+using System;
 using System.Linq;
 using System.Runtime.CompilerServices;
 using System.Threading;
@@ -10,7 +11,7 @@ namespace Sers.Core.Util.Consumer
     /// </summary>
     /// <typeparam name="T"></typeparam>
     /// <typeparam name="Consumer"></typeparam>
-    public class ConsumerCache<T,Consumer> : IConsumer<T>
+    public class ConsumerCascade<T,Consumer> : IConsumer<T>
         where Consumer:IConsumer<T>,new()
     {
 
@@ -18,7 +19,10 @@ namespace Sers.Core.Util.Consumer
 
         public string name { get; set; }
 
+
         public Action<T> processor { get; set; }
+        public Action<T> OnFinish { get; set; }
+        public Action<T> OnTimeout { get; set; }
 
 
         int curRootIndex;
@@ -28,6 +32,11 @@ namespace Sers.Core.Util.Consumer
 
 
         public bool IsRunning { get; private set; } = false;
+
+        public void Init(JObject config)
+        {
+        }
+
         public void Start()
         {
             if (IsRunning) return;

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/Consumer_BlockingCollection.cs → dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/Mode/LongTask.cs

@@ -1,4 +1,5 @@
-using System;
+using Newtonsoft.Json.Linq;
+using System;
 using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
 using System.Threading;
@@ -11,14 +12,17 @@ namespace Sers.Core.Util.Consumer
     /// qps : 260万   producer:16    consumer:16
     /// </summary>
     /// <typeparam name="T"></typeparam>
-    public class Consumer_BlockingCollection<T>: IConsumer<T>
+    public class LongTask<T>: IConsumer<T>
     {
 
         public int workThreadCount { get; set; } = 16;
 
         public string name { get; set; }
 
+
         public Action<T> processor { get; set; }
+        public Action<T> OnFinish { get; set; }
+        public Action<T> OnTimeout { get; set; }
 
 
         BlockingCollection<T> queue = new BlockingCollection<T>();
@@ -26,6 +30,12 @@ namespace Sers.Core.Util.Consumer
 
         public bool IsRunning { get => task.IsRunning; }
 
+
+        public void Init(JObject config)
+        {
+        }
+
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public void Publish(T t) 
         {
@@ -59,8 +69,9 @@ namespace Sers.Core.Util.Consumer
                     #region Process                        
                     while (true)
                     {
-                        //var msgFrame = queue.Take();
-                        processor(queue.Take());
+                        var msgFrame = queue.Take();
+                        processor(msgFrame);
+                        OnFinish?.Invoke(msgFrame);
                     }
                     #endregion
                 }

dotnet/Library/Sers/Sers.Core/Sers.Core/Util/Consumer/Mode/LongTask_TimeLimit.cs

@@ -0,0 +1,100 @@
+using Newtonsoft.Json.Linq;
+using System;
+using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+using Vit.Core.Util.Threading;
+
+namespace Sers.Core.Util.Consumer.Mode
+{
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    public class LongTask_TimeLimit<T>: IConsumer<T>
+    {
+        BlockingCollection<T> queue = new BlockingCollection<T>();
+        LongTaskHelp_TimeLimit task = new LongTaskHelp_TimeLimit();
+
+
+        public string name { get => task.threadName; set => task.threadName = value; }
+
+
+        public int workThreadCount { get => task.threadCount; set => task.threadCount = value; }
+
+
+        /// <summary>
+        ///  超时时间。脉冲间隔。（主动关闭超过此时间的任务,实际任务强制关闭的时间会在1倍超时时间到2倍超时时间内)。单位：ms。(默认300000)
+        /// </summary>
+        public int timeout_ms { get => task.timeout_ms; set => task.timeout_ms = value; }
+
+        public Action<T> processor { get; set; }
+        public Action<T> OnFinish { get; set; }
+        public Action<T> OnTimeout { get; set; }
+
+
+
+        public bool IsRunning { get => task.IsRunning; }
+
+        public void Init(JObject config)
+        {
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public void Publish(T t) 
+        {
+            queue.Add(t);
+        }
+
+
+        public void Start()
+        {
+            task.Start(_GetWork, _DealWork, _OnFinish, _OnTimeout);
+        }
+
+        public void Stop() 
+        {
+            task.Stop();
+        }
+
+
+        
+
+        #region TaskToCallApi
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void _GetWork(LongTaskHelp_TimeLimit.Worker w)
+        {
+            //堵塞获取请求
+            w.workArg = queue.Take();
+        }
+
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void _DealWork(LongTaskHelp_TimeLimit.Worker w)
+        {
+            var msg = (T)w.workArg;
+
+            //处理请求
+            processor(msg);
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void _OnFinish(LongTaskHelp_TimeLimit.Worker w)
+        {
+            var msg = (T)w.workArg;
+             
+            OnFinish?.Invoke(msg);
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        void _OnTimeout(LongTaskHelp_TimeLimit.Worker w)
+        {
+            var msg = (T)w.workArg;
+
+            OnTimeout?.Invoke(msg);
+        }
+
+        #endregion
+
+    }
+}

dotnet/Library/Sers/Sers.Core/Test/Sers.Core.Util.Consumer.Test/ProgramQps.cs

@@ -80,7 +80,7 @@ namespace CLClient
                     consumer = new Consumer_BufferBlock<Product>();   //2 36 800-1000万
                     break;
                 case "BlockingCollection":
-                    consumer = new Consumer_BlockingCollection<Product>();  //16 16 440万          2  2  800万
+                    consumer = new LongTask<Product>();  //16 16 440万          2  2  800万
                     break;
               
   
@@ -92,13 +92,13 @@ namespace CLClient
                 //    break;
 
                 case "ConsumerCache_ActionBlock":
-                    consumer = new ConsumerCache<Product, Consumer_ActionBlock<Product>>(); // 16 16  4000-4200万
+                    consumer = new ConsumerCascade<Product, Consumer_ActionBlock<Product>>(); // 16 16  4000-4200万
                     break;
                 case "ConsumerCache_BufferBlock":
-                    consumer = new ConsumerCache<Product, Consumer_BufferBlock<Product>>(); // 16 16  1500-1600万
+                    consumer = new ConsumerCascade<Product, Consumer_BufferBlock<Product>>(); // 16 16  1500-1600万
                     break;
                 case "ConsumerCache_BlockingCollection":
-                    consumer = new ConsumerCache<Product, Consumer_BlockingCollection<Product>>(); //16 16 4200-4500万
+                    consumer = new ConsumerCascade<Product, LongTask<Product>>(); //16 16 4200-4500万
                     break;
 
 

dotnet/Library/Sers/Sers.Core/Test/Sers.Core.Util.Consumer.Test/README.md

@@ -12,4 +12,60 @@ disruptor = new Disruptor<Entry>(() => new Entry(), 2 << 18, TaskScheduler.Defau
  
 Worker_BlockingCollection  //qps 260Íò
 
- 
+ 
+---------------------------------------------------------------------------------
+
+        public static IConsumer<T> CreateConsumer<T>() 
+        {
+
+            //return new ConsumerCache<T, Consumer_BlockingCollection<T>>();
+            IConsumer<T> consumer;
+            switch (ConsumerMode)
+            {
+                //case "ActionBlock":
+                //    consumer = new Consumer_ActionBlock<T>();  // 16 16 700Íò     24 24 900-1000Íò
+                //    break;
+                //case "BufferBlock":
+                //    consumer = new Consumer_BufferBlock<T>();   //2 36 800-1000Íò
+                //    break;
+                case "BlockingCollection":
+                    consumer = new Consumer_BlockingCollection<T>();  //16 16 440Íò          2  2  800Íò
+                    break;
+
+
+                //case "Disruptor":
+                //    consumer = new Consumer_Disruptor<T>(); // 16 16 800Íò
+                //    break;
+                //case "WorkerPool":
+                //    consumer = new Consumer_WorkerPool<T>(); //16 16 800-900Íò
+                //    break;
+
+                //case "ConsumerCache_ActionBlock":
+                //    consumer = new ConsumerCache<T, Consumer_ActionBlock<T>>(); // 16 16  4000-4200Íò
+                //    break;
+                //case "ConsumerCache_BufferBlock":
+                //    consumer = new ConsumerCache<T, Consumer_BufferBlock<T>>(); // 16 16  1500-1600Íò
+                //    break;
+                case "ConsumerCache_BlockingCollection":
+                    consumer = new ConsumerCache<T, Consumer_BlockingCollection<T>>(); //16 16 4200-4500Íò
+                    break;
+
+
+
+                //case "ConsumerCache_WorkerPool":
+                //    consumer = new ConsumerCache<T, Consumer_WorkerPool<T>>(); //750Íò  Òì³£
+                //    break;
+                //case "WorkerPoolCache":
+                //    consumer = new Consumer_WorkerPoolCache<T>();        //940Íò Òì³£
+                //    break;
+                //case "WorkerPoolCascade":
+                //    consumer = new Consumer_WorkerPoolCascade<T>();   //1400Íò Òì³£
+                //    break;
+
+                default:
+                    consumer = new Consumer_BlockingCollection<T>();
+                    break;
+            }
+
+            return consumer;
+        }

dotnet/Library/Sers/Sers.Gateway/Sers.Gateway/Sers.Gateway.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.Hardware/Sers.Hardware/Sers.Hardware.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Extensions/IWebHostBuilder_UseSerslot_Extensions.cs

@@ -1,6 +1,8 @@
-
-using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Hosting.Server;
 using Newtonsoft.Json.Linq;
+using Sers.Serslot;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace Vit.Extensions
 {
@@ -27,16 +29,18 @@ namespace Vit.Extensions
         /// <param name="hostBuilder"></param>
         /// <returns></returns>
         public static IWebHostBuilder UseSerslot(this IWebHostBuilder hostBuilder)
-        {           
+        {
+            var server = new SerslotServer();
+            server.InitPairingToken(hostBuilder);
 
-            if ("BackgroundTask" == Vit.Core.Util.ConfigurationManager.ConfigurationManager.Instance.GetStringByPath("serslot.Mode"))
-            {
-                return hostBuilder.UseSerslot_BackgroundTask();
-            }
-            else
+            return hostBuilder.ConfigureServices(services =>
             {
-                return hostBuilder.UseSerslot_Async();
-            }
+                services.AddSingleton<IServer>((serviceProvider) =>
+                {
+                    server.serviceProvider = serviceProvider;
+                    return server;
+                });
+            });
         }
 
 

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/BackgroundTask/LocalApiNode.cs → dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/LocalApiNode.cs

@@ -4,7 +4,7 @@ using Newtonsoft.Json;
 using Sers.Core.Module.Api.ApiDesc;
 using Sers.Core.Module.Api.LocalApi;
 
-namespace Sers.Serslot.Mode.BackgroundTask
+namespace Sers.Serslot
 {
     [JsonObject(MemberSerialization.OptIn)]
     public class LocalApiNode : IApiNode

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/Async/Extensions/IWebHostBuilder_UseSerslot_Extensions.cs

@@ -1,34 +0,0 @@
-
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.AspNetCore.Hosting.Server;
-using Microsoft.Extensions.DependencyInjection;
-using Sers.Core.Module.Api.LocalApi;
-using Sers.Serslot.Mode.Async;
-
-namespace Vit.Extensions
-{
-    static partial class IWebHostBuilder_UseSerslot_Extensions
-    {
-        internal static IWebHostBuilder UseSerslot_Async(this IWebHostBuilder hostBuilder)
-        {
-            var server = new SerslotServer();
-            server.InitPairingToken(hostBuilder);
-
-            LocalApiServiceFactory.CreateLocalApiService = () => new Sers.Serslot.Mode.Async.LocalApiService(server);
-
-
-            return hostBuilder.ConfigureServices(services =>
-            {
-                services.AddSingleton<IServer>((serviceProvider)=> {
-                    server.serviceProvider = serviceProvider;
-                    return server;
-                });
-            });
-
-
-        }
-
-
-
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/Async/LocalApiService.cs

@@ -1,134 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Threading;
-using Vit.Extensions;
-using Vit.Core.Module.Log;
-using Sers.Core.Module.Rpc;
-using Sers.Core.Module.Message;
-using Vit.Core.Util.ComponentModel.SsError;
-using Sers.Core.Module.Api.LocalApi.Event;
-using System.Runtime.CompilerServices;
-using Sers.Core.Module.Api.LocalApi;
-using System.Threading.Tasks;
-
-namespace Sers.Serslot.Mode.Async
-{
-    public class LocalApiService: ILocalApiService
-    {
-
-        public LocalApiService(SerslotServer serslotServer) 
-        {
-            this.serslotServer = serslotServer;
-        }
-
-
-
-
-        public void Init()
-        {
-            LocalApiEventMng.Instance.UseApiTraceLog();
-        }
-
-        /// <summary>
-        /// 映射  route -> LocalApiNode
-        /// </summary>
-        protected readonly ApiNodeMng apiNodeMng = new ApiNodeMng();
-
-        public IEnumerable<IApiNode> apiNodes => apiNodeMng.apiNodes;
-
-
-        public ApiNodeMng ApiNodeMng => apiNodeMng;
-
-
-        SerslotServer serslotServer;
-
-        #region CallLocalApi
- 
-        [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        async Task  CallLocalApiAsync(ApiMessage apiRequest, Object sender, Action<object, ApiMessage> callback)
-        {           
-            using (var rpcContext = new RpcContext())
-            using (var localApiEvent = LocalApiEventMng.Instance.CreateApiEvent())
-            {
-                try
-                {
-                    //(x.1) init rpcContext
-                    rpcContext.apiRequestMessage = apiRequest;
-                    rpcContext.apiReplyMessage = new ApiMessage();
-
-                    var rpcData = RpcContextData.FromBytes(apiRequest.rpcContextData_OriData);
-                    rpcContext.rpcData = rpcData;
-
-                    //(x.2) BeforeCallApi
-                    localApiEvent?.BeforeCallApi(rpcData, apiRequest);
-
-
-                    // (x.3)ProcessRequestByRpcAsync
-                    await serslotServer.ProcessRequestByRpcAsync(rpcContext);          
-
-                }
-                catch (Exception ex) when ((ex.GetBaseException() is ThreadInterruptedException))
-                {
-                    //处理超时
-                    rpcContext.apiReplyMessage.rpcContextData_OriData = const_ApiReply_Err_Timeout.rpcContextData_OriData;
-                    rpcContext.apiReplyMessage.value_OriData = const_ApiReply_Err_Timeout.value_OriData;
-                    //throw;
-                }
-                catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
-                {
-                    ex = ex.GetBaseException();
-                    Logger.Error(ex);
-                    SsError error = ex;
-                    rpcContext.apiReplyMessage.InitAsApiReplyMessageByError(error);
-                }
-
-                callback(sender, rpcContext.apiReplyMessage);
-            }
-        }
-        #endregion
-
-
-        static readonly ApiMessage const_ApiReply_Err_Timeout = new ApiMessage().InitAsApiReplyMessageByError(SsError.Err_HandleTimeout);
-
-
-        #region 后台服务
- 
-
-
-
-        [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public void CallApiAsync(Object sender, ApiMessage apiRequest, Action<object, ApiMessage> callback)
-        {
-            CallLocalApiAsync(apiRequest, sender, callback);
-        }
-
-
-
-
-
-        public bool Start()
-        {
-            return true;
-        }
-
-
-        public void Stop()
-        {      
-        }
-     
-
-
-       
-
-        #endregion
-
-
-
-    
-
-
-
-
-
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/Async/SerslotServer.LoadSerslotApi.cs

@@ -1,45 +0,0 @@
-using System;
-using System.Reflection;
-using Sers.Core.Module.Api.ApiDesc;
-using Sers.Core.Module.Api.LocalApi;
-using Vit.Extensions;
-
-
-namespace Sers.Serslot.Mode.Async
-{
-    public partial class SerslotServer 
-    {
-
-        /// <summary>
-        /// µ÷ÓÃSerslot¼ÓÔØÆ÷¼ÓÔØapi
-        /// </summary>
-        /// <param name="data"></param>
-        /// <param name="assembly"></param>
-        public void LoadSerslotApi(ILocalApiService data, Assembly assembly)
-        {
-            if (null == data)
-            {
-                return;
-            }
-
-            #region (x.1) api from host
-            var config = new SersLoader.ApiLoaderConfig { assembly = assembly };
-
-            Func<SsApiDesc, IApiNode> CreateApiNode =
-                (apiDesc) =>
-                {
-                    return new ApiNode_Original(apiDesc: apiDesc);
-                };
-
-            var apiLoader = new Sers.Serslot.ApiLoader(CreateApiNode);
-
-            data.ApiNodeMng.AddApiNode(apiLoader.LoadApi(config));
-            #endregion
-
-
-            //(x.2)load api from appsettings.json::serslot.extApi
-            data.LoadSerslotExtApi();
-
-        }
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/Async/SerslotServer.ProcessRequestByRpcAsync.cs

@@ -1,67 +0,0 @@
-using Microsoft.AspNetCore.Http.Features;
-using Sers.Core.Module.Rpc;
-using System.IO;
-using System.Runtime.CompilerServices;
-using System.Threading.Tasks;
-using Vit.Extensions;
-
-namespace Sers.Serslot.Mode.Async
-{
-    public partial class SerslotServer
-    {
-        [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public async Task ProcessRequestByRpcAsync( RpcContext rpcContext)
-        {
-         
-            HttpRequestFeature requestFeature = null;
-            IHttpResponseFeature responseFeature = null;
-            try
-            {
-                //(x.1) build requestFeature
-                requestFeature = rpcContext.BuildHttpRequestFeature();
-
-                //(x.2) ProcessRequest
-                responseFeature = await ProcessRequestAsync(requestFeature);         
-
-
-                #region (x.3)build reply
-                var rpcReply = new RpcContextData();
-
-                //(x.x.1)StatusCode
-                rpcReply.http.statusCode = responseFeature.StatusCode;
-
-                #region (x.x.2)http_header
-                var replyHeader = responseFeature.Headers;
-                if (replyHeader != null)
-                {
-                    var headers = rpcReply.http.Headers();
-                    foreach (var item in replyHeader)
-                    {
-                        headers[item.Key] = item.Value.ToString();
-                    }
-                }
-                #endregion
-
-                //(x.x.3) reply rpcContextData
-                rpcContext.apiReplyMessage.rpcContextData_OriData = rpcReply.ToBytes().BytesToArraySegmentByte();
-
-
-                #region (x.x.4) reply Body
-                var body = (responseFeature.Body as MemoryStream).ToArray();
-                rpcContext.apiReplyMessage.value_OriData = body.BytesToArraySegmentByte();
-                #endregion
-
-                #endregion
-            }
-            finally
-            {
-                requestFeature?.Body?.Dispose();
-                responseFeature?.Body?.Dispose();
-            }
-
-        }
-
-
-
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/Async/SerslotServer.cs

@@ -1,164 +0,0 @@
-using System;
-using System.Reflection;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.AspNetCore.Hosting.Server;
-using Microsoft.AspNetCore.Http.Features;
-using Vit.Core.Module.Log;
-using Vit.Extensions;
-
-
-namespace Sers.Serslot.Mode.Async
-{
-    public partial class SerslotServer : IServer
-    {
-        /// <summary>
-        /// 
-        /// </summary>
-        public IServiceProvider serviceProvider { get; set; }
-
-        #region PairingToken       
-        string pairingToken;
-        public void InitPairingToken(IWebHostBuilder hostBuilder)
-        {
-            //search "MS-ASPNETCORE-TOKEN" to know why
-            string PairingToken = "TOKEN";
-            pairingToken = hostBuilder.GetSetting(PairingToken) ?? Environment.GetEnvironmentVariable($"ASPNETCORE_{PairingToken}");
-
-        }
-        #endregion
-
-
-
-        #region ProcessRequest       
-
-        Func<FeatureCollection,Task> OnProcessRequest;
-
-        [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.AggressiveInlining)]
-        public async Task<IHttpResponseFeature> ProcessRequestAsync(HttpRequestFeature requestFeature)
-        {
-            requestFeature.InitForSerslot(pairingToken, out var _responseFeature,out var features);
-
-            await OnProcessRequest(features);
-
-            return _responseFeature;
-        }
-
-     
-
-        #endregion
-
-
-
-        public IFeatureCollection Features { get; } = new FeatureCollection();
-
-
-        public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken)
-        {
-            try
-            {
-                #region (x.1) build OnProcessRequest               
-                OnProcessRequest = async (features) =>
-                {
-
-                    Exception _applicationException = null;
-
-                    var httpContext = application.CreateContext(features);
-                    try
-                    {
-
-                        //var httpContext_ = httpContext.GetProperty<object>("HttpContext");
-                        //if (httpContext_ is Microsoft.AspNetCore.Http.HttpContext defaultHttpContext)
-                        //{
-                        //    //if (defaultHttpContext.Response.Body == null)                            
-                        //    defaultHttpContext.Response.Body = features.Get<IHttpResponseFeature>().Body;                           
-                        //}
-
-
-                        // Run the application code for this request
-                        // application.ProcessRequestAsync(httpContext).GetAwaiter().GetResult();
-                       
-                        await application.ProcessRequestAsync(httpContext);
-
-
-                        //var _responseFeature = features.Get<IHttpResponseFeature>() as SerslotResponseFeature;
-                        //if (_responseFeature != null)
-                        //{                           
-                        //    _responseFeature.FireOnStarting();
-                        //    _responseFeature.FireOnCompleted();
-                        //}
-                    }
-                    catch (Exception ex)
-                    {
-                        _applicationException = ex;
-                        Logger.Error(ex);
-                    }
-                    application.DisposeContext(httpContext, _applicationException);
-                };
-                #endregion
-
-
-                #region (x.2) start ServiceStation                
-
-                #region (x.x.1) Init
-                ServiceStation.ServiceStation.Init();
-                Sers.Core.Module.App.SersApplication.onStop += () =>
-                {
-                    if (serviceProvider.GetService(typeof(IApplicationLifetime)) is IApplicationLifetime lifetime)
-                    {
-                        lifetime.StopApplication();
-                    }
-                };
-                #endregion
-
-
-                Logger.Info("[Serslot] Mode: Async");
-
-                #region (x.x.2)¼ÓÔØapi
-
-                ServiceStation.ServiceStation.Instance.LoadApi();
-
-                LoadSerslotApi(ServiceStation.ServiceStation.Instance.localApiService,Assembly.GetEntryAssembly());
-
-                #endregion
-
-                //(x.x.3)Start ServiceStation
-                if (!ServiceStation.ServiceStation.Start())
-                {
-                    Dispose();
-                }
-                #endregion
-
-            }
-            catch
-            {
-                Dispose();
-                throw;
-            }
-        }
-
-        // Graceful shutdown if possible
-        public async Task StopAsync(CancellationToken cancellationToken)
-        {
-
-            try
-            {
-                ServiceStation.ServiceStation.Stop();
-            }
-            catch (Exception ex)
-            {
-                Logger.Error(ex);
-            }
-        }
-
-        // Ungraceful shutdown
-        public void Dispose()
-        {
-            var cancelledTokenSource = new CancellationTokenSource();
-            cancelledTokenSource.Cancel();
-            StopAsync(cancelledTokenSource.Token).GetAwaiter().GetResult();
-        }
-
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/BackgroundTask/Extensions/IWebHostBuilder_UseSerslot_Extensions.cs

@@ -1,28 +0,0 @@
-
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.AspNetCore.Hosting.Server;
-using Microsoft.Extensions.DependencyInjection;
-using Sers.Serslot.Mode.BackgroundTask;
-
-namespace Vit.Extensions
-{
-    static partial class IWebHostBuilder_UseSerslot_Extensions
-    {
-        internal static IWebHostBuilder UseSerslot_BackgroundTask(this IWebHostBuilder hostBuilder)
-        {
-            var server = new SerslotServer();
-            server.InitPairingToken(hostBuilder);
-
-            return hostBuilder.ConfigureServices(services =>
-            {
-                services.AddSingleton<IServer>((serviceProvider) =>
-                {
-                    server.serviceProvider = serviceProvider;
-                    return server;
-                });
-            });
-
-        }
-
-    }
-}

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Sers.Serslot.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/BackgroundTask/SerslotServer.LoadSerslotApi.cs → dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/SerslotServer.LoadSerslotApi.cs

@@ -5,7 +5,7 @@ using Sers.Core.Module.Api.LocalApi;
 using Vit.Extensions;
 
 
-namespace Sers.Serslot.Mode.BackgroundTask
+namespace Sers.Serslot
 {
     public partial class SerslotServer 
     {

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/BackgroundTask/SerslotServer.ProcessRequestByRpcAsync.cs → dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/SerslotServer.ProcessRequestByRpcAsync.cs

@@ -5,15 +5,11 @@ using System.IO;
 using System.Runtime.CompilerServices;
 using Vit.Extensions;
 
-namespace Sers.Serslot.Mode.BackgroundTask
+namespace Sers.Serslot
 {
     public partial class SerslotServer
     {
 
-
-
-
-
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public byte[] ProcessRequestByRpc(ArraySegment<byte> arg_OriData)
         {

dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/Mode/BackgroundTask/SerslotServer.cs → dotnet/Library/Sers/Sers.Serslot/Sers.Serslot/SerslotServer.cs

@@ -8,7 +8,7 @@ using Microsoft.AspNetCore.Http.Features;
 using Vit.Core.Module.Log;
 using Vit.Extensions;
 
-namespace Sers.Serslot.Mode.BackgroundTask
+namespace Sers.Serslot
 {
     public partial class SerslotServer : IServer
     {

dotnet/Library/Sers/Sers.ServiceStation/Sers.ServiceStation/Sers.ServiceStation.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/Library/Vit/Vit.Core/Vit.Core/Util/Threading/LongTaskHelp_TimeLimit.cs

@@ -69,8 +69,7 @@ namespace Vit.Core.Util.Threading
 
 
         /// <summary>
-        /// 超时时间，（主动关闭超过此时间的任务,实际任务强制关闭的时间会在1倍超时时间到2倍超时时间内)。单位：ms。(默认300000)
-        /// 脉冲间隔。
+        /// 超时时间。脉冲间隔。（主动关闭超过此时间的任务,实际任务强制关闭的时间会在1倍超时时间到2倍超时时间内)。单位：ms。(默认300000)
         /// </summary>
         public int timeout_ms = 300000;
 

dotnet/ServiceCenter/App.Gover.Gateway/App.Gover.Gateway.csproj

@@ -8,7 +8,7 @@
 	<PropertyGroup>
 		<OutputType>Exe</OutputType>
 		<TargetFramework>netcoreapp2.1</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 		<PackageProjectUrl>https://github.com/serset/Sers</PackageProjectUrl>
 	</PropertyGroup>
 

dotnet/ServiceCenter/App.ServiceCenter.NetFx/App.ServiceCenter.NetFx/packages.config

@@ -4,14 +4,14 @@
   <package id="Microsoft.DotNet.PlatformAbstractions" version="2.0.0" targetFramework="net461" />
   <package id="Microsoft.Extensions.DependencyModel" version="2.0.0" targetFramework="net461" />
   <package id="Newtonsoft.Json" version="12.0.3" targetFramework="net461" />
-  <package id="Sers.CL.Ipc.NamedPipe" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.CL.Socket.Iocp" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.CL.Socket.ThreadWait" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.CL.WebSocket" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.Core" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.Gover" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.Hardware" version="2.1.9" targetFramework="net461" />
-  <package id="Sers.ServiceCenter" version="2.1.9" targetFramework="net461" />
+  <package id="Sers.CL.Ipc.NamedPipe" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.CL.Socket.Iocp" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.CL.Socket.ThreadWait" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.CL.WebSocket" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.Core" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.Gover" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.Hardware" version="2.1.10-temp" targetFramework="net461" />
+  <package id="Sers.ServiceCenter" version="2.1.10-temp" targetFramework="net461" />
   <package id="System.Buffers" version="4.5.1" targetFramework="net461" />
-  <package id="Vit.Core" version="2.1.9" targetFramework="net461" />
+  <package id="Vit.Core" version="2.1.10-temp" targetFramework="net461" />
 </packages>

dotnet/ServiceCenter/App.ServiceCenter/App.ServiceCenter.csproj

@@ -8,7 +8,7 @@
 	<PropertyGroup>
 		<OutputType>Exe</OutputType>
 		<TargetFramework>netcoreapp2.1</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 		<PackageProjectUrl>https://github.com/serset/Sers</PackageProjectUrl>
 	</PropertyGroup>
 
@@ -36,7 +36,6 @@
 	</ItemGroup>
 
 
-
 	<ItemGroup>
 		<Content Include="..\App.Gover.Gateway\wwwroot\**">
 			<Link>wwwroot\%(RecursiveDir)%(Filename)%(Extension)</Link>
@@ -44,4 +43,5 @@
 		</Content>
 	</ItemGroup>
 
+
 </Project>

dotnet/ServiceCenter/App.ServiceCenter/appsettings.json

@@ -5,11 +5,17 @@
       /* 共用配置，子项共用，覆写子项未指定的配置 */
       "Config": {
 
-        /* 请求超时时间（单位ms，默认300000） */
-        "requestTimeoutMs": 300000,
+        "workThread": {
+
+          "mode": "",
+          /* 后台处理消息的线程个数（单位个，默认2） */
+          "threadCount": 2,
+          /* 请求超时时间（单位ms，默认300000） */
+          "timeoutMs": 300000
+
+
+        },
 
-        /* 后台处理消息的线程个数（单位个，默认2） */
-        "workThreadCount": 2,
 
         //HeartBeat
         /* 心跳检测时间间隔（单位ms，默认10000,若指定为0则不进行心跳检测） */

dotnet/ServiceCenter/Sers.ServiceCenter/Apm/Sers.Gover.Apm.Txt/Sers.Gover.Apm.Txt.csproj

@@ -2,7 +2,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/ServiceCenter/Sers.ServiceCenter/Apm/Sers.Gover.Apm.Zipkin/Sers.Gover.Apm.Zipkin.csproj

@@ -2,7 +2,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/ServiceCenter/Sers.ServiceCenter/Sers.Gover/Sers.Gover.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/ServiceCenter/Sers.ServiceCenter/Sers.ServiceCenter/Sers.ServiceCenter.csproj

@@ -6,7 +6,7 @@
 
 	<PropertyGroup>
 		<TargetFramework>netstandard2.0</TargetFramework>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 	</PropertyGroup>
 
 	<PropertyGroup>

dotnet/ServiceStation/Demo/SersLoader/Did.SersLoader.Demo/Did.SersLoader.Demo.csproj

@@ -8,7 +8,7 @@
   <PropertyGroup>
     <OutputType>Exe</OutputType>
     <TargetFramework>netcoreapp2.1</TargetFramework>
-    <Version>2.1.9</Version>
+    <Version>2.1.10-temp</Version>
     <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
     <Description>https://github.com/serset/Sers</Description>
   </PropertyGroup>

dotnet/ServiceStation/Demo/SersLoader/Did.SersLoader.Demo/appsettings.json

@@ -12,6 +12,11 @@
         /* 后台处理消息的线程个数（单位个，默认2） */
         "workThreadCount": 2,
 
+        /* 队列模式,可不指定。可为 BlockingCollection(默认)、 ConsumerCache_BlockingCollection(高性能) */
+        "ConsumerMode": "BlockingCollection",
+
+
+
         //HeartBeat
         /* 心跳检测时间间隔（单位ms，默认10000,若指定为0则不进行心跳检测） */
         "heartBeatIntervalMs": 10000,
@@ -356,9 +361,6 @@
       "PrintToConsole": true
     },
 
-    /* 队列模式,可不指定。可为 BlockingCollection(默认)、 ConsumerCache_BlockingCollection(高性能) */
-    "ConsumerMode": "BlockingCollection",
-
 
     /* 序列化配置，可不指定 */
     "Serialization": {

dotnet/ServiceStation/Demo/SersLoader/Did.SersLoader.HelloWorld/Did.SersLoader.HelloWorld.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <OutputType>Exe</OutputType>
     <TargetFramework>netcoreapp2.1</TargetFramework>
-    <Version>2.1.9</Version>
+    <Version>2.1.10-temp</Version>
     <Description>https://github.com/serset/Sers</Description>
   </PropertyGroup>
 

dotnet/ServiceStation/Demo/Serslot/Did.Serslot.Demo/appsettings.json

@@ -205,10 +205,6 @@
       "PrintToConsole": true
     },
 
-    /* 队列模式,可不指定。可为 BlockingCollection(默认)、 ConsumerCache_BlockingCollection(高性能) */
-    "ConsumerMode": "BlockingCollection",
-
-
     /* 序列化配置，可不指定 */
     "Serialization": {
       /* 序列化字符编码。可不指定，默认 UTF8。只可为 UTF7,UTF8,UTF32,ASCII,Unicode。 */
@@ -222,11 +218,6 @@
 
   "serslot": {
 
-    /* 处理请求转发的模式,可不指定。可为 Async(默认)、BackgroundTask */
-    /* Async 异步转发请求，没有线程数量限制。此模式下extApi的reply无效。 */
-    /* BackgroundTask 启动固定数量的线程转发请求，线程数量为Sers.LocalApiService.workThreadCount。 */
-    "//Mode": "BackgroundTask",
-
     /* 额外api。SsApiDesc格式。除了系统api外，可以指定额外api。若指定返回数据（reply）,则直接返回数据；若不指定返回数据，则由host进行处理 */
     "//extApi": [
       { //为 SsApiDesc格式

dotnet/ServiceStation/Demo/StressTest/App.Robot.Station/App.Robot.Station.csproj

@@ -9,7 +9,7 @@
 		<OutputType>Exe</OutputType>
 		<TargetFramework>netcoreapp2.1</TargetFramework>
 		<RunPostBuildEvent>Always</RunPostBuildEvent>
-		<Version>2.1.9</Version>
+		<Version>2.1.10-temp</Version>
 		<Description>https://github.com/serset/Sers</Description>
 	</PropertyGroup>