命令模式的深度解析：从标准实现到TPL Dataflow高性能架构

命令模式是对一类对象公共操作的抽象，它们具有相同的方法签名，所以具有类似的操作，可以被抽象出来，成为一个抽象的命令对象。实际操作的调用者就不是和一组对象打交道，它是需要以来这个命令对象的方法签名，并根据这个签名调用相关的方法。
以上是命令模式的大概含义，这里可以联想到事件驱动，command和handler,也可以联想到AOP的思想。联想到数据流的操作我就写了个数据流操作类库。

之前写了一些有关AOP的，但是感觉还是差点意思，补上这次的可能在项目中会弥补一些短板回来，就是灵活性。
但是该项目重点是数据流的处理，所以web端来实现只是一个例子，大量数据的处理最主要的是后台任务吧，通过接口调用只是一个实例展示。
有关数据流这块代码核心如下：

1. using System;
2. using System.Collections.Concurrent;
3. using System.Collections.Generic;
4. using System.Linq;
5. using System.Reflection;
6. using System.Threading;
7. using System.Threading.Tasks;
8. using System.Threading.Tasks.Dataflow;
9. using Microsoft.Extensions.DependencyInjection;
10. using Microsoft.Extensions.Logging;
11. using Common.Bus.Core;
12. using Common.Bus.Monitoring;
14. namespace Common.Bus.Implementations
15. {
16.     /// <summary>
17.     /// 基于TPL数据流的高性能CommandBus实现
18.     /// 支持并行处理、背压控制和监控
19.     /// </summary>
20.     public class DataflowCommandBus : ICommandBus, IDisposable
21.     {
22.         private readonly IServiceProvider _provider;
23.         private readonly ILogger<DataflowCommandBus>? _logger;
24.         private readonly ConcurrentDictionary<Type, Func<object>> _handlerCache = new();
25.         private readonly ConcurrentDictionary<Type, Func<object[]>> _behaviorsCache = new();
26.         
27.         // 数据流网络
28.         private ActionBlock<DataflowCommandRequest> _commandProcessor = null!;
29.         
30.         // 背压控制
31.         private readonly SemaphoreSlim _concurrencyLimiter;
32.         private readonly int _maxConcurrency;
33.         
34.         // 监控指标
35.         private long _processedCommands;
36.         private long _failedCommands;
37.         private long _totalProcessingTime;
39.         public DataflowCommandBus(IServiceProvider serviceProvider, ILogger<DataflowCommandBus>? logger = null,
40.             int? maxConcurrency = null)
41.         {
42.             _provider = serviceProvider;
43.             _logger = logger;
44.             _maxConcurrency = maxConcurrency ?? Environment.ProcessorCount * 2;
45.             _concurrencyLimiter = new SemaphoreSlim(_maxConcurrency, _maxConcurrency);
46.             
47.             // 创建数据流网络
48.             CreateDataflowNetwork();
49.         }
51.         private void CreateDataflowNetwork()
52.         {
53.             // 创建命令处理器
54.             _commandProcessor = new ActionBlock<DataflowCommandRequest>(
55.                 async request =>
56.                 {
57.                     try
58.                     {
59.                         await _concurrencyLimiter.WaitAsync();
60.                         var startTime = DateTime.UtcNow;
61.                         
62.                         // 执行完整的命令处理管道
63.                         var result = await ProcessCommandPipeline(request);
64.                         
65.                         var processingTime = DateTime.UtcNow - startTime;
66.                         Interlocked.Add(ref _totalProcessingTime, processingTime.Ticks);
67.                         Interlocked.Increment(ref _processedCommands);
68.                         
69.                         request.TaskCompletionSource.SetResult(result);
70.                     }
71.                     catch (Exception ex)
72.                     {
73.                         Interlocked.Increment(ref _failedCommands);
74.                         _logger?.LogError(ex, "Command processing failed for {CommandType}", request.CommandType.Name);
75.                         request.TaskCompletionSource.SetException(ex);
76.                     }
77.                     finally
78.                     {
79.                         _concurrencyLimiter.Release();
80.                     }
81.                 },
82.                 new ExecutionDataflowBlockOptions
83.                 {
84.                     MaxDegreeOfParallelism = _maxConcurrency,
85.                     BoundedCapacity = _maxConcurrency * 2
86.                 });
87.         }
89.         public async Task<TResult> SendAsync<TCommand, TResult>(TCommand command, CancellationToken ct = default)
90.             where TCommand : ICommand<TResult>
91.         {
92.             var commandType = typeof(TCommand);
93.             var requestId = Guid.NewGuid();
94.             var tcs = new TaskCompletionSource<object>();
95.             
96.             var request = new DataflowCommandRequest(requestId, commandType, typeof(TResult), command, tcs);
97.             
98.             // 发送到数据流网络
99.             if (!_commandProcessor.Post(request))
100.             {
101.                 throw new InvalidOperationException("Unable to queue command for processing - system may be overloaded");
102.             }
103.             
104.             try
105.             {
106.                 var result = await tcs.Task.WaitAsync(ct);
107.                 return (TResult)result;
108.             }
109.             catch (OperationCanceledException) when (ct.IsCancellationRequested)
110.             {
111.                 _logger?.LogWarning("Command {CommandType} was cancelled", commandType.Name);
112.                 throw;
113.             }
114.         }
116.         private async Task<object> ProcessCommandPipeline(DataflowCommandRequest request)
117.         {
118.             // 使用反射调用泛型方法
119.             var method = typeof(DataflowCommandBus).GetMethod(nameof(ProcessCommandPipelineGeneric), BindingFlags.NonPublic | BindingFlags.Instance);
120.             var genericMethod = method!.MakeGenericMethod(request.CommandType, request.ResultType);
121.             
122.             var task = (Task)genericMethod.Invoke(this, new object[] { request })!;
123.             await task;
124.             
125.             var resultProperty = task.GetType().GetProperty("Result");
126.             return resultProperty?.GetValue(task) ?? throw new InvalidOperationException("Failed to get result from task");
127.         }
129.         private async Task<TResult> ProcessCommandPipelineGeneric<TCommand, TResult>(DataflowCommandRequest request)
130.             where TCommand : ICommand<TResult>
131.         {
132.             // 获取处理器和行为的工厂函数
133.             var handlerFactory = GetCachedHandler<TCommand, TResult>(request.CommandType);
134.             var behaviorsFactory = GetCachedBehaviors<TCommand, TResult>(request.CommandType);
135.             
136.             // 创建处理器和行为的实例
137.             var handler = handlerFactory();
138.             var behaviors = behaviorsFactory();
139.             
140.             // 构建处理管道
141.             Func<Task<TResult>> pipeline = () => ExecuteHandler<TCommand, TResult>(handler, (TCommand)request.Command);
142.             
143.             // 按顺序应用管道行为
144.             foreach (var behavior in behaviors.Reverse())
145.             {
146.                 var currentBehavior = behavior;
147.                 var currentPipeline = pipeline;
148.                 pipeline = async () => (TResult)await ExecuteBehavior(currentBehavior, (TCommand)request.Command, currentPipeline);
149.             }
150.             
151.             return await pipeline();
152.         }
154.         private async Task<object> ExecuteBehavior<TCommand, TResult>(
155.             ICommandPipelineBehavior<TCommand, TResult> behavior,
156.             TCommand command,
157.             Func<Task<TResult>> next)
158.             where TCommand : ICommand<TResult>
159.         {
160.             try
161.             {
162.                 var result = await behavior.Handle(command, next, CancellationToken.None);
163.                 return result!;
164.             }
165.             catch (Exception ex)
166.             {
167.                 throw new InvalidOperationException($"Error executing behavior {behavior.GetType().Name}: {ex.Message}", ex);
168.             }
169.         }
171.         private Func<ICommandHandler<TCommand, TResult>> GetCachedHandler<TCommand, TResult>(Type commandType)
172.             where TCommand : ICommand<TResult>
173.         {
174.             return (Func<ICommandHandler<TCommand, TResult>>)_handlerCache.GetOrAdd(commandType, _ =>
175.             {
176.                 return new Func<ICommandHandler<TCommand, TResult>>(() =>
177.                 {
178.                     using var scope = _provider.CreateScope();
179.                     var handler = scope.ServiceProvider.GetService<ICommandHandler<TCommand, TResult>>();
180.                     if (handler == null)
181.                         throw new InvalidOperationException($"No handler registered for {commandType.Name}");
182.                     return handler;
183.                 });
184.             });
185.         }
187.         private Func<ICommandPipelineBehavior<TCommand, TResult>[]> GetCachedBehaviors<TCommand, TResult>(Type commandType)
188.             where TCommand : ICommand<TResult>
189.         {
190.             return (Func<ICommandPipelineBehavior<TCommand, TResult>[]>)_behaviorsCache.GetOrAdd(commandType, _ =>
191.             {
192.                 return new Func<ICommandPipelineBehavior<TCommand, TResult>[]>(() =>
193.                 {
194.                     using var scope = _provider.CreateScope();
195.                     var behaviors = scope.ServiceProvider.GetServices<ICommandPipelineBehavior<TCommand, TResult>>().ToArray();
196.                     return behaviors;
197.                 });
198.             });
199.         }
201.         private async Task<TResult> ExecuteHandler<TCommand, TResult>(ICommandHandler<TCommand, TResult> handler, TCommand command)
202.             where TCommand : ICommand<TResult>
203.         {
204.             return await handler.HandleAsync(command, CancellationToken.None);
205.         }
207.         private async Task<object> ExecuteHandler(object handler, object command)
208.         {
209.             var handlerType = handler.GetType();
210.             var handleMethod = handlerType.GetMethod("HandleAsync");
211.             
212.             if (handleMethod == null)
213.                 throw new InvalidOperationException($"Handler {handlerType.Name} does not have HandleAsync method");
215.             var task = (Task)handleMethod.Invoke(handler, new object[] { command, CancellationToken.None })!;
216.             await task;
217.             
218.             var resultProperty = task.GetType().GetProperty("Result");
219.             return resultProperty?.GetValue(task) ?? throw new InvalidOperationException("Failed to get result from task");
220.         }
222.         private Func<object> GetCachedHandler(Type commandType)
223.         {
224.             return _handlerCache.GetOrAdd(commandType, _ =>
225.             {
226.                 // 获取命令类型实现的ICommand<TResult>接口
227.                 var commandInterface = commandType.GetInterfaces()
228.                     .FirstOrDefault(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(ICommand<>));
229.                
230.                 if (commandInterface == null)
231.                     throw new InvalidOperationException($"Command type {commandType.Name} does not implement ICommand<TResult>");
232.                
233.                 var resultType = commandInterface.GetGenericArguments()[0];
234.                 var handlerType = typeof(ICommandHandler<,>).MakeGenericType(commandType, resultType);
235.                
236.                 // 返回一个工厂函数，而不是直接返回处理器实例
237.                 return new Func<object>(() =>
238.                 {
239.                     using var scope = _provider.CreateScope();
240.                     var handler = scope.ServiceProvider.GetService(handlerType);
241.                     if (handler == null)
242.                         throw new InvalidOperationException($"No handler registered for {commandType.Name}");
243.                     return handler;
244.                 });
245.             });
246.         }
248.         private Func<object[]> GetCachedBehaviors(Type commandType)
249.         {
250.             return _behaviorsCache.GetOrAdd(commandType, _ =>
251.             {
252.                 // 获取命令类型实现的ICommand<TResult>接口
253.                 var commandInterface = commandType.GetInterfaces()
254.                     .FirstOrDefault(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(ICommand<>));
255.                
256.                 if (commandInterface == null)
257.                     throw new InvalidOperationException($"Command type {commandType.Name} does not implement ICommand<TResult>");
258.                
259.                 var resultType = commandInterface.GetGenericArguments()[0];
260.                 var behaviorType = typeof(ICommandPipelineBehavior<,>).MakeGenericType(commandType, resultType);
261.                
262.                 // 返回一个工厂函数，而不是直接返回行为实例
263.                 return new Func<object[]>(() =>
264.                 {
265.                     using var scope = _provider.CreateScope();
266.                     var behaviors = scope.ServiceProvider.GetServices(behaviorType).Where(b => b != null).ToArray();
267.                     return behaviors!;
268.                 });
269.             });
270.         }
272.         // 监控和统计方法
273.         public DataflowMetrics GetMetrics()
274.         {
275.             return new DataflowMetrics
276.             {
277.                 ProcessedCommands = Interlocked.Read(ref _processedCommands),
278.                 FailedCommands = Interlocked.Read(ref _failedCommands),
279.                 TotalProcessingTime = TimeSpan.FromTicks(Interlocked.Read(ref _totalProcessingTime)),
280.                 AverageProcessingTime = _processedCommands > 0
281.                     ? TimeSpan.FromTicks(Interlocked.Read(ref _totalProcessingTime) / _processedCommands)
282.                     : TimeSpan.Zero,
283.                 AvailableConcurrency = _concurrencyLimiter.CurrentCount,
284.                 MaxConcurrency = _maxConcurrency,
285.                 InputQueueSize = _commandProcessor.InputCount
286.             };
287.         }
289.         public void ClearCache()
290.         {
291.             _handlerCache.Clear();
292.             _behaviorsCache.Clear();
293.         }
295.         public void Dispose()
296.         {
297.             _commandProcessor?.Complete();
298.             _concurrencyLimiter?.Dispose();
299.         }
300.     }
302.     // 辅助类
303.     internal class DataflowCommandRequest
304.     {
305.         public Guid Id { get; }
306.         public Type CommandType { get; }
307.         public Type ResultType { get; }
308.         public object Command { get; }
309.         public TaskCompletionSource<object> TaskCompletionSource { get; }
311.         public DataflowCommandRequest(Guid id, Type commandType, Type resultType, object command, TaskCompletionSource<object> tcs)
312.         {
313.             Id = id;
314.             CommandType = commandType;
315.             ResultType = resultType;
316.             Command = command;
317.             TaskCompletionSource = tcs;
318.         }
319.     }
321. }

