從上一篇文章大家可以看出，實現一個自己的訊息匯流排框架是非常重要的內容，訊息匯流排可以將界限上下文之間進行解耦，也可以為大併發訪問提供必要的支援。

訊息匯流排的作用：

1.界限上下文解耦：在DDD第一波文章中，當更新了訂單資訊後，我們通過呼叫經銷商界限上下文的領域模型和倉儲，進行了經銷商資訊的更新，這造成了耦合。通過一個訊息匯流排，可以在訂單界限上下文的WebApi服務(來源微服務-生產者)更新了訂單資訊後，釋出一個事件訊息到訊息匯流排的某個佇列中，經銷商界限上下文的WebApi服務(消費者)訂閱這個事件訊息，然後交給自己的Handler進行訊息處理，更新自己的經銷商資訊。這樣就實現了訂單界限上下文與經銷商界限上下文解耦。

2.大併發支援：可以通過訊息匯流排進一步提升下單的效能。我們可以將使用者下單的操作直接交給一個下單命令WebApi接收，下單命令WebApi接收到命令後，直接丟給一個訊息匯流排的佇列，然後立即給前端返回下單結果。這樣使用者就不用等待後續的複雜訂單業務邏輯，加快速度。後續訂單的一系列處理交給訊息的Handler進行後續的處理與訊息的進一步投遞。

訊息匯流排設計重點：

1.定義訊息(事件)的介面：所有需要投遞與處理的訊息，都從這個訊息介面繼承，因為需要約束訊息中必須包含的內容，比如訊息的ID、訊息產生的時間等。

public interface IEvent
    {
        Guid Id { 
 
get; set; }
        DateTime CreateDate { get; set; }
    }

2.定義訊息(事件)處理器介面：當訊息投遞到訊息匯流排佇列中後，一定有消費者WebApi接收並處理這個訊息，具體的處理方法邏輯在訂閱方處理器中實現，這裡先需要定義處理器的介面，便於在訊息匯流排框架中使用。

public interface IEventHandler
    {
        Task<bool> HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent;
    }

從上面程式碼可以看出，訊息(事件)處理器處理的型別就是從IEvent介面繼承的訊息類。

3.定義訊息(事件)與訊息(事件)處理器關聯介面：一種型別的訊息被投遞後，一定要在訂閱方找到這種訊息的處理器進行處理，所以一定要定義二者的關聯介面，這樣才能將訊息與訊息處理器對應起來，才能實現訊息被訂閱後的處理。

 public interface IEventHandlerExecutionContext
    {
        void RegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent
            where TEventHandler : IEventHandler;
        bool IsRegisterEventHandler<TEvent, TEventHandler>() where TEvent : IEvent
            where TEventHandler : IEventHandler;
        Task HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent;
    }

RegisterEventHandler方法就是建立訊息與訊息處理器的關聯，這個方法其實是在訂閱方使用，訂閱方告訴訊息匯流排，什麼樣的訊息應該交給我的哪個處理器進行處理。

IsRegisterEventHandler方法是判斷訊息與處理器之間是否已經存在關聯。

HandleAsync方法是通過查詢到訊息對應的處理器後，然後呼叫處理器自己的Handle方法進行訊息的處理.

4.定義訊息釋出、訂閱與訊息匯流排介面：訊息匯流排至少要支援兩個功能，一個是生產者能夠釋出訊息到我的訊息匯流排，另一個是訂閱方需要能夠從我這個訊息匯流排訂閱訊息。

 public interface IEventPublisher
    {
        void Publish<TEvent>(TEvent @event) where TEvent : IEvent;
    }

從上面程式碼可以看出，生產者釋出的訊息仍然要從IEvent繼承的型別。

 public interface IEventSubscriber
    {
        void Subscribe<TEvent, TEventHandler>() where TEvent : IEvent
            where TEventHandler : IEventHandler;
    }

