[Abp vNext 源碼分析] - 13. 本地事件匯流排與分散式事件匯流排 (Rabbit MQ)

一、簡要介紹 ABP vNext 封裝了兩種事件匯流排結構，第一種是 ABP vNext 自己實現的本地事件匯流排，這種事件匯流排無法跨項目發佈和訂閱。第二種則是分散式事件匯流排，ABP vNext 自己封裝了一個抽象層進行定義，並使用 RabbitMQ 編寫了一個基本實現。在使用方式上，兩種事件匯流排的作 ...

一、簡要介紹

ABP vNext 封裝了兩種事件匯流排結構，第一種是 ABP vNext 自己實現的本地事件匯流排，這種事件匯流排無法跨項目發佈和訂閱。第二種則是分散式事件匯流排，ABP vNext 自己封裝了一個抽象層進行定義，並使用 RabbitMQ 編寫了一個基本實現。

在使用方式上，兩種事件匯流排的作用基本相同。

事件匯流排分佈在兩個模塊，在 Volo.Abp.EventBus 模塊內部，定義了事件匯流排的抽象介面，以及本地事件匯流排 (ILocalEventBus) 的實現。分散式事件匯流排的具體實現，是在 Volo.Abp.EventBus.RabbitMQ 模塊內部進行定義，從項目名稱可以看出來，這個模塊是基於 RabbitMQ 消息隊列實現的。

但是該項目並不是直接引用 RabbitMQ.Client 包，而是在 Volo.Abp.RabbitMQ 項目內部引用。這是因為除了分散式事件匯流排以外，ABP 還基於 RabbitMQ 實現了一個後臺作業管理器。

ABP vNext 框架便將一些對象抽象出來，放在 Volo.Abp.RabbitMQ 項目內部進行定義和實現。

二、源碼分析

2.1 事件處理器的註冊

分析源碼，首先從一個項目的模塊開始，Volo.Abp.EventBus 庫的模塊 AbpEventBusModule 只幹了一件事情。在組件註冊的時候，根據組件的實現介面 (ILocalEventHandler 或 IDistributedEventHandler) 不同，將其賦值給 AbpLocalEventBusOptions 與 AbpDistributedEventBusOptions 的 Handlers 屬性。

也就是說，開發人員定義的事件處理程式 (Handler) 都會在依賴註入的時候，都會將其類型 (Type) 添加到事件匯流排的配置類當中，方便後續進行使用。

2.2 事件匯流排的介面

通過事件匯流排模塊的單元測試我們可以知道，事件的發佈與訂閱都是通過 IEventBus 的兩個子介面 (ILocalEventBus/IDistributedEventBus) 進行的。在 IEventBus 介面的定義中，有三種行為，分別是發佈、訂閱、取消訂閱。

對於 ILocalEventBus 介面和 IDistributedEventBus 介面來說，它們都提供了一個，針對本地事件處理器和分散式處理器的特殊訂閱方法。

ILocalEventBus：

/// <summary>
/// Defines interface of the event bus.
/// </summary>
public interface ILocalEventBus : IEventBus
{
    /// <summary>
    /// Registers to an event. 
    /// Same (given) instance of the handler is used for all event occurrences.
    /// </summary>
    /// <typeparam name="TEvent">Event type</typeparam>
    /// <param name="handler">Object to handle the event</param>
    IDisposable Subscribe<TEvent>(ILocalEventHandler<TEvent> handler)
        where TEvent : class;
}

IDistributedEventBus：

public interface IDistributedEventBus : IEventBus
{
    /// <summary>
    /// Registers to an event. 
    /// Same (given) instance of the handler is used for all event occurrences.
    /// </summary>
    /// <typeparam name="TEvent">Event type</typeparam>
    /// <param name="handler">Object to handle the event</param>
    IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler)
        where TEvent : class;
}

2.3 事件匯流排基本流程和實現

同其他模塊一樣，因為有分散式事件匯流排和本地事件匯流排，ABP vNext 同樣抽象了一個 EventBusBase 類型，作為它們的基類實現。

一般的流程，我們是先定義某個事件，然後訂閱該事件並指定事件處理器，最後在某個時刻發佈事件。例如下麵的代碼：

首先定義了一個事件處理器，專門用於處理 EntityChangedEventData<MyEntity> 事件。

public class MyEventHandler : ILocalEventHandler<EntityChangedEventData<MyEntity>>
{
    public int EntityChangedEventCount { get; set; }

    public Task HandleEventAsync(EntityChangedEventData<MyEntity> eventData)
    {
        EntityChangedEventCount++;
        return Task.CompletedTask;
    }
}