复制代码

 
这里如果不是数据流方式可以使用通用模式：

1. using System;
2. using System.Collections.Concurrent;
3. using System.Collections.Generic;
4. using System.Linq;
5. using System.Text;
6. using System.Threading.Tasks;
7. using Microsoft.Extensions.DependencyInjection;
8. using Common.Bus.Core;
10. namespace Common.Bus.Implementations
11. {
12.     public class CommandBus : ICommandBus
13.     {
14.         private readonly IServiceProvider _provider;
15.         private readonly ConcurrentDictionary<Type, Func<object>> _handlerCache = new();
16.         private readonly ConcurrentDictionary<Type, Func<object[]>> _behaviorsCache = new();
17.         private readonly ConcurrentDictionary<Type, Func<object, object, CancellationToken, Task<object>>> _pipelineCache = new();
19.         public CommandBus(IServiceProvider serviceProvider)
20.         {
21.             _provider = serviceProvider;
22.         }
24.         // 添加清理缓存的方法，用于测试或动态重新加载
25.         public void ClearCache()
26.         {
27.             _handlerCache.Clear();
28.             _behaviorsCache.Clear();
29.             _pipelineCache.Clear();
30.         }
32.         public async Task<TResult> SendAsync<TCommand, TResult>(TCommand command, CancellationToken ct = default) where TCommand : ICommand<TResult>
33.         {
34.             var commandType = typeof(TCommand);
35.             
36.             // 获取缓存的Handler
37.             var handler = GetCachedHandler<TCommand, TResult>(commandType);
38.             
39.             // 获取缓存的Pipeline
40.             var pipeline = GetCachedPipeline<TCommand, TResult>(commandType);
41.             
42.             // 执行Pipeline
43.             var result = await pipeline(handler, command, ct);
44.             return (TResult)result;
45.         }
47.         private ICommandHandler<TCommand, TResult> GetCachedHandler<TCommand, TResult>(Type commandType) where TCommand : ICommand<TResult>
48.         {
49.             var handlerFactory = (Func<object>)_handlerCache.GetOrAdd(commandType, _ =>
50.             {
51.                 return new Func<object>(() =>
52.                 {
53.                     using var scope = _provider.CreateScope();
54.                     var handler = scope.ServiceProvider.GetService(typeof(ICommandHandler<TCommand, TResult>));
55.                     if (handler == null)
56.                         throw new InvalidOperationException($"No handler registered for {commandType.Name}");
57.                     return handler;
58.                 });
59.             });
60.             return (ICommandHandler<TCommand, TResult>)handlerFactory();
61.         }
63.         private ICommandPipelineBehavior<TCommand, TResult>[] GetCachedBehaviors<TCommand, TResult>(Type commandType) where TCommand : ICommand<TResult>
64.         {
65.             var behaviorsFactory = (Func<object[]>)_behaviorsCache.GetOrAdd(commandType, _ =>
66.             {
67.                 return new Func<object[]>(() =>
68.                 {
69.                     using var scope = _provider.CreateScope();
70.                     var behaviors = scope.ServiceProvider.GetServices<ICommandPipelineBehavior<TCommand, TResult>>().ToArray();
71.                     return behaviors.Cast<object>().ToArray();
72.                 });
73.             });
74.             return behaviorsFactory().Cast<ICommandPipelineBehavior<TCommand, TResult>>().ToArray();
75.         }
77.         private Func<object, object, CancellationToken, Task<object>> GetCachedPipeline<TCommand, TResult>(Type commandType) where TCommand : ICommand<TResult>
78.         {
79.             return _pipelineCache.GetOrAdd(commandType, _ =>
80.             {
81.                 var behaviors = GetCachedBehaviors<TCommand, TResult>(commandType);
82.                
83.                 // 预构建Pipeline，避免每次调用时重新构建
84.                 return async (handler, command, ct) =>
85.                 {
86.                     if (handler == null || command == null)
87.                         throw new ArgumentNullException("Handler or command cannot be null");
88.                         
89.                     var typedHandler = (ICommandHandler<TCommand, TResult>)handler;
90.                     var typedCommand = (TCommand)command;
92.                     // 如果没有behaviors，直接调用handler
93.                     if (behaviors.Length == 0)
94.                     {
95.                         var result = await typedHandler.HandleAsync(typedCommand, ct);
96.                         return (object)result!;
97.                     }
99.                     // 使用递归方式构建pipeline，减少委托创建
100.                     var pipelineResult = await ExecutePipeline(typedHandler, typedCommand, behaviors, 0, ct);
101.                     return (object)pipelineResult!;
102.                 };
103.             });
104.         }
106.         private async Task<TResult> ExecutePipeline<TCommand, TResult>(
107.             ICommandHandler<TCommand, TResult> handler,
108.             TCommand command,
109.             ICommandPipelineBehavior<TCommand, TResult>[] behaviors,
110.             int behaviorIndex,
111.             CancellationToken ct) where TCommand : ICommand<TResult>
112.         {
113.             if (behaviorIndex >= behaviors.Length)
114.             {
115.                 return await handler.HandleAsync(command, ct);
116.             }
118.             var behavior = behaviors[behaviorIndex];
119.             return await behavior.Handle(command, () => ExecutePipeline(handler, command, behaviors, behaviorIndex + 1, ct), ct);
120.         }
121.     }
122. }

