.NET中RabbitMQ的使用

時間 2019-11-26

原文原文鏈接

概述html

　　MQ全稱爲Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通訊方法。RabbitMQ是一個在AMQP基礎上完整的，可複用的企業消息系統。他遵循Mozilla Public License開源協議。AMQP(高級消息隊列協議) 是一個異步消息傳遞所使用的應用層協議規範，做爲線路層協議，而不是API（例如JMS），AMQP 客戶端可以無視消息的來源任意發送和接受信息。AMQP的原始用途只是爲金融界提供一個能夠彼此協做的消息協議，而如今的目標則是爲通用消息隊列架構提供通用構建工具。所以，面向消息的中間件（MOM）系統，例如發佈/訂閱隊列，沒有做爲基本元素實現。AMQP當中有四個概念很是重要（一個虛擬主機持有一組交換機、隊列和綁定）：數據庫

virtual host，虛擬主機
exchange，交換機
queue，隊列
binding，綁定

Window下安裝RabbbitMQ服務器

文件下載安裝架構

Rabbit MQ 是創建在強大的Erlang OTP平臺上，所以安裝Rabbit MQ的前提是安裝Erlang。經過下面兩個鏈接下載安裝3.2.3 版本：異步

下載並安裝 Erlang OTP For Windows (vR16B03)
運行安裝 Rabbit MQ Server Windows Installer (v3.2.3)

默認安裝的Rabbit MQ 監聽端口是5672。先安裝Erlang OTP後安裝RabbitMQ，安裝方式默認便可，RabbitMQ能夠勾選安裝後臺服務、服務啓動和中止等操做。工具

激活Rabbit MQ's Management Pluginfetch

使用Rabbit MQ 管理插件，能夠更好的可視化方式查看Rabbit MQ 服務器實例的狀態，打開CMD命令，cd到安裝目錄（..\rabbitmq_server-3.2.3\sbin）下，輸入下面的命令激活：spa

rabbitmq-plugins enable rabbitmq_management

要重啓服務才能生效，能夠執行插件

net stop RabbitMQ && net start RabbitMQ

輸入網址，打開監控頁面： http://localhost:15672 (默認帳號和密碼：guest 和guest)3d

配置RabbitMQ用戶權限

RabbitMQ是存在用戶權限的，默認是guest 密碼也是guest，隸屬於Administrator管理員下。現須要配置新用戶和權限，繼續打開CMD命令，cd到安裝目錄sbin下：

用戶操做指令：

::查詢服務狀態
rabbitmqctl status


::列舉虛擬主機列表
rabbitmqctl list_vhosts
::列舉用戶列表
rabbitmqctl list_users

:: 添加用戶和密碼
rabbitmqctl  add_user  hao  abc123

:: 設置權限   
rabbitmqctl  set_permissions  yy  ".*"  ".*"  ".*"

:: 分配用戶組
rabbitmqctl  set_user_tags yy administrator

:: 刪除guest用戶
rabbitmqctl delete_user guest
::修改用戶密碼
rabbitmqctl change_password {username}  {newpassowrd}

像ADO.Net的五大對象也是，操做數據庫先進行鏈接connection,而後使用command過濾出要選擇的數據，使用DataReader或DataSet、DataAdapter,在RabbitMQ的使用當中也有基本固定的步驟。

1、生產者

1.建立鏈接connection:無論是生產者仍是消費者都須要先於RabbitMQ服務器鏈接，才能進行數據交換

2.建立通道 Channel:生產者、消費者的消息傳遞是在通道下傳遞的

3.聲明交換器、隊列

4.交換器與隊列進行綁定

5.經過交換器BasicPublish數據到隊列

2、消費者

1.建立鏈接 :和生產者同樣

2.建立通道:和生產者同樣

3.聲明交換器、隊列

4.交換器與隊列進行綁定

5.經過BasicGet方法獲取隊列中的數據

上面1、二是大體的基本步驟，按照大體的步驟來基本不會出現大的問題，其實生產者和消費者的前4個步驟基本同樣，主要是第5個步驟，一個是生產BasicPublish發佈，一個是獲取Get。

3、出現的錯誤

在ConnectionFactory建立鏈接對象時出現上面提到的bug:None of the specified endpoints were reachable.

4、demo

1.首先在生產者端和消費者端引入RabbitMQ

2.生產者端

using RabbitMQ.Client;
using System;
using System.Text;

