從前面文章可以看出，訊息匯流排是EDA(事件驅動架構)與微服務架構的核心部件，沒有訊息匯流排，就無法很好的實現微服務之間的解耦與通訊。通常我們可以利用現有成熟的訊息代理產品或雲平臺提供的訊息服務來構建自己的訊息匯流排；也可以自己完全寫一個訊息代理產品，然後基於它構建自己的訊息匯流排。通常我們不用重複造輪子（除非公司有特殊的要求，比如一些大型網際網路公司考慮到自主可控的白盒子），可以利用比如像RabbitMq這樣成熟的訊息代理產品作為訊息匯流排的底層支援。

RabbitMq核心元件解釋：

Connection:訊息的傳送方或訂閱方通過它連線到RabbitMq伺服器。

Channel：訊息的傳送方或訂閱方通過Connection連線到RabbitMq伺服器後，通過Channel建立會話通道。

Exchange：訊息的傳送方向Exchange傳送訊息，通過RabbitMq伺服器中Exchange與Queue的繫結關係，Exchange會將訊息路由到匹配的Queue中。

Queue：訊息的承載者，訊息的傳送者的訊息最終通過Exchange路由到匹配的Queue，訊息的接收者從Queue接收訊息並進行處理。

Exchange模式：在訊息傳送到Exchange時，需要路由到匹配的Queue中，至於如何路由，則是由Exchange模式決定的。

1.Direct模式：特定的路由鍵(訊息型別)轉發到該Exchange的指定Queue中。

2.Fanout模式：傳送到該Exchange的訊息，被同時傳送到Exchange下繫結的所有Queue中。

3.Topic模式：具有某種特徵的訊息轉發到該Exchange的指定Queue中。

我們最常見的使用是Direct模式，如果訊息要被多個消費者消費，則可以使用Fanout模式。

實現基於RabbitMq的訊息匯流排：
我們首先需要安裝Erlang與RabbitMq到伺服器上，然後就可以進行基於RabbitMq的訊息匯流排的開發了，開發的總體思路與步驟如下：

1.首先建立一個專案作為訊息匯流排，然後引入Rabbitmq.Client 這個nuget包，這樣就有了RabbitMq開發的支援。

2.前面實現了基本的訊息匯流排，所有基於RabbitMq的訊息匯流排是從它繼承下來的，並需要傳入特定的引數到訊息匯流排的建構函式中：

 public RabbitMqEB(IConnectionFactory connectionFactory,IEventHandlerExecutionContext context,
            string exchangeName,string exchangeType,string queueName,int publisherorconsumer,
            bool autoAck = true) : base(context)
        {
            this.connectionFactory = connectionFactory;
            this.connection = this.connectionFactory.CreateConnection();
            this.exchangeName = exchangeName;
            this.exchangeType = exchangeType;
            this.autoAck = autoAck;
            this.queueName = queueName;
            if (publisherorconsumer == 2)
            {
                this.channel = CreateComsumerChannel();
            }
        }

connectionFactory：RabbitMq.Client中的型別，用於與RabbitMq伺服器建立連線時需要使用的物件。

context：訊息與訊息處理器之間的關聯關係的物件。

exchangeName：生產者或消費者需要連線到的Exchange的名字。

exchangeType：前面所描述的Exchange模式。

queueName：生產者或消費者傳送或接收訊息時的Queue的名字。

publisherorconsumer：指定連線到訊息匯流排的元件是訊息匯流排的生產者還是消費者，消費者和生產者會有不同，消費者(publisherorconsumer==2)會構建一個消費通道，用於從Queue接收訊息並呼叫父類的ieventHandlerExecutionContext的HandleAsync方法來處理訊息。

3.建立到RabbitMq的連線： 

//判斷是否已經建立了連線
public bool IsConnected
        {
            get { return this.connection != null && this.connection.IsOpen; }
        }
        public bool TryConnect()
        {
           //出現連線異常時的重試策略,通常通過第三方nuget包實現重試功能,這裡出現連線異常時，每個1秒重試一次，共重試5次
            var policy = RetryPolicy.Handle<SocketException>().Or<BrokerUnreachableException>()
                .WaitAndRetry(5, p => TimeSpan.FromSeconds(1),(ex,time)=> {
                    //記錄錯誤日誌
                });
            policy.Execute(() =>
            {
                //建立RabbitMq Server的連線
                this.connection = this.connectionFactory.CreateConnection();
            });
            if (IsConnected)
            {
                return true;
            }
            return false;
        }

 4.建立消費者通道：

private IModel CreateComsumerChannel()
        {
            if (!IsConnected)
            {
                TryConnect();
            }
            var channel = this.connection.CreateModel();            
            channel.ExchangeDeclare(exchange: exchangeName, type: exchangeType, durable: true);
            channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false,
                arguments: null);
            var consumer = new EventingBasicConsumer(channel);
          //消費者接收到訊息的處理
            consumer.Received += async (model, ea) =>
                {
                    var eventbody = ea.Body;
                    var json = Encoding.UTF8.GetString(eventbody);
                    var @event = (IEvent)JsonConvert.DeserializeObject(json);
                   //呼叫關聯物件中訊息對應的處理器的處理方法
                    await this.eventHandlerExecutionContext.HandleAsync(@event);
                   //向會話通道確認此訊息已被處理
                    channel.BasicAck(ea.DeliveryTag, multiple: false);
                };
            channel.BasicConsume(queue: this.queueName, autoAck: false, consumer: consumer);
            
            channel.CallbackException += (sender, ea) =>
            {
                this.channel.Dispose();
                this.channel = CreateComsumerChannel();
             };
            return channel;
        }

5.對生產者釋出訊息到交換機佇列的支援：

 public override void Publish<TEvent>(TEvent @event)
        {
            if (!IsConnected)
            {
                TryConnect();
            }
            using(var channel = this.connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: exchangeName, type: exchangeType, durable: true);
                var message = JsonConvert.SerializeObject(@event);
                var body = Encoding.UTF8.GetBytes(message);
              //釋出到交換機，根據交換機與佇列的繫結以及交換機模式，最終釋出到指定的佇列中
                channel.BasicPublish(this.exchangeName, @event.GetType().FullName,null, body);
            }
        }

6.對訂閱者從交換機佇列中訂閱訊息的支援：

 public override void Subscribe<TEvent, TEventHandler>()
        {
           //註冊接收到的訊息型別到訂閱方的處理器之間的關係
            if (!this.eventHandlerExecutionContext.IsRegisterEventHandler < TEvent,TEventHandler>()){
                this.eventHandlerExecutionContext.RegisterEventHandler<TEvent, TEventHandler>();
              //消費者進行佇列繫結
                this.channel.QueueBind(this.queueName, this.exchangeName, typeof(TEvent).FullName);
            }
        }

從上面的6個步驟，我們基本上就完成了基於RabbitMq訊息匯流排的基本功能，這裡需要說明的是，上述程式碼只是演示，在實際生產環境中，不能直接使用以上程式碼，還需要小心的重構此程式碼以保證可靠性與效能。

QQ討論群：309287205 

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