var handler = new MyEventHandler();

LocalEventBus.Subscribe<EntityChangedEventData<MyEntity>>(handler);

await LocalEventBus.PublishAsync(new EntityCreatedEventData<MyEntity>(new MyEntity()));

2.3.1 事件的訂閱

可以看到，這裡使用的是 ILocalEventBus 定義的訂閱方法，跳轉到內部實現，它還是調用的 EventBus 的方法。

public virtual IDisposable Subscribe<TEvent>(ILocalEventHandler<TEvent> handler) where TEvent : class
{
    // 調用基類的 Subscribe 方法，並傳遞 TEvent 的類型，和事件處理器。
    return Subscribe(typeof(TEvent), handler);
}

public virtual IDisposable Subscribe(Type eventType, IEventHandler handler)
{
    return Subscribe(eventType, new SingleInstanceHandlerFactory(handler));
}

可以看到，這裡傳遞了一個 SingleInstanceHandlerFactory 對象，這玩意兒是幹嘛用的呢？從名字可以看出來，這是一個工廠，是用來創建 Handler (事件處理器) 的工廠，並且是一個單實例的事件處理器工廠。

下麵就是 IEventHandlerFactory 介面的定義，以及 SingleInstanceHandlerFactory 實現。

public interface IEventHandlerFactory
{
    // 獲得一個事件處理器包裝對象，即事件處理器執行完畢之後，可以調用
    // IEventHandlerDisposeWrapper.Dispose() 進行釋放。
    IEventHandlerDisposeWrapper GetHandler();

    // 判斷在已有的事件處理器工廠集合中，是否已經存在了相同的事件處理器。
    bool IsInFactories(List<IEventHandlerFactory> handlerFactories);
}

public class SingleInstanceHandlerFactory : IEventHandlerFactory
{
    // 構造工廠時，傳遞的事件處理器實例。
    public IEventHandler HandlerInstance { get; }


    public SingleInstanceHandlerFactory(IEventHandler handler)
    {
        HandlerInstance = handler;
    }

    // 通過 EventHandlerDisposeWrapper 包裝事件處理器實例。
    public IEventHandlerDisposeWrapper GetHandler()
    {
        return new EventHandlerDisposeWrapper(HandlerInstance);
    }

    // 判斷針對 HandlerInstance 的事件處理器是否已經存在。
    public bool IsInFactories(List<IEventHandlerFactory> handlerFactories)
    {
        return handlerFactories
            .OfType<SingleInstanceHandlerFactory>()
            .Any(f => f.HandlerInstance == HandlerInstance);
    }
}

針對 IEventHandlerFactory 工廠，還擁有 3 個不同的實現，下表分別說明它們的應用場景。

實現類型	作用
`IocEventHandlerFactory`	每個工廠對應一個事件處理器的的類型，並通過 `ScopeFactory` 解析具體的事件處理器。生命周期由 `scope` 控制，當 `scope` 釋放時，對應的事件處理器實例也會被銷毀。
`SingleInstanceHandlerFactory`	每個工廠對應單獨的一個事件處理器實例，事件處理器實例是由創建者控制的。
`TransientEventHandlerFactory`	每個工廠對應一個事件處理器的類型，區別是它不由 IoC 解析實例，而是使用的 `Activator.CreateInstance()` 方法構造實例，是一個瞬時對象，調用包裝器的 `Dispose` 即會進行釋放。
`TransientEventHandlerFactory<THandler>`	每個工廠對應指定的 `THandler` 事件處理器，生命周期同上面的工廠一樣。

這幾種工廠都是在訂閱操作時，不同的訂閱重載使用不同的工廠，或者是自己指定事件處理器的工廠均可。

public virtual IDisposable Subscribe<TEvent, THandler>()
    where TEvent : class
    where THandler : IEventHandler, new()
{
    return Subscribe(typeof(TEvent), new TransientEventHandlerFactory<THandler>());
}

public virtual IDisposable Subscribe(Type eventType, IEventHandler handler)
{
    return Subscribe(eventType, new SingleInstanceHandlerFactory(handler));
}

不過有一種特殊的行為，開發人員可以 不用顯式訂閱。在 EventBus 類型中，定義了一個 SubscribeHandlers(ITypeList<IEventHandler> handlers) 方法。該方法接收一個類型集合，通過遍歷集合，從事件處理器的定義當中，取得事件處理器監聽的事件類型 TEvent。

在取得了事件類型，並知曉了事件處理器類型以後，事件匯流排就可以訂閱 TEvent 類型的事件，並使用 IocEventHandlerFactory 工廠來構造事件處理器。

