网站内部优化的方法,php做网站商城系统怎么样,营销网站建设教学,长宁区网站建设公司最近在看微软eShopOnContainers 项目#xff0c;看到事件总线觉得不错#xff0c;和大家分享一下看完此文你将获得什么#xff1f;eShop中是如何设计事件总线的实现一个InMemory事件总线eShop中是没有InMemory实现的#xff0c;这算是一个小小小的挑战发布订阅模式发布订阅… 最近在看微软eShopOnContainers 项目看到事件总线觉得不错和大家分享一下看完此文你将获得什么eShop中是如何设计事件总线的实现一个InMemory事件总线eShop中是没有InMemory实现的这算是一个小小小的挑战发布订阅模式发布订阅模式可以让应用程序组件之间解耦这是我们使用这种模式最重要的理由之一如果你完全不知道这个东西建议你先通过搜索引擎了解一下这种模式网上的资料很多这里就不再赘述了。eShop中的EventBus就是基于这种模式的发布/订阅。发布订阅模式核心概念有三个发布者、订阅者、调度中心这些概念在消息队列中就是生产者、消费者、MQ实例。在eShop中有两个EventBus的实现基于RabbitMq的EventBusRabbitMQ基于AzureServiceBus的EventBusServiceBus。从IEventBus开始先来看一看所有EventBus的接口IEventBuspublic interface IEventBus{ void Publish(IntegrationEvent event); void SubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT; void SubscribeDynamicTH(string eventName) where TH : IDynamicIntegrationEventHandler; void UnsubscribeDynamicTH(string eventName) where TH : IDynamicIntegrationEventHandler; void UnsubscribeT, TH() where TH : IIntegrationEventHandlerT where T : IntegrationEvent;
}嗯乍一看看是有点眼晕的仔细看它的核心功能只有三个Publish 发布Subscribe 订阅Unsubscribe 取消订阅这对应着发布订阅模式的基本概念不过对于事件总线的接口添加了许多约束发布的内容消息必须是IntegrationEvent及其子类订阅事件必须指明要订阅事件的类型并附带处理器类型处理器必须是IIntegrationEventHandler的实现类Ok看到这里先不要管Dynamic相关的方法然后记住这个两个关键点事件必须继承IntegrationEvent处理器必须实现IIntegrationEventHandlerT且T是IntegrationEvent子类另外看下 IntegrationEvent有什么public class IntegrationEvent{ public IntegrationEvent() {Id Guid.NewGuid();CreationDate DateTime.UtcNow;} public Guid Id { get; } public DateTime CreationDate { get; }
}IEventBusSubscriptionsManager是什么public interface IEventBusSubscriptionsManager{ bool IsEmpty { get; } event EventHandlerstring OnEventRemoved; void AddDynamicSubscriptionTH(string eventName) where TH : IDynamicIntegrationEventHandler; void AddSubscriptionT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT; void RemoveSubscriptionT, TH() where TH : IIntegrationEventHandlerT where T : IntegrationEvent; void RemoveDynamicSubscriptionTH(string eventName) where TH : IDynamicIntegrationEventHandler; bool HasSubscriptionsForEventT() where T : IntegrationEvent; bool HasSubscriptionsForEvent(string eventName); Type GetEventTypeByName(string eventName); void Clear(); IEnumerableSubscriptionInfo GetHandlersForEventT() where T : IntegrationEvent; IEnumerableSubscriptionInfo GetHandlersForEvent(string eventName); string GetEventKeyT();
}这个接口看起来稍显复杂些我们来简化下看看public interface IEventBusSubscriptionsManager{ void AddSubscriptionT, TH() void RemoveSubscriptionT, TH()IEnumerableSubscriptionInfo GetHandlersForEventT()
}最终这三个方法就是我们要关注的添加订阅、移除订阅、获取指定事件的订阅信息。SubscriptionInfo是什么public bool IsDynamic { get; }public Type HandlerType{ get; }SubscriptionInfo中只有两个信息这是不是一个Dynamic类型的Event以及这个Event所对应的处理器的类型。这是你可能会有另一个疑问这个和IEventBus有什么关系IEventBusSubscriptionsManager含有更多功能查看是否有订阅获取事件的Type获取事件的处理器等等IEventBusSubscriptionsManager由IEventBus使用在RabbitMq和ServiceBus的实现中都使用Manager去存储事件的信息例如下面的代码public void SubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT
{ // 查询事件的全名var eventName _subsManager.GetEventKeyT(); //向mq添加注册DoInternalSubscription(eventName); // 向manager添加订阅_subsManager.AddSubscriptionT, TH();
}private void DoInternalSubscription(string eventName){ var containsKey _subsManager.HasSubscriptionsForEvent(eventName); if (!containsKey){ if (!_persistentConnection.IsConnected){_persistentConnection.TryConnect();} using (var channel _persistentConnection.CreateModel()){channel.QueueBind(queue: _queueName,exchange: BROKER_NAME,routingKey: eventName);}}
}查询事件的名字是manager做的订阅的时候是先向mq添加订阅之后又加到manager中manager管理着订阅的基本信息。另外一个重要功能是获取事件的处理器信息在rabbit mq的实现中ProcessEvent方法中用manager获取了事件的处理器再用依赖注入获得处理器的实例反射调用Handle方法处理事件信息 private async Task ProcessEvent(string eventName, string message) { // 从manager查询信息if (_subsManager.HasSubscriptionsForEvent(eventName)){ using (var scope _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME)){ // 从manager获取处理器var subscriptions _subsManager.GetHandlersForEvent(eventName); foreach (var subscription in subscriptions){ // Di 反射调用处理事件两个都是只是针对是否是dynamic做了不同的处理if (subscription.IsDynamic){ var handler scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler; dynamic eventData JObject.Parse(message); await handler.Handle(eventData);} else{ var eventType _subsManager.GetEventTypeByName(eventName); var integrationEvent JsonConvert.DeserializeObject(message, eventType); var handler scope.ResolveOptional(subscription.HandlerType); var concreteType typeof(IIntegrationEventHandler).MakeGenericType(eventType); await (Task)concreteType.GetMethod(Handle).Invoke(handler, new object[] { integrationEvent });}}}}}IEventBusSubscriptionsManager的默认实现在eShop中只有一个实现就是InMemoryEventBusSubscriptionsManager类这个类中有两个重要的字段 private readonly Dictionarystring, ListSubscriptionInfo _handlers; private readonly ListType _eventTypes;他们分别存储了事件列表和事件处理器信息词典接下来就是实现一个基于内存的事件总线了我们要做什么呢IEventBusSubscriptionsManager 已经有了InMemory的实现了我们可以直接拿来用所以我们只需要自己实现一个EventBus就好了先贴出最终代码public class InMemoryEventBus : IEventBus{ private readonly IServiceProvider _provider; private readonly ILoggerInMemoryEventBus _logger; private readonly ISubscriptionsManager _manager; private readonly IListIntegrationEvent _events; public InMemoryEventBus(IServiceProvider provider,ILoggerInMemoryEventBus logger, ISubscriptionsManager manager) {_provider provider;_logger logger;_manager manager;} public void Publish(IntegrationEvent e) { var eventType e.GetType(); var handlers _manager.GetHandlersForEvent(eventType.FullName); foreach (var handlerInfo in handlers){ var handler _provider.GetService(handlerInfo.HandlerType); var method handlerInfo.HandlerType.GetMethod(Handle);method.Invoke(handler, new object[] { e });}} public void SubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT{_manager.AddSubscriptionT, TH();} public void SubscribeDynamicTH(string eventName) where TH : IDynamicIntegrationEventHandler{ throw new NotImplementedException();} public void UnsubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT{_manager.RemoveSubscriptionT, TH();} public void UnsubscribeDynamicTH(string eventName) where TH : IDynamicIntegrationEventHandler{ throw new NotImplementedException();}
}首先构造函数中声明我们要使用的东西public InMemoryEventBus(IServiceProvider provider,ILoggerInMemoryEventBus logger, ISubscriptionsManager manager)
{ _provider provider; _logger logger; _manager manager;
}这里要注意的就是IServiceProvider provider这是 DI容器当我们在切实处理事件的时候我们选择从DI获取处理器的实例而不是反射创建这要做的好处在于处理器可以依赖于其它东西并且可以是单例的public void SubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT
{_manager.AddSubscriptionT, TH();} public void UnsubscribeT, TH() where T : IntegrationEvent where TH : IIntegrationEventHandlerT
{_manager.RemoveSubscriptionT, TH();
}订阅和取消订阅很简单因为我们是InMemory的所以只调用了manager的方法。接下来就是最重要的Publish方法实现Publish有两种方式使用额外的线程和Queue让发布和处理异步为了简单起见我们先写个简单易懂的同步的public void Publish(IntegrationEvent e){ // 首先要拿到集成事件的Type信息var eventType e.GetType(); // 获取属于这个事件的处理器列表可能有很多注意获得的是SubscriptionInfovar handlers _manager.GetHandlersForEvent(eventType.FullName); // 不解释循环foreach (var handlerInfo in handlers){ // 从DI中获取类型的实例var handler _provider.GetService(handlerInfo.HandlerType); // 拿到Handle方法var method handlerInfo.HandlerType.GetMethod(Handle); // 调用方法method.Invoke(handler, new object[] { e });}
}OK我们的InMemoryEventBus就写好了要实践这个InMemoryEventBus那么还需要一个IntegrationEvent的子类和一个IIntegrationEventHandlerT的实现类这些都不难例如我们做一个添加用户的事件A在添加用户后发起一个事件并将新用户的名字作为事件数据B去订阅事件并在自己的处理器中处理名字信息。思路是这样的写一个 AddUserEvent:IntegrationEvent,里面有一个UserId和一个UserName。写一个AddUserEventHandler:IIntegrationEventHandlerAddUserEvent,在Handle方法中输出UserId和Name到日志。注册DI你要注册下面这些服务IEventBusInMemoryEventBusISubscriptionsManagerInMemorySubscriptionsManagerAddUserEventHandlerAddUserEventHandler在Startup中为刚刚写的事件和处理器添加订阅在这里已经可以获取到IEventBus实例了写一个Api接口或是什么调用IEventBus的Publish方法new 一个新的AddUserEvent作为参数传进去。OK到这里一个切实可用的InMemoryEventBus就可以使用了。相关文章利用Service Fabric承载eShop On Containers开篇有益-解析微软微服务架构eShopOnContainers一Identity Service - 解析微软微服务架构eShopOnContainers二Catalog Service - 解析微软微服务架构eShopOnContainers三EventBus In eShop -- 解析微软微服务架构eShopOnContainers四Health Check in eShop -- 解析微软微服务架构Demo五我眼中的ASP.NET Core之微服务微服务中的异步消息通讯.NET Core 事件总线,分布式事务解决方案CAP微服务的概念——《微服务设计》读书笔记微服务架构师的职责——《微服务设计读书笔记》建模:确定服务的边界——《微服务设计》读书笔记微服务集成——《微服务设计》读书笔记服务的协作服务间的消息传递——《微服务设计》读书笔记拆分:分解单块系统——《微服务设计》读书笔记部署:持续集成CI与持续交付CD——《微服务设计》读书笔记测试——《微服务设计》读书笔记监控——《微服务设计》读书笔记安全——《微服务设计》读书笔记康威定律和系统设计——《微服务设计》读书笔记规模化微服务——《微服务设计》读书笔记Net分布式系统之微服务架构原文https://www.cnblogs.com/rocketRobin/p/8510198.html.NET社区新闻深度好文欢迎访问公众号文章汇总 http://www.csharpkit.com