namespace RabbitMQProduct
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Welcome to RabbitMQ Product!");
            DirectExchangeSendMsg();
            // TopicExchangeSendMsg();
            Console.WriteLine("按任意值，退出程序");
            Console.ReadKey();
        }

        /// <summary>
        /// 鏈接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
        {
            UserName = "howdyadmin",
            Password = "123456",
            Port = 5672,
            VirtualHost = "howdyVirtualHost"
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "howdy.exchange";

        //隊列名稱
        const string QueueName = "howdy.queue";

        /// <summary>
        /// 路由名稱
        /// </summary>
        const string TopExchangeName = "topic.howdy.exchange";

        //隊列名稱
        const string TopQueueName = "topic.howdy.queue";


        /// <summary>
        ///  單點精確路由模式
        /// </summary>
        public static void DirectExchangeSendMsg()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ExchangeName, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);

                    var props = channel.CreateBasicProperties();
                    props.Persistent = true;
                    string vadata = Console.ReadLine();
                    while (vadata != "exit")
                    {
                        var msgBody = Encoding.UTF8.GetBytes(vadata);
                        channel.BasicPublish(exchange: ExchangeName, routingKey: QueueName, basicProperties: props, body: msgBody);
                        Console.WriteLine(string.Format("***發送時間:{0}，發送完成，輸入exit退出消息發送",
                            DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
                        vadata = Console.ReadLine();
                    }
                }
            }
        }
        /// <summary>
        /// topic 模糊匹配模式，符號「#」匹配一個或多個詞，符號「*」匹配很少很多一個詞。所以「log.#」可以匹配到「log.info.oa」，可是「log.*」 只會匹配到「log.error」
        /// </summary>
        public static void TopicExchangeSendMsg()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName,ExchangeType.Topic, durable: false, autoDelete: false, arguments: null);
                    channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
                    //var props = channel.CreateBasicProperties();
                    //props.Persistent = true;
                    string vadata = Console.ReadLine();
                    while (vadata != "exit")
                    {
                        var msgBody = Encoding.UTF8.GetBytes(vadata);
                        channel.BasicPublish(exchange: TopExchangeName, routingKey: TopQueueName, basicProperties: null, body: msgBody);
                        Console.WriteLine(string.Format("***發送時間:{0}，發送完成，輸入exit退出消息發送", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
                        vadata = Console.ReadLine();
                    }
                }
            }
        }
    }
}

上面的代碼分別建立了兩個路由和兩個隊列，一種是DirectExchange，一種是TopicExchange，驗證時須要生產者和消費者使用同一種的ExChange。

3.消費者端

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;

namespace RabbitMQConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Welcome to RabbitMQ Consumer!");
            //DirectAcceptExchange();
             //DirectAcceptExchangeEvent();
            DirectAcceptExchangeTask();
            //TopicAcceptExchange();
            Console.WriteLine("按任意值，退出程序");
            Console.ReadKey();
        }

        /// <summary>
        /// 鏈接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
        {
            HostName = "127.0.0.1",
            UserName = "howdyadmin",
            Password = "123456",
            Port = 5672,
            VirtualHost = "howdyVirtualHost"
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "howdy.exchange";

        //隊列名稱
        const string QueueName = "howdy.queue";

        /// <summary>
        /// 路由名稱
        /// </summary>
        const string TopExchangeName = "topic.howdy.exchange";

        //隊列名稱
        const string TopQueueName = "topic.howdy.queue";


        /// <summary>
        /// 基於時間輪詢的，每隔一段時間獲取一次
        /// </summary>
        public static void DirectAcceptExchange()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(ExchangeName, ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    while (true)
                    {
                        BasicGetResult msgResponse = channel.BasicGet(QueueName, true);
                        if (msgResponse != null)
                        {
                            var msgBody = Encoding.UTF8.GetString(msgResponse.Body);
                            Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                        }

                        System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
                    }
                }
            }
        }
        /// <summary>
        /// 基於事件的，當消息到達時觸發事件，獲取數據
        /// </summary>
        public static void DirectAcceptExchangeEvent()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    //channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    //channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                    };
                    channel.BasicConsume(QueueName,  true, consumer: consumer);
                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }
        /// <summary>
        /// 基於事件的，當消息到達時觸發事件，獲取數據
        /// </summary>
        public static void DirectAcceptExchangeTask()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    //channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
                    channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//告訴broker同一時間只處理一個消息
                    //channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                        int dots = msgBody.Split('.').Length - 1;
                        System.Threading.Thread.Sleep(dots * 1000);
                        //處理完成，告訴Broker能夠服務端能夠刪除消息，分配新的消息過來
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    };
                    //noAck設置false,告訴broker，發送消息以後，消息暫時不要刪除，等消費者處理完成再說
                    channel.BasicConsume(QueueName, false, consumer: consumer);

                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }
        /// <summary>
        /// topic 模糊匹配模式，符號「#」匹配一個或多個詞，符號「*」匹配很少很多一個詞。所以「log.#」可以匹配到「log.info.oa」，可是「log.*」 只會匹配到「log.error」
        /// </summary>
        public static void TopicAcceptExchange()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName, ExchangeType.Topic, durable: false, autoDelete: false, arguments: null);
                    channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
                    channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var msgBody = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(string.Format("***接收時間:{0}，消息內容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
                        int dots = msgBody.Split('.').Length - 1;
                        System.Threading.Thread.Sleep(dots * 1000);
                        Console.WriteLine(" [x] Done");
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    };
                    channel.BasicConsume(TopQueueName,  false, consumer: consumer);

                    Console.WriteLine("按任意值，退出程序");
                    Console.ReadKey();
                }
            }
        }
    }
}

消費者端也是兩個路由兩個隊列，在實現DirectExchange時使用了三種方式，DirectAcceptExchange是基於時間輪詢的，每隔一段時間獲取一次，DirectAcceptExchangeEvent、DirectAcceptExchangeTask是基於事件的，當消息到達時觸發事件，獲取數據。

4.實驗截圖

參考原文地址(感謝分享)：

http://www.javashuo.com/article/p-swpmboxa-hr.html

http://www.cnblogs.com/xibei666/p/5931267.html#3680686