致敬1024,《手搓》轻量级EventBus

技术分享 5个月前 (10-25) 0 999+

关注

一、MediatR

.NET事件总线一般使用MediatR

或者基于MediatR二次封装

笔者手搓事件总线和MediatR对比一下

二、事件处理的基本概念

1. 事件

表示已经发生的事情,需要通知其他模块进行处理

2. 事件发布器

负责发布事件的对象

3. 事件处理器

实际接收到通知并处理事件的对象

4. 事件分发器

负责将事件通知给对应的事件处理器

5. 事件总线

用于协调各个组件

三、先看一下MediatR的Case

用INotification来表示事件

用INotificationHandler来表示事件处理器

IMediator双重身份,既是事件发布器,又是事件总线

INotificationPublisher是事件分发器

1. 测试代码

使用GenericNotification表示事件

CreateNotification()表示主业务,使用异步处理

A和B两个处理器表示其他模块业务订阅该事件,并进行异步处理

以上模拟一个简单事件使用的过程

record GenericNotification(string Name) : INotification; static async Task<GenericNotification> CreateNotification() {     var notification = new GenericNotification("GenericNotification");     await Task.Delay(1000);     return await Task.FromResult(notification); } class GenericANotificationHandler : INotificationHandler<GenericNotification> {     public async Task Handle(GenericNotification notification, CancellationToken cancellationToken)     {         Console.WriteLine($"GenericANotificationHandler {notification.Name},ThreadId:{Environment.CurrentManagedThreadId}");         await Task.Delay(1000, cancellationToken);     } } class GenericBNotificationHandler : INotificationHandler<GenericNotification> {     public async Task Handle(GenericNotification notification, CancellationToken cancellationToken)     {         Console.WriteLine($"GenericBNotificationHandler {notification.Name},ThreadId:{Environment.CurrentManagedThreadId}");         await Task.Delay(2000, cancellationToken);     } }

2. 运行一下

使用MediatR的Publish方法异步发布事件

var mediator = serviceCollection.GetRequiredService<IMediator>(); var sw = Stopwatch.StartNew(); var notification = await CreateNotification(); Console.WriteLine($"Publish {notification.Name},ThreadId:{Environment.CurrentManagedThreadId}"); await mediator.Publish(notification); Console.WriteLine($"总耗时：{sw.ElapsedMilliseconds}ms"); // Publish GenericNotification,ThreadId:11 // GenericANotificationHandler GenericNotification,ThreadId:11 // GenericBNotificationHandler GenericNotification,ThreadId:11 // 总耗时：4015ms

3. 分析一下结果

首先业务需求是很好的实现了

虽然全都用了异步,却感觉是同步执行

连实际线程ID都是一样一样的

耗时更是惨不忍睹,如果发布一个事件会拖累主业务,那谁还敢用事件呢?

有人可能会说去掉Publish的await就行了,不await你能保证处理器都全部能执行完吗?

为此笔者手搓一个事件总线,来解决这个问题

四、手搓事件总线

1. 测试代码

还是使用GenericNotification表示事件

依然使用CreateNotification()表示主业务

A和B两个处理器表示其他模块业务订阅该事件,并进行异步处理

这次用的是ITaskEventHandler接口,表示事件异步处理器

record GenericNotification(string Name) : INotification; static async Task<GenericNotification> CreateNotification() {     var notification = new GenericNotification("GenericNotification");     await Task.Delay(1000);     return await Task.FromResult(notification); } internal class AEventHandler : ITaskEventHandler<GenericNotification> {     public async Task TaskHandle(GenericNotification @event, CancellationToken cancellationToken)     {         Console.WriteLine($"AEventHandler {@event.Name},ThreadId:{Environment.CurrentManagedThreadId}");         await Task.Delay(1000, cancellationToken);     } } internal class BEventHandler : ITaskEventHandler<GenericNotification> {     public async Task TaskHandle(GenericNotification @event, CancellationToken cancellationToken)     {         Console.WriteLine($"BEventHandler {@event.Name},ThreadId:{Environment.CurrentManagedThreadId}");         await Task.Delay(1000, cancellationToken);     } }

