快捷導(dǎo)航

C#利用RabbitMQ實現(xiàn)點對點消息傳輸

更新時間：2021年05月17日 08:46:22 作者：Alan.hsiang

RabbitMQ做為消息代理，負(fù)責(zé)接收和轉(zhuǎn)發(fā)消息，可以將RabbitMQ比喻為一個郵筒、一個郵局和一個郵遞員。本文主要以一個簡單的小例子，簡述RabbitMQ實現(xiàn)消息傳輸?shù)南嚓P(guān)內(nèi)容，僅供學(xué)習(xí)分享使用，如有不足之處，還請指正。

消息隊列模型

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

RabbitMQ設(shè)置

RabbitMQ是通過交換機將消息轉(zhuǎn)發(fā)到對應(yīng)隊列，所以隊列需要和交換機進行綁定。本例將隊列綁定到默認(rèn)的amq.direct交換機，并設(shè)置Routing key，如下圖所示：

RabbitMQ動態(tài)庫安裝

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

RabbitMQ.Client相關(guān)知識點

ConnectionFactory：構(gòu)造一個實例，主要創(chuàng)建連接。
IConnection：表示一個基于AMQP協(xié)議的連接。
IModel：表示一個RabbitMQ通道，可用于聲明一個隊列，然后開始消費。
EventingBasicConsumer：基于獨立事件監(jiān)聽的基礎(chǔ)消費者，可以監(jiān)聽并接收消息。
生產(chǎn)者基本步驟：1. 創(chuàng)建連接 2. 基于連接創(chuàng)建通道 3. 基于通道聲明隊列，4. 開始生產(chǎn)并發(fā)布消息
消費者基本步驟：1. 創(chuàng)建連接 2. 基于連接創(chuàng)建通道 3. 基于通道聲明隊列，4. 創(chuàng)建消費者，5. 綁定通道和消費者，并開始消費

示例效果圖

本例主要有一個生產(chǎn)者，一個消費者，通過消息隊列進行消息轉(zhuǎn)發(fā)和接收。

生產(chǎn)者負(fù)責(zé)消息發(fā)送，如下圖所示：

消費者負(fù)責(zé)消息接收，如下圖所示：

核心代碼

代碼結(jié)構(gòu)：主要包括生產(chǎn)者，消費者，公共基礎(chǔ)代碼，如下所示：

RabbitMqHelper主要創(chuàng)建連接，如下所示：

public class RabbitMqHelper
    {

        /// <summary>
        /// 創(chuàng)建連接
        /// </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用于發(fā)送消息，如下所示：

public class RabbmitMqSendHelper : RabbitMqHelper
    {
        /// <summary>
        /// 發(fā)送消息
        /// </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;
            }
        }
    }