protected virtual void SubscribeHandlers(ITypeList<IEventHandler> handlers)
{
    // 遍歷事件處理器的類型，其實這裡的就是模塊啟動時，傳遞給 XXXOptions 的集合。
    foreach (var handler in handlers)
    {
        // 獲得事件處理器的所有介面定義，並遍歷介面進行檢查。
        var interfaces = handler.GetInterfaces();
        foreach (var @interface in interfaces)
        {
            // 如果介面沒有實現 IEventHandler 類型，則忽略。
            if (!typeof(IEventHandler).GetTypeInfo().IsAssignableFrom(@interface))
            {
                continue;
            }

            // 從泛型參數當中，獲得定義的事件類型。
            var genericArgs = @interface.GetGenericArguments();
            // 泛型參數完全匹配 1 時，才進行訂閱操作。
            if (genericArgs.Length == 1)
            {
                Subscribe(genericArgs[0], new IocEventHandlerFactory(ServiceScopeFactory, handler));
            }
        }
    }
}

這個訂閱方法在 EventBus 當中是一個抽象方法，分別在本地事件匯流排和分散式事件匯流排有實現，這裡我們首先講解本地事件的邏輯。

public class LocalEventBus : EventBusBase, ILocalEventBus, ISingletonDependency
{
    protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }

    public LocalEventBus(
        IOptions<AbpLocalEventBusOptions> options,
        IServiceScopeFactory serviceScopeFactory)
        : base(serviceScopeFactory)
    {
        Options = options.Value;
        Logger = NullLogger<LocalEventBus>.Instance;

        HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();

        // 調用父類的方法，將模塊初始化時掃描到的事件處理器，都嘗試進行訂閱。
        SubscribeHandlers(Options.Handlers);
    }

    public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
    {
        GetOrCreateHandlerFactories(eventType)
            // 鎖住集合，以確保線程安全。
            .Locking(factories =>
                {
                    // 如果在集合內部，已經有了對應的工廠，則不進行添加。
                    if (!factory.IsInFactories(factories))
                    {
                        factories.Add(factory);
                    }
                }
            );

        // 返回一個事件處理器工廠註銷器，當調用 Dispose() 方法時，會取消之前訂閱的事件。
        return new EventHandlerFactoryUnregistrar(this, eventType, factory);
    }

    private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
    {
        // 根據事件的類型，從字典中獲得該類型的所有事件處理器工廠。
        return HandlerFactories.GetOrAdd(eventType, (type) => new List<IEventHandlerFactory>());
    }
}

上述流程結合 EventBus 和 LocalEventBus 講解了事件的訂閱流程，事件的訂閱操作都是對 HandlerFactories 的操作，往裡面添加指定事件的事件處理器工廠，而每個工廠都是跟具體的事件處理器實例/類型進行關聯的。

2.3.2 事件的發佈

當開發人員需要發佈事件的時候，一般都是通過對應的 EventBus，調用響應的 PublishAsync 方法，傳遞要觸發的事件類型與事件數據。介面和基類當中，定義了兩種發佈方法的簽名與實現：

public virtual Task PublishAsync<TEvent>(TEvent eventData) where TEvent : class
{
    return PublishAsync(typeof(TEvent), eventData);
}

public abstract Task PublishAsync(Type eventType, object eventData);

第二種方法一共也分為本地事件匯流排的實現，和分散式事件匯流排的實現，本地事件比較簡單，我們先分析本地事件匯流排的實現。

public override async Task PublishAsync(Type eventType, object eventData)
{
    // 定義了一個異常集合，用於接收多個事件處理器執行時，產生的所有異常。
    var exceptions = new List<Exception>();

    // 觸發事件處理器。
    await TriggerHandlersAsync(eventType, eventData, exceptions);

    // 如果有任何異常產生，則拋出到之前的調用棧。
    if (exceptions.Any())
    {
        if (exceptions.Count == 1)
        {
            exceptions[0].ReThrow();
        }

        throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions);
    }
}

可以看到真正的觸發行為是在 TriggerHandlersAsync(Type eventType, object eventData, List<Exception> exceptions) 內部進行實現的。

