1. 引言

事件總線這個概念對你來說可能很陌生，但提到觀察者（發布-訂閱）模式，你也許就很熟悉。事件總線是對發布-訂閱模式的一種實現。它是一種集中式事件處理機制，允許不同的組件之間進行彼此通信而又不需要相互依賴，達到一種解耦的目的。

從上圖可(ke)知，核心就4個角色：

1. 事件（事件源+事件處理）
2. 事件發布者
3. 事件訂閱者
4. 事件總線

實現事(shi)件總線的(de)關鍵是：

1. 事件總線維護一個事件源與事件處理的映射字典；
2. 通過單例模式，確保事件總線的唯一入口；
3. 利用反射完成事件源與事件處理的初始化綁定；
4. 提供統一的事件注冊、取消注冊和觸發接口。

以上源于我在事件總線知多少（1）中(zhong)對于(yu)EventBus的(de)(de)分(fen)析(xi)和簡單總(zong)結(jie)。基于(yu)以上的(de)(de)簡單認知，我們來梳理下eShopOnContainers中(zhong)EventBus的(de)(de)實現機制·。

2. 高屋建瓴--看類圖

我們直接以上帝視角，來看下其實現機制，上類圖。

我們知道事件的本質是：事件源+事件處理。
針對事件源，其定義了IntegrationEvent基類(lei)來處理。默認僅包含一(yi)個(ge)guid和一(yi)個(ge)創建日期(qi)，具體的(de)事件可以通過繼承該類(lei)，來完善(shan)事件的(de)描(miao)述信息。

這里有必要解釋下Integration Event（集成事件）。因為在(zai)(zai)微服(fu)務中(zhong)事件(jian)的(de)消費不(bu)再局限(xian)于(yu)當前領域(yu)內，而是多(duo)個微服(fu)務可能共(gong)享(xiang)同一個事件(jian)，所(suo)以這里要和DDD中(zhong)的(de)領域(yu)事件(jian)區分開來。集(ji)成事件(jian)可用于(yu)跨多(duo)個微服(fu)務或(huo)外(wai)部系統同步領域(yu)狀(zhuang)態，這是通過在(zai)(zai)微服(fu)務之外(wai)發(fa)布(bu)集(ji)成事件(jian)來實(shi)現(xian)的(de)。

針對事件處理，其本質是對事件的反應，一個事件可引起多個反應，所以，它們之間是一對多的關系。
eShopOnContainers中抽(chou)象(xiang)了兩個事件(jian)處理的接口：

1. IIntegrationEventHandler
2. IDynamicIntegrationEventHandler

二者都定義了一個Handle方法用于響應事件。不同之處在于方法參數的類型：
第一個接受的是一個強類型的IntegrationEvent。第二個接收的是一個動態類型dynamic。
為什么要單獨提供一個事件源為dynamic類型的接口呢？
不是每一個事件源都需要詳細的事件信息，所以一個強類型的參數約束就沒有必要，通過dynamic可以簡(jian)化事件源的構建，更趨(qu)于靈活。

有了事件源和事件處理，接下來就是事件的注冊和訂閱了。為了方便進行訂閱管理，系統提供了額外的一層抽象IEventBusSubscriptionsManager，其用于維護事件的訂閱和注銷，以及訂閱信息的持久化。其默認的實現InMemoryEventBusSubscriptionsManager就是使用內存進行存儲事件源和事件處理的映射字典。
從類圖中看InMemoryEventBusSubscriptionsManager中定義了一個內部類SubscriptionInfo，其主要用(yong)于表示事件訂閱(yue)(yue)方(fang)的訂閱(yue)(yue)類型和事件處(chu)理的類型。

我們來近距離看下InMemoryEventBusSubscriptionsManager的定義：

//InMemoryEventBusSubscriptionsManager.cs
//定義的事件名稱和事件訂閱的字典映射（1:N）
private readonly Dictionary<string, List<SubscriptionInfo>> _handlers;
//保存所有的事件處理類型
private readonly List<Type> _eventTypes;
//定義事件移除后事件
public event EventHandler<string> OnEventRemoved;

