[Abp vNext 原始碼分析] - 13. 本地事件總線與分布式事件總線 (Rabbit MQ)

一、簡要介紹

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 個不同的實作，下表分別說明它們的應用場景，

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

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 = https://www.cnblogs.com/myzony/p/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 = https://www.cnblogs.com/myzony/p/Serializer.Deserialize(ea.Body, eventType);

    await TriggerHandlersAsync(eventType, eventData);
}

三、總結

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

四、其他

360 大病救助 : 在這里向大家求助一下，病人是我親戚，情況屬實，對于他們家庭來說，經濟壓力很大，希望大家能幫助或轉發一下，謝謝大家，

標籤：.NET Core

上一篇：C# 添加文本、圖片到PDF檔案（基于Spire.Cloud.SDK for .NET）

下一篇：Asp.net Core 3.0 Identity 使用smtp賬戶確認和密碼恢復