上面程式碼是訂閱方用於從訊息匯流排訂閱訊息，從程式碼中可以看出，它的最終的實現其實就是建立訊息與處理器之間的關聯。

 public interface IEventBus:IEventPublisher,IEventSubscriber
    {
    }

訊息(事件)匯流排從兩個介面繼承下來，同時支援訊息的釋出與訊息的訂閱。

5.實現事件基類：上面已經訂閱了訊息(事件)的介面，這裡來實現事件的基類，其實就是實現訊息ID與產生的時間：

  public class BaseEvent : IEvent
    {
        public Guid Id { get; set; }
        public DateTime CreateDate { get; set; }
        public BaseEvent()
        {
            this.Id = Guid.NewGuid();
            this.CreateDate = DateTime.Now;
        }
    }

6.實現訊息匯流排基類：訊息匯流排底層的依賴可以是各種訊息代理產品，比如RabbitMq、Kafaka或第三方雲平臺提供的訊息代理產品，通常我們要封裝這些訊息代理產品。在封裝之前，我們需要定義頂層的訊息匯流排基類實現，主要的目的是未來依賴於它的具體實現可替換，另外也將訊息與訊息處理器的關聯介面傳遞進來，便於訂閱方使用。

public abstract class BaseEventBus : IEventBus
    {
        protected readonly IEventHandlerExecutionContext eventHandlerExecutionContext;
        protected BaseEventBus(IEventHandlerExecutionContext eventHandlerExecutionContext)
        {
            this.eventHandlerExecutionContext = eventHandlerExecutionContext;
        }
        public abstract void Publish<TEvent>(TEvent @event)
            where TEvent : IEvent;
        public abstract void Subscribe<TEvent, TEventHandler>()
            where TEvent : IEvent
            where TEventHandler : IEventHandler;
    }

7.實現訊息與處理器關聯：訊息必須與處理器關聯，訂閱方收到特定型別的訊息後，才知道交給哪個處理器處理。

 public class EventHandlerExecutionContext : IEventHandlerExecutionContext
    {
        private readonly IServiceCollection registry;
        private readonly IServiceProvider serviceprovider;
        private Dictionary<Type, List<Type>> registrations = new Dictionary<Type, List<Type>>();
        public EventHandlerExecutionContext(IServiceCollection registry,Func<IServiceCollection,
            IServiceProvider> serviceProviderFactory = null)
        {
            this.registry = registry;
            this.serviceprovider = this.registry.BuildServiceProvider();
        }

       //查詢訊息關聯的處理器，然後呼叫處理器的處理方法
        public async Task HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent
        {
            var eventtype = @event.GetType();
            if(registrations.TryGetValue(eventtype,out List<Type> handlertypes) && handlertypes.Count > 0)
            {
                using(var childscope = this.serviceprovider.CreateScope())
                {
                    foreach(var handlertype in handlertypes)
                    {
                        var handler = Activator.CreateInstance(handlertype) as IEventHandler;
                        await handler.HandleAsync(@event);
                    }
                }
            }
        }

       //判斷訊息與處理器之間是否有關聯
        public bool IsRegisterEventHandler<TEvent, TEventHandler>()
            where TEvent : IEvent
            where TEventHandler : IEventHandler
        {
            if(registrations.TryGetValue(typeof(TEvent),out List<Type> handlertypelist))
            {
                return handlertypelist != null && handlertypelist.Contains(typeof(IEventHandler));
            }
            return false;
        }
       
      //將訊息與處理器關聯起來，可以在記憶體中建立關聯，也可以建立在資料庫單獨表中
        public void RegisterEventHandler<TEvent, TEventHandler>()
            where TEvent : IEvent
            where TEventHandler : IEventHandler
        {
            Utils.DictionaryRegister(typeof(TEvent), typeof(TEventHandler), registrations);
        }
    }

上面我們基本上就將訊息匯流排的架子搭建起來了，也實現了基本的功能，下一章我們基於它來實現RabbitMq的訊息匯流排。

QQ討論群：309287205 

微服務實戰視訊請關注微信公眾號：