2. 运行一下

使用IEventBus的Publish同步方法发布事件

var eventBus = serviceCollection.GetRequiredService<IEventBus>(); var sw = Stopwatch.StartNew(); var notification = await CreateNotification(); Console.WriteLine($"Publish {notification.Name},ThreadId:{Environment.CurrentManagedThreadId}"); eventBus.Publish(notification); Console.WriteLine($"总耗时：{sw.ElapsedMilliseconds}ms"); // Publish GenericNotification,ThreadId:11 // 总耗时：1008ms // AEventHandler GenericNotification,ThreadId:10 // BEventHandler GenericNotification,ThreadId:11

3. 分析一下结果

业务需求同样是很好的实现了

IMediator换成了IEventBus

但耗时只有1秒钟,也就是说没有拖慢主业务

前面显示事件处理的日志都在主业务之后,更说明是异步处理的

五. 揭秘手搓事件总线

1. IEventBus的Publish是同步方法

但实际内部是异步处理的

是调用自定义线程池来执行的

是不是有点拗口,同步方法实际是异步处理

MediatR的Publish异步方法,实际要同步处理

public interface IEventBus {     void Publish<TEvent>(TEvent @event); }

2. EventBus配置代码

使用ScanEventHandler扫描并注册事件处理器

通过EventBusOptions配置EventBus内置线程池的并发

通过内置线程池给事件处理单独提供了线程资源和并发控制

既可以保障事件处理,还可以避免事件处理耗完整个程序的CPU资源

最后注册EventBus

var serviceCollection = new ServiceCollection()     .ScanEventHandler(ServiceLifetime.Scoped, Assembly.GetExecutingAssembly())     .AddSingleton(new EventBusOptions { ConcurrencyLevel = 10 })     .AddEventBus<EventBus>();

3. 支持同步事件处理

IEventHandler是同步事件处理接口

一般最好涉及IO操作才用异步

只是简单操作用同步性能更好

如果是CPU密集操作就需要单独调用自定义线程池来排队执行

避免拖垮整个程序

internal class CEventHandler : IEventHandler<GenericNotification> {     public void Handle(GenericNotification @event)     {         Console.WriteLine($"CEventHandler {@event.Name},ThreadId:{Environment.CurrentManagedThreadId}");     } }

4. 同步执行结果

Publish GenericNotification,ThreadId:11 总耗时：1009ms AEventHandler GenericNotification,ThreadId:10 BEventHandler GenericNotification,ThreadId:11 CEventHandler GenericNotification,ThreadId:10

5. 手搓事件总线的边缘

支持net4.5+和netstandard1.1+,也就是说非.net core也支持

但是ServiceCollection支持的没有这么全啊

因此提供了Hand.EventDictionaryProvider,不依赖IOC环境

5.1 使用代码

EventBus也支持EventHandlerDictionaryProvider

只是注册事件处理器没有IOC方便,但至少能用

EventHandlerDictionaryProvider provider = new(); var handler = new Handler(); provider.AddHandler(handler); EventBus bus = new(provider, new EventBusOptions { ConcurrencyLevel = 1 });

6. 手搓事件总线并发控制测试

6.1 事件处理代码

Handler是同步事件处理器

TaskHandler是异步事件处理器

事件是string类型,手搓事件总线的事件类型没有限制

这也是和MediatR不一样的地方,MediatR只能用INotification接口表示事件

当然实际项目最好定义事件接口或者基类,便于管理和维护

从技术角度是不需要限制事件的类型

