.NET中RabbitMQ的使用

概述

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

1. virtual host，虛擬主機
2. exchange，交換機
3. queue，隊列
4. binding，綁定

Window下安裝RabbbitMQ

文件下載安裝

Rabbit MQ 是建立在強大的Erlang OTP平台上，因此安裝Rabbit MQ的前提是安裝Erlang。通過下面兩個連接下載安裝3.2.3 版本：

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

默認安裝的Rabbit MQ 監聽端口是5672。先安裝Erlang OTP后安裝RabbitMQ，安裝方式默認即可，RabbitMQ可以勾選安裝后台服務、服務啟動和停止等操作。

激活Rabbit MQ's Management Plugin

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

rabbitmq-plugins enable rabbitmq_management

要重啟服務才能生效，可以執行

net stop RabbitMQ && net start RabbitMQ

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

配置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.創建連接connection:不管是生產者還是消費者都需要先於RabbitMQ服務器連接，才能進行數據交換

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

3.聲明交換器、隊列

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

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

二、消費者

1.創建連接 :和生產者一樣

2.創建通道:和生產者一樣

3.聲明交換器、隊列

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

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

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

三、出現的錯誤

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

四、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.實驗截圖

 

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

https://www.cnblogs.com/5ishare/p/6784149.html

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