.Net Core 5.x Api開發筆記 -- 消息隊列RabbitMQ實現事件總線EventBus(一)

本文主要圍繞RabbitMQ消息隊列和EventBus事件總線做筆記，其中有很多自己的理解和觀點，有不對之處還望大神指出，我也學習一下。

1，消息隊列

什么是消息隊列就不說了，這里只介紹為什么用它！！！

RabbitMQ提供了可靠的消息機制、跟蹤機制和靈活的消息路由，支持消息集群和分布式部署。適用於排隊算法、秒殺活動、消息分發、異步處理、數據同步、處理耗時任務、CQRS等應用場景。它的以上好處在高並發等三高場景是十分必要的。

前提：使用RabbitMQ必須考慮高可用性

1、高可用：如果使用消息隊列，基本要配合集群的，因為如果MQ服務器崩了，那就整個服務災難了。

2、數據安全：必須保證數據不能丟失，也就是要考慮好最終一致性，做好補償機制。

3、合理的消費。

2，事件總線

事件總線是對發布-訂閱模式的一種實現。它是一種集中式事件處理機制，允許不同的組件之間進行彼此通信而又不需要相互依賴，達到一種解耦的目的。

說人話就是：集中對消息隊列中處理的事件進行訂閱和綁定管理。

實現事件總線的關鍵是：

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

public class EventBus
{
    /// <summary>
    /// EventBus 單例
    /// </summary>
    public static EventBus Default { get; private set; }

    /// <summary>
    /// 事件與事件的處理程序映射
    /// </summary>
    private static readonly ConcurrentDictionary<Type, List<Type>> m_EventAndHandlersMapping = new ConcurrentDictionary<Type, List<Type>>();

    /// <summary>
    /// 加鎖對象
    /// </summary>
    private object m_lockObj;

    static EventBus()
    {
        Default = new EventBus();
    }

    private EventBus()
    {
        m_lockObj = new object();
    }

    /// <summary>
    /// 注冊事件源和事件處理程序
    /// </summary>
    /// <typeparam name="TEvent">事件類型</typeparam>
    /// <typeparam name="THandler">處理程序類型</typeparam>
    public void RegisterHandler<T, THandler>()
        where T : EventBase
        where THandler : IEventHandler<T>
    {
        AddMapping(typeof(T), typeof(THandler));
    }

    /// <summary>
    /// 新增映射 事件源+事件處理程序
    /// </summary>
    /// <param name="eventType"></param>
    /// <param name="handlerType"></param>
    private void AddMapping(Type eventType, Type handlerType)
    {
        var handlers = m_EventAndHandlersMapping.GetOrAdd(eventType, type => new List<Type>());
        lock (m_lockObj)
        {
            if (!handlers.Contains(handlerType))
            {
                handlers.Add(handlerType);
            }
        }
    }
  
    /// <summary>
    /// 異步觸發事件
    /// </summary>
    /// <param name="e"></param>
    /// <returns></returns>
    public async Task TriggerAsync<TEvent>(TEvent e)
        where TEvent : EventBase
    {
        List<Type> value;
        if (m_EventAndHandlersMapping.TryGetValue(e.GetType(), out value))
        {
            foreach (var handler in value.OrderBy(p => p.Name))
            {
                var instance = IocManager.ServiceProvider.GetService(handler) as IEventHandler<TEvent>;
                await instance?.HandleEvent(e);
            }
        }
    }
}

上邊就是一個簡單的事件總線管理類，主要任務就是對一個事件源與事件處理的映射，包括事件處理程序的觸發。

3，事件源+事件處理

為了集中對事件進行處理，就需要進一步提取基類來處理，如下所示：

/// <summary>
/// 事件處理程序基類
/// </summary>
public interface IBaseEventHandler
{
}

/// <summary>
/// 事件處理程序
/// 解釋：where T : EventBase 表示傳入的類型必須繼承 EventBase類
/// </summary>
public interface IEventHandler<T> : IBaseEventHandler
    where T : EventBase
{
    /// <summary>
    /// 處理事件
    /// </summary>
    /// <param name="e">事件傳遞的數據</param>
    Task HandleEvent(T e);
}

然后還需要聲明一個事件基類來傳遞數據信息，這里聲明一個強類型的帶數據的事件

/// <summary>
/// 事件基類
/// </summary>
public abstract class EventBase
{
    protected EventBase()
        : this(null)
    { 
    }
    protected EventBase(object source)
    {
        this.Source = source;
    }
    /// <summary>
    /// 引發事件的源
    /// </summary>
    public object Source { get; private set; }
}

/// <summary>
/// 聲明一個強類型的帶數據的事件 傳遞數據
/// </summary>
public class EventWithData<TData> : EventBase
{

    /// <summary>
    /// 初始化 EventBase
    /// </summary>
    /// <param name="data">傳遞的數據</param>
    public EventWithData(TData data)
    {
        this.Data = data;
    }

    /// <summary>
    /// 事件傳遞的數據
    /// </summary>
    public TData Data { get; private set; }

    /// <summary>
    /// 創建 EventWithData'T'
    /// </summary>
    /// <param name="data">數據</param>
    /// <returns></returns>
    public static EventWithData<TData> New(TData data)
    {
        return new EventWithData<TData>(data);
    }
}

到這里關於事件總線的設計就初步完成了，接下來是關於消息隊列 發布+訂閱的設計。