快捷導(dǎo)航

C#利用RabbitMQ實(shí)現(xiàn)點(diǎn)對(duì)點(diǎn)消息傳輸

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

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

消息隊(duì)列模型

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

RabbitMQ設(shè)置

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

RabbitMQ動(dòng)態(tài)庫安裝

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

RabbitMQ.Client相關(guān)知識(shí)點(diǎn)

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

示例效果圖

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

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

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

核心代碼

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

RabbitMqHelper主要?jiǎng)?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;
            }
        }
    }

作者：Alan.hsiang
出處：http://www.cnblogs.com/hsiang/

以上就是C#利用RabbitMQ實(shí)現(xiàn)點(diǎn)對(duì)點(diǎn)消息傳輸?shù)膶?shí)現(xiàn)示例的詳細(xì)內(nèi)容，更多關(guān)于c# 用RabbitMQ實(shí)現(xiàn)點(diǎn)對(duì)點(diǎn)消息傳輸?shù)馁Y料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: