C#利用RabbitMQ實現點對點消息傳輸

RabbitMQ做為消息代理，負責接收和轉發消息，可以將RabbitMQ比喻為一個郵筒、一個郵局和一個郵遞員。本文主要以一個簡單的小例子，簡述RabbitMQ實現消息傳輸的相關內容，僅供學習分享使用，如有不足之處，還請指正。

消息隊列模型

所有 MQ 產品從模型抽象上來說都是一樣的過程：
消費者（consumer）訂閱某個隊列。生產者（producer）創建消息，然后發布到隊列（queue）中，最后將消息發送到監聽的消費者。

RabbitMQ設置

RabbitMQ是通過交換機將消息轉發到對應隊列，所以隊列需要和交換機進行綁定。本例將隊列綁定到默認的amq.direct交換機，並設置Routing key，如下圖所示：

RabbitMQ動態庫安裝

通過NuGet包管理器進行安裝RabbitMQ.Client，如下所示：

RabbitMQ.Client相關知識點

• ConnectionFactory：構造一個實例，主要創建連接。
• IConnection：表示一個基於AMQP協議的連接。
• IModel：表示一個RabbitMQ通道，可用於聲明一個隊列，然后開始消費。
• EventingBasicConsumer：基於獨立事件監聽的基礎消費者，可以監聽並接收消息。
• 生產者基本步驟：1. 創建連接 2. 基於連接創建通道 3. 基於通道聲明隊列，4. 開始生產並發布消息
• 消費者基本步驟：1. 創建連接 2. 基於連接創建通道 3. 基於通道聲明隊列，4. 創建消費者，5. 綁定通道和消費者，並開始消費

示例效果圖

本例主要有一個生產者，一個消費者，通過消息隊列進行消息轉發和接收。

生產者負責消息發送，如下圖所示：

消費者負責消息接收，如下圖所示：

核心代碼

代碼結構：主要包括生產者，消費者，公共基礎代碼，如下所示：

RabbitMqHelper主要創建連接，如下所示：

    public class RabbitMqHelper
    {
        
        /// <summary>
        /// 創建連接
        /// </summary>
        /// <returns></returns>
        public IConnection GetConnection()
        {
            try
            {
                var factory = new ConnectionFactory()
                {
                    HostName = "127.0.0.1",
                    Port = 5672,
                    UserName = "guest",
                    Password = "guest",
                    VirtualHost = "/ShortMsgHost"
                };
                var conn = factory.CreateConnection();
                return conn;
            }
            catch (Exception ex) {
                throw ex;
            }
        }

   

    }

RabbmitMqSendHelper用於發送消息，如下所示：

    public class RabbmitMqSendHelper : RabbitMqHelper
    {
        /// <summary>
        /// 發送消息
        /// </summary>
        /// <param name="msg"></param>
        /// <returns></returns>
        public bool SendMsg(string msg)
        {
            try
            {
                using (var conn = GetConnection())
                {
                    using (var channel = conn.CreateModel())
                    {
                        channel.QueueDeclare(queue: "ShortMsgQueue",
                                     durable: true,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                        var body = Encoding.UTF8.GetBytes(msg);

                        channel.BasicPublish(exchange: "amq.direct",
                                             routingKey: "ShortMsgKey",
                                             basicProperties: null,
                                             body: body);

                        //Console.WriteLine(" [x] Sent {0}", message);
                    };
                };
                return true;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }

RabbitMqReceiveHelper主要用於接收信息，如下所示：

    public class RabbitMqReceiveHelper : RabbitMqHelper
    {
        public RabbitMqReceiveEventHandler OnReceiveEvent;

        private IConnection conn;

        private IModel channel;

        private EventingBasicConsumer consumer;

        public bool StartReceiveMsg()
        {
            try
            {
                conn = GetConnection();

                channel = conn.CreateModel();

                channel.QueueDeclare(queue: "ShortMsgQueue",
                                durable: true,
                                exclusive: false,
                                autoDelete: false,
                                arguments: null);

                consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body.ToArray();
                    var message = Encoding.UTF8.GetString(body);
                    //Console.WriteLine(" [x] Received {0}", message);
                    if (OnReceiveEvent != null)
                    {
                        OnReceiveEvent(message);
                    }
                };
                channel.BasicConsume(queue: "ShortMsgQueue",
                                        autoAck: true,
                                        consumer: consumer);
                return true;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }

關於RabbitMQ的基礎知識介紹，可參考前幾篇博文。

備注

浣溪沙·堤上游人逐畫船

歐陽修 〔宋代〕

堤上游人逐畫船，拍堤春水四垂天。綠楊樓外出秋千。
白發戴花君莫笑，六幺催拍盞頻傳。人生何處似尊前！