//構造函數初始化
public InMemoryEventBusSubscriptionsManager()
{
    _handlers = new Dictionary<string, List<SubscriptionInfo>>();
    _eventTypes = new List<Type>();
}
//添加動態類型事件訂閱（需要手動指定事件名稱）
public void AddDynamicSubscription<TH>(string eventName)
    where TH : IDynamicIntegrationEventHandler
{
    DoAddSubscription(typeof(TH), eventName, isDynamic: true);
}
//添加強類型事件訂閱（事件名稱為事件源類型）
public void AddSubscription<T, TH>()
    where T : IntegrationEvent
    where TH : IIntegrationEventHandler<T>
{
    var eventName = GetEventKey<T>();

    DoAddSubscription(typeof(TH), eventName, isDynamic: false);

    if (!_eventTypes.Contains(typeof(T)))
    {
        _eventTypes.Add(typeof(T));
    }
}
//移除動態類型事件訂閱
public void RemoveDynamicSubscription<TH>(string eventName)
    where TH : IDynamicIntegrationEventHandler
{
    var handlerToRemove = FindDynamicSubscriptionToRemove<TH>(eventName);
    DoRemoveHandler(eventName, handlerToRemove);
}

//移除強類型事件訂閱
public void RemoveSubscription<T, TH>()
    where TH : IIntegrationEventHandler<T>
    where T : IntegrationEvent
{
    var handlerToRemove = FindSubscriptionToRemove<T, TH>();
    var eventName = GetEventKey<T>();
    DoRemoveHandler(eventName, handlerToRemove);
}

添加了這么一層抽象，即符合了單一職責原則，又完成了代碼重用。IEventBus的具體實現通過注入對IEventBusSubscriptionsManager的依賴，即可完成訂閱管理。
你這里可能會好奇，為什么要暴露一個OnEventRemoved事件？這(zhe)里(li)先按住不表，留給大(da)家(jia)思考。

3. 使用RabbitMQ實現EventBus

3.1. 為什么需要RabbitMQ?

微服務的一大特點就是分布式。若需要做到動一發而牽全身，就需要一個持久化的集中式的EventBus。這就要求各個微服務內部雖然分別持有一個對EventBus的引用，但它們背后都必須連接著同一個用于持久化的數據源。

那(nei)你可能(neng)會說：那(nei)這個(ge)很(hen)好實現，使用同一個(ge)數(shu)據庫就好了(le)。為(wei)什(shen)么非(fei)要用個(ge)什(shen)么RabbitMQ？問(wen)的好！這就要去(qu)探討下(xia)RabbitMQ是為(wei)了(le)解決什(shen)么問(wen)題了(le)。

RabbitMQ提(ti)供了可靠的(de)消(xiao)息(xi)(xi)機制(zhi)、跟蹤機制(zhi)和靈活(huo)的(de)消(xiao)息(xi)(xi)路由，支持消(xiao)息(xi)(xi)集群和分(fen)布式部(bu)署。適用(yong)于排隊算(suan)法、秒殺活(huo)動、消(xiao)息(xi)(xi)分(fen)發(fa)、異步處理、數(shu)據(ju)同步、處理耗(hao)時任務、CQRS等應用(yong)場(chang)景。

而關于RabbitMQ的具體使用，這里不再展開，可參考RabbitMQ知多少。

3.2. EventBus集成RabbitMQ的核心

集成(cheng)RabbitMQ的(de)(de)關鍵(jian)在于理(li)解其對消(xiao)息的(de)(de)處理(li)機制：

1. 消息的生產者和消費者通過與服務器（Broker）建立連接，然后基于創建的信道（Chanel）進行消息的發生和接收。
2. 消息的生產者可以通過聲明指定的隊列（queue）或交換機（exchange）以及路由（routingKey）進行消息的發送。
3. 消息的消費者通過綁定到相應的隊列（queue）或交換機（exchange）監聽相應的路由（routingKey），進行消息的接收。
4. 消息的消費者通過構造消費者實例綁定消息接收后的事件委托來進行消息消費。

3.3. 源碼一覽

基于以上的認知，我們再與EventBusRabbitMQ源碼親密接觸。

3.3.1. 構造函數定義

public class EventBusRabbitMQ : IEventBus, IDisposable
{
    const string BROKER_NAME = "eshop_event_bus";

    private readonly IRabbitMQPersistentConnection _persistentConnection;
    private readonly ILogger<EventBusRabbitMQ> _logger;
    private readonly IEventBusSubscriptionsManager _subsManager;
    private readonly ILifetimeScope _autofac;
    private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
    private readonly int _retryCount;

    private IModel _consumerChannel;
    private string _queueName;