internal class Handler(ITestOutputHelper output) : IEventHandler<string> {     private readonly ITestOutputHelper _output = output;      public void Handle(string @event)     {         Thread.Sleep(10);         _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Handler: {@event},{DateTime.Now:HH:mm:ss.fff}");     } } internal class TaskHandler(ITestOutputHelper output) : ITaskEventHandler<string> {     private readonly ITestOutputHelper _output = output;      public async Task TaskHandle(string @event, CancellationToken cancellationToken)     {         await Task.Delay(10, cancellationToken);         _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} TaskHandler: {@event},{DateTime.Now:HH:mm:ss.fff}");     } }

6.2 单并发测试

ConcurrencyLevel配置为1,连续发布20个事件

Handler都是Thread10在执行

TaskHandler虽然在多个线程上执行,但基本和Handler是交替执行

由于TaskHandler异步线程是由系统默认线程池执行的,无法完全控制线程ID

但是内置的线程池控制了异步线程的开始和结束,所以也控制住了异步事件处理器的并发

EventHandlerDictionaryProvider provider = new(); EventBus bus = new(provider, new EventBusOptions { ConcurrencyLevel = 1 }); var handler = new Handler(_output); provider.AddHandler(handler); var taskHandler = new TaskHandler(_output); provider.AddEventHandler(taskHandler); for (int i = 0; i < 20; i++)     bus.Publish("Event" + i); _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Sleep"); // Thread15 Sleep // Thread11 TaskHandler: Event0,01:41:41.633 // Thread10 Handler: Event0,01:41:41.647 // Thread10 Handler: Event1,01:41:41.663 // Thread11 TaskHandler: Event1,01:41:41.663 // Thread10 Handler: Event2,01:41:41.679 // Thread11 TaskHandler: Event2,01:41:41.679 // Thread10 Handler: Event3,01:41:41.695 // Thread11 TaskHandler: Event3,01:41:41.695 // Thread10 Handler: Event4,01:41:41.711 // Thread11 TaskHandler: Event4,01:41:41.711 // Thread10 Handler: Event5,01:41:41.726 // Thread11 TaskHandler: Event5,01:41:41.726 // Thread10 Handler: Event6,01:41:41.742 // Thread11 TaskHandler: Event6,01:41:41.742 // Thread10 Handler: Event7,01:41:41.758 // Thread11 TaskHandler: Event7,01:41:41.758 // Thread10 Handler: Event8,01:41:41.774 // Thread11 TaskHandler: Event8,01:41:41.774 // Thread10 Handler: Event9,01:41:41.790 // Thread11 TaskHandler: Event9,01:41:41.790 // Thread10 Handler: Event10,01:41:41.806 // Thread11 TaskHandler: Event10,01:41:41.806 // Thread10 Handler: Event11,01:41:41.822 // Thread11 TaskHandler: Event11,01:41:41.822 // Thread10 Handler: Event12,01:41:41.838 // Thread11 TaskHandler: Event12,01:41:41.838 // Thread10 Handler: Event13,01:41:41.854 // Thread11 TaskHandler: Event13,01:41:41.854 // Thread10 Handler: Event14,01:41:41.870 // Thread11 TaskHandler: Event14,01:41:41.870 // Thread10 Handler: Event15,01:41:41.886 // Thread11 TaskHandler: Event15,01:41:41.886 // Thread10 Handler: Event16,01:41:41.902 // Thread32 TaskHandler: Event16,01:41:41.902 // Thread10 Handler: Event17,01:41:41.918 // Thread32 TaskHandler: Event17,01:41:41.918 // Thread10 Handler: Event18,01:41:41.934 // Thread32 TaskHandler: Event18,01:41:41.934 // Thread10 Handler: Event19,01:41:41.950 // Thread32 TaskHandler: Event19,01:41:41.950

6.3 多并发测试

这次ConcurrencyLevel配置为4,连续发布100个事件

Handler都是8、11、31和32等4个线程在执行,且每个线程都执行了25次

TaskHandler和Handler还是交替执行,说明并发控制没有问题

另外笔者抽空再补一篇文章,专门介绍手搓线程池的使用和原理

