引言

大家好像對分析源碼厭倦了，說實在我也會厭倦，不過不看是無法分析其后面的東西，從易到難是一個必要的過程。

今天說下EventBus，前幾天園里的大神已經把其解刨事件總線（Event Bus）知多少，我今天就借著大神的肩膀，分析下在eShop項目中EventBus的實現。

最近發覺轉發文章不寫出處的，特此加上鏈接：http://inday.cnblogs.com

解析源碼

我們知道使用EventBus是為了解除Publisher和Subscriber之間的依賴性，這樣我們的Publisher就不需要知道有多少Subscribers，只需要通過EventBus進行注冊管理就好了，在eShop項目中，有一個這樣的接口IEventBus（eShopOnContainers\src\BuildingBlocks\EventBus\EventBus\Abstractions）

public interface IEventBus { ?
??void Subscribe<T, TH>(Func<TH> handler) ? ? ? ?where T : IntegrationEvent ? ? ? ?where TH : IIntegrationEventHandler<T>; ? ?

? ?void Unsubscribe<T, TH>() ? ? ? ?where TH : IIntegrationEventHandler<T> where T : IntegrationEvent; ?

??void Publish(IntegrationEvent @event); }

我們可以看到這個接口定義了EventBus所需的一些操作， 對比大神的EventBus，相關功能都是一致的，我們看下它的實現類：EventBusRabbitMQ，從名字上可以看出，這是一個通過RabbitMQ來進行管理的EventBus，我們可以看到它使用了IEventBusSubscriptionsManager進行訂閱存儲，也就是大神文中的：

private readonly ConcurrentDictionary<Type, List<Type>> _eventAndHandlerMapping;

微軟在Demo中把其提取出了接口，把一些常用方法給提煉了出來，但是核心還是Dictionary<string, List<Delegate>>， 使用Dictionary進行Map映射。通過Subscribe和UnSubscribe進行訂閱和取消，使用Publish方法進行發布操作。

public void Subscribe<T, TH>(Func<TH> handler) ? ?where T : IntegrationEvent ? ?where TH : IIntegrationEventHandler<T>{ ?
?var eventName = typeof(T).Name; ?
?var containsKey = _subsManager.HasSubscriptionsForEvent<T>(); ?
?if (!containsKey){ ? ? ?
? ? ? ?if (!_persistentConnection.IsConnected){_persistentConnection.TryConnect();} ? ? ?
? ? ??using (var channel = _persistentConnection.CreateModel()){channel.QueueBind(queue: _queueName,exchange: BROKER_NAME,routingKey: eventName);}}_subsManager.AddSubscription<T, TH>(handler); }

我們看到在訂閱的時候，EventBus會檢查下在Map中是否有相應的注冊，如果沒有的話首先回去RabbitMQ中創建一個新的channel進行綁定，隨后在Map中進行注冊映射。

UnSubscribe則直接從Map中取消映射，通過OnEventRemoved事件判斷Map下此映射的subscriber是否為空，為空則從RabbitMQ中關閉channel。

在RabbitMQ的構造方法中，我們看到這樣一個創建：CreateConsumerChannel()，這里創建了一個EventingBasicConsumer，當Queue中有新的消息時會通過ProcessEvent執行Map中注冊的handler（subscribers），看圖可能更清晰些：

在ProcessEvent方法中，回去Map中找尋subscribers，然后通過動態反射進行執行：

private async Task ProcessEvent(string eventName, string message) { ? ?if (_subsManager.HasSubscriptionsForEvent(eventName)){ var eventType = _subsManager.GetEventTypeByName(eventName); ? ? ? ?var integrationEvent = JsonConvert.DeserializeObject(message, eventType); ? ? ? ?var handlers = _subsManager.GetHandlersForEvent(eventName); ? ?
? ?foreach (var handlerfactory in handlers){ ? ? ? ? ?
??var handler = handlerfactory.DynamicInvoke(); ? ? ?
?? var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType); ?
? ? ?await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });}} }

微軟通過簡單的代碼解耦了Publisher和Subscribers之間的依賴關系，我們引用大神的總結：

應用

在catalog.api中，微軟出現了EventBus，我在上一篇中也提到了，這是我的一個疑惑，因為在catalog中并沒有訂閱操作，直接執行了Publish操作，原先以為是一個空操作，后來看了Basket.Api我才知道為何微軟要用RabbitMQ。

使用RabbitMQ，我們不僅是從類之間的解耦，更可以跨項目，跨語言，跨平臺的解耦，publisher僅僅需要把消息體(IntegrationEvent)傳送到RabbitMQ,Consumer從Queue中獲取消息體，然后推送到Subscribers執行相應的操作。我們看下Basket.Api.Startup.cs：

protected virtual void ConfigureEventBus(IApplicationBuilder app) { ?
??var catalogPriceHandler = app.ApplicationServices ? ? ? ?.GetService<IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>>(); ? ?var orderStartedHandler = app.ApplicationServices.GetService<IIntegrationEventHandler<OrderStartedIntegrationEvent>>(); ? ?var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>(() => app.ApplicationServices.GetRequiredService<ProductPriceChangedIntegrationEventHandler>());eventBus.Subscribe<OrderStartedIntegrationEvent, OrderStartedIntegrationEventHandler>(() => app.ApplicationServices.GetRequiredService<OrderStartedIntegrationEventHandler>()); }

在這個方法里，我們看到了Subscribe操作，想想之前的提問有點搞笑，不過研究明白了也不錯，對吧！

總結

今天我們看了EventBus在Demo中的應用，總結一下。

1、EventBus可以很好的解耦訂閱者和發布者之間的依賴

2、使用RabbitMQ能夠跨項目、跨平臺、跨語言的解耦訂閱者和發布者

雖然在Demo中我們看到對訂閱者的管理是通過Dictionary內存的方式，所以我們的Subscribe僅僅只在Basket.Api中看到，但微軟是通過IEventBusSubscriptionsManager接口定義的，我們可以通過自己的需求來進行定制，可以做成分布式的，比如使用memcached。

寫在最后

每個月到下旬就會比較忙，所以文章發布會比較慢，但我也會堅持學習完eShop的，為了學習，我建了個群，大家可以進來一起學習，有什么建議和問題都可以進來哦。

eShop雖好，但不建議大家放到生產環境，畢竟是一個Demo，而且目前還是ALPHA版本，用來學習是一個很好的教材，這就是一個大雜燴，學習中你會學到很多新的東西，大家如果看好core的發展，可以一起研究下。

QQ群：376248054

相關文章：

• 事件總線（Event Bus）知多少

• 開篇有益-解析微軟微服務架構eShopOnContainers（一）

• Identity Service - 解析微軟微服務架構eShopOnContainers（二）

• Catalog Service - 解析微軟微服務架構eShopOnContainers（三）
  

原文地址：http://www.cnblogs.com/inday/p/eventbus-in-eshopcontainers.html

---

.NET社區新聞，深度好文，微信中搜索dotNET跨平臺或掃描二維碼關注

總結

以上是生活随笔為你收集整理的EventBus In eShop -- 解析微软微服务架构eShopOnContainers（四）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。