    public EventBusRabbitMQ(IRabbitMQPersistentConnection persistentConnection, ILogger<EventBusRabbitMQ> logger,
        ILifetimeScope autofac, IEventBusSubscriptionsManager subsManager, string queueName = null, int retryCount = 5)
    {
        _persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
        _queueName = queueName;
        _consumerChannel = CreateConsumerChannel();
        _autofac = autofac;
        _retryCount = retryCount;
        _subsManager.OnEventRemoved += SubsManager_OnEventRemoved;
    }

    private void SubsManager_OnEventRemoved(object sender, string eventName)
    {
        if (!_persistentConnection.IsConnected)
        {
            _persistentConnection.TryConnect();
        }

        using (var channel = _persistentConnection.CreateModel())
        {
            channel.QueueUnbind(queue: _queueName, exchange: BROKER_NAME, routingKey: eventName);

            if (_subsManager.IsEmpty)
            {
                _queueName = string.Empty;
                _consumerChannel.Close();
            }
        }
    }
//....
}

構造函(han)數主(zhu)要做了以下幾件事：

1. 注入IRabbitMQPersistentConnection以便連接到對應的Broke。
2. 使用空對象模式注入IEventBusSubscriptionsManager ，進行訂閱管理。
3. 創建消費者信道，用于消息消費。
4. 注冊OnEventRemoved事件，取消隊列的綁定。（這也就回答了上面遺留的問題）

3.3.2. 事件訂閱的邏輯：

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);
        }
    }
}

從(cong)上面我們可以(yi)看到事(shi)件(jian)的訂閱主要是進行(xing)rabbitmq隊(dui)列的綁定。以(yi)eventName為routingKey進行(xing)路由。

3.3.3. 事件的發布邏輯

public void Publish(IntegrationEvent @event)
{
    if (!_persistentConnection.IsConnected)
    {
        _persistentConnection.TryConnect();
    }

    var policy = RetryPolicy.Handle<BrokerUnreachableException>()
        .Or<SocketException>()
        .WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) =>
        {
            _logger.LogWarning(ex.ToString());
        });

    using (var channel = _persistentConnection.CreateModel())
    {
        var eventName = @event.GetType()
            .Name;

        channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct");

        var message = JsonConvert.SerializeObject(@event);
        var body = Encoding.UTF8.GetBytes(message);

        policy.Execute(() =>
        {
            var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2; // persistent

            channel.BasicPublish(exchange: BROKER_NAME, routingKey: eventName, mandatory:true, basicProperties: properties, body: body);
        });
    }
}

這里面有以下(xia)幾個知識點：

1. 使用Polly，以2的階乘的時間間隔進行重試。（第一次2s后，第二次4s后，第三次8s后...重試）
2. 使用direct全匹配、單播形式的路由機制進行消息分發
3. 消息主體是格式化的json字符串
4. 指定DeliveryMode = 2進行消息持久化
5. 指定mandatory: true告知服務器當根據指定的routingKey和消息找不到對應的隊列時，直接返回消息給生產者。

3.3.4. 然后看看事件消息的監聽

private IModel CreateConsumerChannel()
{
    if (!_persistentConnection.IsConnected)
    {
        _persistentConnection.TryConnect();
    }
    var channel = _persistentConnection.CreateModel();
    channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct");
    channel.QueueDeclare(queue: _queueName, durable: true, exclusive: false,autoDelete: false, arguments: null);
    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += async (model, ea) =>
    {
        var eventName = ea.RoutingKey;
        var message = Encoding.UTF8.GetString(ea.Body);
        await ProcessEvent(eventName, message);
        channel.BasicAck(ea.DeliveryTag, multiple:false);
    };
    channel.BasicConsume(queue: _queueName, autoAck: false, consumer: consumer);
    channel.CallbackException += (sender, ea) =>
    {
        _consumerChannel.Dispose();
        _consumerChannel = CreateConsumerChannel();
    };
    return channel;
}

以(yi)上代碼演(yan)示了如(ru)創建消(xiao)費信道進行消(xiao)息(xi)處理的步驟：

1. 創建信道Channel
2. 并申明Exchange
3. 實例化綁定Channel的消費者實例
4. 注冊Received事件委托處理消息接收事件
5. 調用channel.BasicConsume啟動監聽

3.3.5. 具體的事件處理

private async Task ProcessEvent(string eventName, string message)
{
    if (_subsManager.HasSubscriptionsForEvent(eventName))
    {
        using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
        {
            var subscriptions = _subsManager.GetHandlersForEvent(eventName);
            foreach (var subscription in subscriptions)
            {
                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 });
                }
            }
        }
    }
}