protected virtual async Task TriggerHandlersAsync(Type eventType, object eventData, List<Exception> exceptions)
{
    // 針對於這個的作用，等同於 ConfigureAwait(false) 。
    // 具體可以參考 https://blogs.msdn.microsoft.com/benwilli/2017/02/09/an-alternative-to-configureawaitfalse-everywhere/。
    await new SynchronizationContextRemover();

    // 根據事件的類型，得到它的所有事件處理器工廠。
    foreach (var handlerFactories in GetHandlerFactories(eventType))
    {
        // 遍歷所有的事件處理器工廠，通過 Factory 獲得事件處理器，調用 Handler 的 HandleEventAsync 方法。
        foreach (var handlerFactory in handlerFactories.EventHandlerFactories)
        {
            await TriggerHandlerAsync(handlerFactory, handlerFactories.EventType, eventData, exceptions);
        }
    }

    // 如果類型繼承了 IEventDataWithInheritableGenericArgument 介面，那麼會檢測泛型參數是否有父類。
    // 如果有父類，則會使用當前的事件數據，為其父類發佈一個事件。
    if (eventType.GetTypeInfo().IsGenericType &&
        eventType.GetGenericArguments().Length == 1 &&
        typeof(IEventDataWithInheritableGenericArgument).IsAssignableFrom(eventType))
    {
        var genericArg = eventType.GetGenericArguments()[0];
        var baseArg = genericArg.GetTypeInfo().BaseType;
        if (baseArg != null)
        {
            // 構造基類的事件類型，使用當前一樣的泛型定義，只是泛型參數使用基類。
            var baseEventType = eventType.GetGenericTypeDefinition().MakeGenericType(baseArg);
            // 構建類型的構造參數。
            var constructorArgs = ((IEventDataWithInheritableGenericArgument)eventData).GetConstructorArgs();
            // 通過事件類型和構造參數，構造一個新的事件數據實例。
            var baseEventData = Activator.CreateInstance(baseEventType, constructorArgs);
            // 發佈父類的同類事件。
            await PublishAsync(baseEventType, baseEventData);
        }
    }
}

在上述代碼內部，都還沒有真正執行事件處理器，真正的事件處理器執行程式是在下麵的方法進行執行的。ABP vNext 通過引入 IEventDataWithInheritableGenericArgument 介面，實現了 類型繼承事件 的觸發，該介面提供了一個 GetConstructorArgs() 方法定義，方便後面生成構造參數。

例如有一個基礎事件叫做 EntityEventData<Student>，如果 Student 繼承自 Person，那麼在觸發該事件的時候，也會發佈一個 EntityEventData<Person> 事件。

2.3.3 事件處理器的執行

真正事件處理器的執行，是通過下麵的方法實現的，大概思路就是通過事件匯流排工廠，構建了事件處理器的實例。通過反射，調用事件處理器的 HandleEventAsync() 方法。如果在處理過程當中，出現了異常，則將異常數據放置在 List<Exception> 集合當中。

protected virtual async Task TriggerHandlerAsync(IEventHandlerFactory asyncHandlerFactory, Type eventType, object eventData, List<Exception> exceptions)
{
    using (var eventHandlerWrapper = asyncHandlerFactory.GetHandler())
    {
        try
        {
            // 獲得事件處理器的類型。
            var handlerType = eventHandlerWrapper.EventHandler.GetType();

            // 判斷事件處理器是本地事件還是分散式事件。
            if (ReflectionHelper.IsAssignableToGenericType(handlerType, typeof(ILocalEventHandler<>)))
            {
                // 獲得方法定義。
                var method = typeof(ILocalEventHandler<>)
                    .MakeGenericType(eventType)
                    .GetMethod(
                        nameof(ILocalEventHandler<object>.HandleEventAsync),
                        new[] { eventType }
                    );

                // 使用工廠創建的實例調用方法。
                await (Task)method.Invoke(eventHandlerWrapper.EventHandler, new[] { eventData });
            }
            else if (ReflectionHelper.IsAssignableToGenericType(handlerType, typeof(IDistributedEventHandler<>)))
            {
                var method = typeof(IDistributedEventHandler<>)
                    .MakeGenericType(eventType)
                    .GetMethod(
                        nameof(IDistributedEventHandler<object>.HandleEventAsync),
                        new[] { eventType }
                    );

                await (Task)method.Invoke(eventHandlerWrapper.EventHandler, new[] { eventData });
            }
            else
            {
                // 如果都不是，則說明類型不正確，拋出異常。
                throw new AbpException("The object instance is not an event handler. Object type: " + handlerType.AssemblyQualifiedName);
            }
        }
        // 捕獲到異常都統一添加到異常集合當中。
        catch (TargetInvocationException ex)
        {
            exceptions.Add(ex.InnerException);
        }
        catch (Exception ex)
        {
            exceptions.Add(ex);
        }
    }
}

2.4 分散式事件匯流排

分散式事件匯流排的實現都存放在 Volo.Abp.EventBus.RabbitMQ，該項目的代碼比較少，由三個文件構成。