EventHandlerDictionaryProvider provider = new(); EventBus bus = new(provider, new EventBusOptions { ConcurrencyLevel = 4 }); var handler = new Handler(_output); provider.AddHandler(handler); var taskHandler = new TaskHandler(_output); provider.AddEventHandler(taskHandler); for (int i = 0; i < 100; i++)     bus.Publish("Event" + i); _output.WriteLine($"Thread{Environment.CurrentManagedThreadId} Sleep"); // Thread16 Sleep // Thread34 TaskHandler: Event3,01:52:53.795 // Thread31 Handler: Event2,01:52:53.794 // Thread36 TaskHandler: Event0,01:52:53.795 // Thread32 Handler: Event1,01:52:53.794 // Thread8 Handler: Event3,01:52:53.794 // Thread33 TaskHandler: Event1,01:52:53.795 // Thread35 TaskHandler: Event2,01:52:53.795 // Thread11 Handler: Event0,01:52:53.794 // Thread8 Handler: Event4,01:52:53.810 // Thread32 Handler: Event6,01:52:53.810 // Thread31 Handler: Event5,01:52:53.810 // Thread11 Handler: Event7,01:52:53.810 // Thread33 TaskHandler: Event7,01:52:53.810 // Thread33 TaskHandler: Event4,01:52:53.810 // Thread34 TaskHandler: Event5,01:52:53.810 // Thread35 TaskHandler: Event6,01:52:53.810 // Thread8 Handler: Event8,01:52:53.826 // Thread32 Handler: Event11,01:52:53.826 // Thread31 Handler: Event10,01:52:53.826 // Thread11 Handler: Event9,01:52:53.826 // Thread34 TaskHandler: Event9,01:52:53.826 // Thread33 TaskHandler: Event10,01:52:53.826 // Thread34 TaskHandler: Event8,01:52:53.826 // Thread35 TaskHandler: Event11,01:52:53.826 // Thread31 Handler: Event14,01:52:53.841 // Thread11 Handler: Event13,01:52:53.841 // Thread8 Handler: Event12,01:52:53.841 // Thread32 Handler: Event15,01:52:53.841 // Thread35 TaskHandler: Event13,01:52:53.841 // Thread35 TaskHandler: Event12,01:52:53.841 // Thread33 TaskHandler: Event15,01:52:53.841 // Thread34 TaskHandler: Event14,01:52:53.841 // Thread32 Handler: Event19,01:52:53.857 // Thread34 TaskHandler: Event16,01:52:53.857 // Thread31 Handler: Event16,01:52:53.857 // Thread8 Handler: Event18,01:52:53.857 // Thread11 Handler: Event17,01:52:53.857 // Thread33 TaskHandler: Event19,01:52:53.857 // Thread34 TaskHandler: Event18,01:52:53.857 // Thread35 TaskHandler: Event17,01:52:53.857 // Thread11 Handler: Event22,01:52:53.873 // Thread31 Handler: Event23,01:52:53.873 // Thread8 Handler: Event21,01:52:53.873 // Thread35 TaskHandler: Event23,01:52:53.873 // Thread32 Handler: Event20,01:52:53.873 // Thread35 TaskHandler: Event20,01:52:53.873 // Thread33 TaskHandler: Event22,01:52:53.873 // Thread34 TaskHandler: Event21,01:52:53.873 // Thread8 Handler: Event25,01:52:53.889 // Thread11 Handler: Event26,01:52:53.889 // Thread32 Handler: Event27,01:52:53.889 // Thread31 Handler: Event24,01:52:53.889 // Thread33 TaskHandler: Event27,01:52:53.889 // Thread34 TaskHandler: Event25,01:52:53.889 // Thread35 TaskHandler: Event26,01:52:53.889 // Thread36 TaskHandler: Event24,01:52:53.889 // Thread11 Handler: Event29,01:52:53.905 // Thread31 Handler: Event28,01:52:53.905 // Thread8 Handler: Event30,01:52:53.905 // Thread32 Handler: Event31,01:52:53.905 // Thread36 TaskHandler: Event31,01:52:53.905 // Thread36 TaskHandler: Event29,01:52:53.905 // Thread35 TaskHandler: Event28,01:52:53.905 // Thread34 TaskHandler: Event30,01:52:53.905 // Thread32 Handler: Event32,01:52:53.920 // Thread11 Handler: Event35,01:52:53.920 // Thread31 Handler: Event34,01:52:53.920 // Thread8 Handler: Event33,01:52:53.920 // Thread34 TaskHandler: Event35,01:52:53.920 // Thread36 TaskHandler: Event34,01:52:53.920 // Thread35 TaskHandler: Event32,01:52:53.920 // Thread33 TaskHandler: Event33,01:52:53.920 // Thread32 Handler: Event39,01:52:53.935 // Thread35 TaskHandler: Event39,01:52:53.935 // Thread8 Handler: Event38,01:52:53.935 // Thread11 Handler: Event37,01:52:53.935 // Thread31 Handler: Event36,01:52:53.935 // Thread33 TaskHandler: Event38,01:52:53.935 // Thread36 TaskHandler: Event36,01:52:53.935 // Thread37 TaskHandler: Event37,01:52:53.936 // Thread11 Handler: Event42,01:52:53.951 // Thread8 Handler: Event40,01:52:53.951 // Thread32 Handler: Event43,01:52:53.951 // Thread37 TaskHandler: Event41,01:52:53.951 // Thread31 Handler: Event41,01:52:53.951 // Thread33 TaskHandler: Event42,01:52:53.951 // Thread37 TaskHandler: Event40,01:52:53.951 // Thread36 TaskHandler: Event43,01:52:53.951 // Thread11 Handler: Event45,01:52:53.967 // Thread31 Handler: Event47,01:52:53.967 // Thread32 Handler: Event46,01:52:53.967 // Thread35 TaskHandler: Event44,01:52:53.967 // Thread8 Handler: Event44,01:52:53.967 // Thread36 TaskHandler: Event47,01:52:53.967 // Thread37 TaskHandler: Event46,01:52:53.967 // Thread33 TaskHandler: Event45,01:52:53.967 // Thread11 Handler: Event48,01:52:53.983 // Thread8 Handler: Event49,01:52:53.983 // Thread31 Handler: Event50,01:52:53.983 // Thread33 TaskHandler: Event51,01:52:53.983 // Thread32 Handler: Event51,01:52:53.983 // Thread33 TaskHandler: Event48,01:52:53.983 // Thread36 TaskHandler: Event49,01:52:53.983 // Thread37 TaskHandler: Event50,01:52:53.983 // Thread11 Handler: Event53,01:52:53.999 // Thread31 Handler: Event55,01:52:53.999 // Thread32 Handler: Event54,01:52:53.999 // Thread8 Handler: Event52,01:52:53.999 // Thread33 TaskHandler: Event55,01:52:53.999 // Thread33 TaskHandler: Event53,01:52:53.999 // Thread37 TaskHandler: Event54,01:52:53.999 // Thread36 TaskHandler: Event52,01:52:53.999 // Thread31 Handler: Event58,01:52:54.015 // Thread11 Handler: Event59,01:52:54.015 // Thread8 Handler: Event57,01:52:54.015 // Thread32 Handler: Event56,01:52:54.015 // Thread37 TaskHandler: Event56,01:52:54.015 // Thread36 TaskHandler: Event59,01:52:54.015 // Thread37 TaskHandler: Event57,01:52:54.015 // Thread33 TaskHandler: Event58,01:52:54.015 // Thread8 Handler: Event61,01:52:54.031 // Thread32 Handler: Event63,01:52:54.031 // Thread31 Handler: Event60,01:52:54.031 // Thread11 Handler: Event62,01:52:54.031 // Thread33 TaskHandler: Event63,01:52:54.032 // Thread36 TaskHandler: Event60,01:52:54.032 // Thread37 TaskHandler: Event61,01:52:54.032 // Thread35 TaskHandler: Event62,01:52:54.032 // Thread11 Handler: Event64,01:52:54.046 // Thread31 Handler: Event67,01:52:54.046 // Thread8 Handler: Event65,01:52:54.046 // Thread35 TaskHandler: Event67,01:52:54.047 // Thread37 TaskHandler: Event65,01:52:54.047 // Thread32 Handler: Event66,01:52:54.046 // Thread36 TaskHandler: Event66,01:52:54.047 // Thread33 TaskHandler: Event64,01:52:54.047 // Thread31 Handler: Event68,01:52:54.062 // Thread32 Handler: Event71,01:52:54.062 // Thread11 Handler: Event69,01:52:54.062 // Thread8 Handler: Event70,01:52:54.062 // Thread35 TaskHandler: Event71,01:52:54.062 // Thread36 TaskHandler: Event68,01:52:54.063 // Thread33 TaskHandler: Event70,01:52:54.063 // Thread37 TaskHandler: Event69,01:52:54.063 // Thread32 Handler: Event72,01:52:54.078 // Thread11 Handler: Event75,01:52:54.078 // Thread8 Handler: Event73,01:52:54.078 // Thread31 Handler: Event74,01:52:54.078 // Thread33 TaskHandler: Event75,01:52:54.078 // Thread37 TaskHandler: Event74,01:52:54.078 // Thread36 TaskHandler: Event73,01:52:54.078 // Thread34 TaskHandler: Event72,01:52:54.078 // Thread8 Handler: Event79,01:52:54.094 // Thread32 Handler: Event78,01:52:54.094 // Thread31 Handler: Event77,01:52:54.094 // Thread11 Handler: Event76,01:52:54.094 // Thread34 TaskHandler: Event79,01:52:54.094 // Thread36 TaskHandler: Event76,01:52:54.094 // Thread34 TaskHandler: Event77,01:52:54.094 // Thread37 TaskHandler: Event78,01:52:54.094 // Thread32 Handler: Event82,01:52:54.110 // Thread8 Handler: Event80,01:52:54.110 // Thread11 Handler: Event83,01:52:54.110 // Thread31 Handler: Event81,01:52:54.110 // Thread37 TaskHandler: Event83,01:52:54.110 // Thread36 TaskHandler: Event81,01:52:54.110 // Thread34 TaskHandler: Event82,01:52:54.110 // Thread33 TaskHandler: Event80,01:52:54.110 // Thread11 Handler: Event87,01:52:54.126 // Thread31 Handler: Event85,01:52:54.126 // Thread8 Handler: Event86,01:52:54.126 // Thread32 Handler: Event84,01:52:54.126 // Thread36 TaskHandler: Event86,01:52:54.126 // Thread34 TaskHandler: Event87,01:52:54.126 // Thread33 TaskHandler: Event84,01:52:54.126 // Thread37 TaskHandler: Event85,01:52:54.126 // Thread11 Handler: Event90,01:52:54.142 // Thread31 Handler: Event89,01:52:54.142 // Thread8 Handler: Event88,01:52:54.142 // Thread32 Handler: Event91,01:52:54.142 // Thread33 TaskHandler: Event89,01:52:54.142 // Thread37 TaskHandler: Event91,01:52:54.142 // Thread34 TaskHandler: Event88,01:52:54.142 // Thread36 TaskHandler: Event90,01:52:54.142 // Thread11 Handler: Event94,01:52:54.158 // Thread8 Handler: Event92,01:52:54.158 // Thread31 Handler: Event95,01:52:54.158 // Thread32 Handler: Event93,01:52:54.158 // Thread37 TaskHandler: Event94,01:52:54.158 // Thread36 TaskHandler: Event95,01:52:54.158 // Thread34 TaskHandler: Event93,01:52:54.158 // Thread33 TaskHandler: Event92,01:52:54.159 // Thread8 Handler: Event98,01:52:54.174 // Thread11 Handler: Event99,01:52:54.174 // Thread32 Handler: Event97,01:52:54.174 // Thread31 Handler: Event96,01:52:54.174 // Thread33 TaskHandler: Event97,01:52:54.174 // Thread34 TaskHandler: Event99,01:52:54.174 // Thread37 TaskHandler: Event96,01:52:54.174 // Thread36 TaskHandler: Event98,01:52:54.175