以上代碼主要包括以下知識點：

1. Json字符串的反序列化
2. 利用依賴注入容器解析集成事件（Integration Event）和事件處理（Event Handler）類型
3. 反射調用具體的事件處理方法

4. EventBus的集成和使用

以上介紹了EventBus的實現要點，那各個微服務是如何集成呢？
1. 注冊IRabbitMQPersistentConnection服務用于設置RabbitMQ連接

services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
{
    var logger = sp.GetRequiredService<ILogger<DefaultRabbitMQPersistentConnection>>();
    //...
    return new DefaultRabbitMQPersistentConnection(factory, logger, retryCount);
});

2. 注冊單例模式的IEventBusSubscriptionsManager用于訂閱管理
services.AddSingleton<IEventBusSubscriptionsManager, InMemoryEventBusSubscriptionsManager>();

3. 注冊單例模式的EventBusRabbitMQ

services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>
{
    var rabbitMQPersistentConnection = sp.GetRequiredService<IRabbitMQPersistentConnection>();
    var iLifetimeScope = sp.GetRequiredService<ILifetimeScope>();
    var logger = sp.GetRequiredService<ILogger<EventBusRabbitMQ>>();
    var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();

    var retryCount = 5;
    if (!string.IsNullOrEmpty(Configuration["EventBusRetryCount"]))
    {
        retryCount = int.Parse(Configuration["EventBusRetryCount"]);
    }

    return new EventBusRabbitMQ(rabbitMQPersistentConnection, logger, iLifetimeScope, eventBusSubcriptionsManager, subscriptionClientName, retryCount);
});

完成了以(yi)上集成，就(jiu)可(ke)以(yi)在代(dai)碼中使用事件總線，進行(xing)事件的(de)發布和訂閱。

4. 發布事件
若要發布事件，需要根據是否需要事件源（參數傳遞）來決定是否需要申明相應的集成事件，需要則繼承自IntegrationEvent進行申明。然后在需要發布事件的地方進行實例化，并通過調用IEventBus的實例的Publish方法進行發布。

//事件源的聲明
public class ProductPriceChangedIntegrationEvent : IntegrationEvent
{        
    public int ProductId { get; private set; }

    public decimal NewPrice { get; private set; }

    public decimal OldPrice { get; private set; }

    public ProductPriceChangedIntegrationEvent(int productId, decimal newPrice, decimal oldPrice)
    {
        ProductId = productId;
        NewPrice = newPrice;
        OldPrice = oldPrice;
    }
}

//聲明事件源
var priceChangedEvent = new ProductPriceChangedIntegrationEvent(1001, 200.00, 169.00)
//發布事件
_eventBus.Publish(priceChangedEvent）

5. 訂閱事件
若要訂閱事件，需要根據需要處理的事件類型，申明對應的事件處理類，繼承自IIntegrationEventHandler或IDynamicIntegrationEventHandler，并注冊到IOC容器。然后創建IEventBus的實例調用Subscribe方法進行顯式訂閱。

//定義事件處理
public class ProductPriceChangedIntegrationEventHandler : IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
{
    public async Task Handle(ProductPriceChangedIntegrationEvent @event)
    {
        //do something
    }
}

//事件訂閱
var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>();

6. 跨服務事件消費
在微服務中跨服務事件消費很普遍，這里有一點需要說明的是如果訂閱的強類型事件非當前微服務中訂閱的事件，需要復制定義訂閱的事件類型。換句話說，比如在A服務發布的TestEvent事件，B服務訂閱該事件，同樣需要在B服務復制定義一個TestEvent。
這也是微服務(wu)的(de)一個通病，重復代碼(ma)。

5. 最后

通過一步(bu)一步(bu)的源(yuan)碼梳(shu)理，我們發現(xian)eShopOnContainers中事件(jian)總(zong)線的總(zong)體實現(xian)思路(lu)與(yu)引言部分的介紹十(shi)分契合。所以(yi)對于事件(jian)總(zong)線，不(bu)要(yao)覺得高深，明確參與(yu)的幾(ji)個角色(se)以(yi)及(ji)基本的實現(xian)步(bu)驟，那么不(bu)管是基于RabbitMQ實現(xian)也好(hao)(hao)還是基于Azure Service Bus也好(hao)(hao)，萬變不(bu)離其(qi)宗！