复制代码

 
其他批量操作、带监控等模式就参考其他代码：
exercisebook/AOP/EventBusAOP/AopNew at main · liuzhixin405/exercisebook

一下是项目更详细介绍，如有错误多多指正：
# CommandBus AOP 项目
这是一个基于AOP（面向切面编程）的CommandBus项目，使用TPL Dataflow进行数据流处理优化，支持多种CommandBus实现和实时监控。
## CommandBus实现类型
### 1. Standard CommandBus- **类型**: `CommandBusType.Standard`- **特点**: 标准同步处理，适合简单场景- **控制器**: `StandardCommandBusController`
### 2. Dataflow CommandBus- **类型**: `CommandBusType.Dataflow`- **特点**: 基于TPL Dataflow的异步并发处理，适合高并发场景- **控制器**: `DataflowCommandBusController`
### 3. Batch Dataflow CommandBus- **类型**: `CommandBusType.BatchDataflow`- **特点**: 支持批量处理，适合大批量数据场景- **控制器**: `BatchDataflowCommandBusController`
### 4. Typed Dataflow CommandBus- **类型**: `CommandBusType.TypedDataflow`- **特点**: 强类型安全，适合复杂业务场景- **控制器**: `TypedDataflowCommandBusController`
### 5. Monitored CommandBus- **类型**: `CommandBusType.Monitored`- **特点**: 包含性能监控，适合生产环境- **控制器**: `MonitoredCommandBusController` 这里有一个扩展点behavior，可以注入前后时间，当前代代码只做了业务前的拦截，业务后的可以如法炮制。这样的话就是一个aop，那么跟aop切面编程又有什么区别和共同点呢？

1.             // 构建处理管道
2.             Func<Task<TResult>> pipeline = () => ExecuteHandler<TCommand, TResult>(handler, (TCommand)request.Command);
3.             
4.             // 按顺序应用管道行为
5.             foreach (var behavior in behaviors.Reverse())
6.             {
7.                 var currentBehavior = behavior;
8.                 var currentPipeline = pipeline;
9.                 pipeline = async () => (TResult)await ExecuteBehavior(currentBehavior, (TCommand)request.Command, currentPipeline);
10.             }
11.             
12.             return await pipeline();
13.         }

复制代码

 
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！