7. 集成事件的探讨

集成事件就是分布式事件,跨进程的事件

这种事件肯定要借助消息队列中间件

毕竟不是所有事件都需要发布到消息队列

笔者以为做一个事件转发器(继承事件处理接口)

筛选需要转发跨进程的事件

事件转发器手写也并不复杂

如何封装事件转发器笔者还没考虑清楚

下次考虑清楚了再补发一篇文章

8. 总结

IEventBus作为事件总线和事件发布者

IEventHandler和ITaskEventHandler作为事件处理器

IEventHandlerProvider用于查找事件处理器,支持非IOC环境

EventDispatcher是事件发布者包含在EventBus内部

用于调用内置线程池发布事件和跟踪异步事件执行完成

六. 手搓事件总线非常轻量级

1. MediatR已经非常轻量级了

MediatR.Contracts.dll是7K

MediatR.dll是76K

2. 手搓事件总线更轻量级

本示例用到手搓的6个类库,共61K

Hand.EventBuses.dll是9K

Hand.EventServiceProvider.dll是6K

Hand.Core.dll是14K

Hand.Tasks.dll是13K

Hand.Reflection.dll是11K

Hand.Job.dll是9K

其中只有Hand.EventBuses.dll和Hand.EventServiceProvider.dll是专用事件总线的,共15K