在 RabbitMQ 模塊的內部，只幹了兩件事情。首先從 JSON 配置文件當中，獲取 AbpRabbitMqEventBusOptions 配置的三個參數，然後解析 RabbitMqDistributedEventBus 實例，並調用初始化方法 (Initialize())。

[DependsOn(
    typeof(AbpEventBusModule),
    typeof(AbpRabbitMqModule))]
public class AbpEventBusRabbitMqModule : AbpModule
{
    public override void ConfigureServices(ServiceConfigurationContext context)
    {
        var configuration = context.Services.GetConfiguration();

        // 從配置文件讀取配置。
        Configure<AbpRabbitMqEventBusOptions>(configuration.GetSection("RabbitMQ:EventBus"));
    }

    public override void OnApplicationInitialization(ApplicationInitializationContext context)
    {
        // 調用初始化方法。
        context
            .ServiceProvider
            .GetRequiredService<RabbitMqDistributedEventBus>()
            .Initialize();
    }
}

2.4.1 分散式事件匯流排的初始化

public void Initialize()
{
    // 創建一個消費者，並配置交換器和隊列。
    Consumer = MessageConsumerFactory.Create(
        new ExchangeDeclareConfiguration(
            AbpRabbitMqEventBusOptions.ExchangeName,
            type: "direct",
            durable: true
        ),
        new QueueDeclareConfiguration(
            AbpRabbitMqEventBusOptions.ClientName,
            durable: true,
            exclusive: false,
            autoDelete: false
        ),
        AbpRabbitMqEventBusOptions.ConnectionName
    );

    // 消費者在消費消息的時候，具體的執行邏輯。
    Consumer.OnMessageReceived(ProcessEventAsync);

    // 調用基類的方法，自動訂閱對應的事件。
    SubscribeHandlers(AbpDistributedEventBusOptions.Handlers);
}

2.4.2 分散式事件的訂閱

在定義分散式事件的時候，我們必須使用 EventNameAttribute 為事件聲明路由鍵。

public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
{
    var handlerFactories = GetOrCreateHandlerFactories(eventType);

    if (factory.IsInFactories(handlerFactories))
    {
        return NullDisposable.Instance;
    }

    handlerFactories.Add(factory);

    if (handlerFactories.Count == 1) //TODO: Multi-threading!
    {
        // 為消費者綁定一個路由鍵，在收到對應的事件時，就會觸發之前綁定的方法。
        Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType));
    }

    return new EventHandlerFactoryUnregistrar(this, eventType, factory);
}

訂閱的時候，除了 Consumer.BindAsync() 以外，基本流程和本地事件匯流排基本一致。

2.4.3 分散式事件的發佈

分散式事件匯流排一樣重寫了發佈方法，內部首先使用 IRabbitMqSerializer 序列化器 (基於 JSON.NET) 將事件數據進行序列化，然後將消息投遞出去。

public override Task PublishAsync(Type eventType, object eventData)
{
    var eventName = EventNameAttribute.GetNameOrDefault(eventType);
    // 序列化事件數據。
    var body = Serializer.Serialize(eventData);

    // 創建一個通道用於通訊。
    using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
    {
        channel.ExchangeDeclare(
            AbpRabbitMqEventBusOptions.ExchangeName,
            "direct",
            durable: true
        );
        
        // 更改投遞模式為持久化模式。
        var properties = channel.CreateBasicProperties();
        properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent;

        // 發佈一個新的事件。
        channel.BasicPublish(
            exchange: AbpRabbitMqEventBusOptions.ExchangeName,
            routingKey: eventName,
            mandatory: true,
            basicProperties: properties,
            body: body
        );
    }

    return Task.CompletedTask;
}

2.4.4 分散式事件的執行

執行邏輯都存放在 ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) 方法內部，基本就是監聽到指定的消息，首先反序列化消息，調用父類的 TriggerHandlersAsync 去執行具體的事件處理器。

private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea)
{
    var eventName = ea.RoutingKey;
    var eventType = EventTypes.GetOrDefault(eventName);
    if (eventType == null)
    {
        return;
    }

    var eventData = Serializer.Deserialize(ea.Body, eventType);

    await TriggerHandlersAsync(eventType, eventData);
}

三、總結

ABP vNext 為我們實現了比較完善的本地事件匯流排，和基於 RabbitMQ 的分散式事件匯流排。在平時開發過程中，我們本地事件匯流排的使用頻率應該還是比較高，而分散式事件匯流排目前仍處於一個半成品，很多高級特性還沒實現，例如重試策略等。所以分散式事件匯流排要使用的話，建議使用較為成熟的 CAP 庫替代 ABP vNext 的分散式事件匯流排。