其他4个类库是《手搓》系列的通用类库,可以复用于其他场景

3. 对比一下MassTransit大家对轻量级就有概念了

MassTransit.Abstractions.dll是706K

MassTransit.dll是4,063K

MassTransit.RabbitMqTransport.dll是277K

七、插一个异步的题外话

1. 模拟一个异步业务场景

有3个异步逻辑的结果聚合

每个异步逻辑都需要1秒钟

public static async Task<int> Sum(params int[] input) {     await Task.Delay(1000);     return input.Sum(); } public static async Task<int> One() {     await Task.Delay(1000);     return 1; } public static async Task<int> Two() {     await Task.Delay(1000);     return 2; } public static async Task<int> Tree() {     await Task.Delay(1000);     return 3; }

2. 先给一个错误的示范

var sw = Stopwatch.StartNew(); var one = await One(); var two = await Two(); var tree = await Tree(); var sum = await Sum(one, two, tree); Assert.Equal(6, sum); sw.Stop(); _output.WriteLine($"Async Total Time: {sw.ElapsedMilliseconds} ms"); // Async Total Time: 4020 ms

以上代码看着很漂亮,但是实际执行时间是4秒

因为每个异步逻辑都是顺序执行的

这显然不是我们想要的结果

3. 可以这么优化

var sw = Stopwatch.StartNew(); var oneTask = One(); var twoTask = Two(); var treeTask = Tree(); var list = await Task.WhenAll(oneTask, twoTask, treeTask); var sum = await Sum(list); Assert.Equal(6, sum); sw.Stop(); _output.WriteLine($"WhenAll Async Total Time: {sw.ElapsedMilliseconds} ms"); // WhenAll Async Total Time: 2016 ms

以上代码耗时2秒

因为我们将异步逻辑并行执行了

很多时候不要看到Task就加await

更别觉得有Task有await,就是异步了

4. 不用WhenAll可以这么写

var sw = Stopwatch.StartNew(); var oneTask = One(); var twoTask = Two(); var treeTask = Tree(); var one = await oneTask; var two = await twoTask; var tree = await treeTask; var sum = await Sum(one, two, tree); Assert.Equal(6, sum); sw.Stop(); _output.WriteLine($"Parallel Async Total Time: {sw.ElapsedMilliseconds} ms"); // Parallel Async Total Time: 2017 ms