RabbitMQ.Net 應用（1）

阿新 • • 發佈：2018-02-24

utf8 msg ssa 金融 count 不知道 six clu bind

概述

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

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

　　更多理論性東西可以參考（在Windows上安裝Rabbit MQ 指南），針對隊列的講解相當詳細

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 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}

.NET中RabbitMQ使用

1、Nuget下載RabbitMQ.Client第三方類庫，版本V3.6.5，高版本與.NET Framework 4.5有沖突，RabbitMQ Client地址

2、利用RabbitMQ Clinet類庫編碼(代碼內容有註釋，此處不做詳細解釋，文章後有完整代碼的下載地址)

　　<1>RabbitMQ的direct類型Exchange

　　　Producter發送消息代碼：

        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){
            HostName ="192.168.1.8",UserName="hao",Password="abc123",Port= 5672
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "justin.exchange";

        //隊列名稱
        const string QueueName = "justin.queue";
        public static void DirectExchangeSendMsg()
        {
            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 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();
                    }                   
                }
            }
        }

　　Customer接收消息代碼：

        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {
            HostName = "192.168.1.8", UserName = "hao", Password = "abc123", Port = 5672
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string ExchangeName = "justin.exchange";

        //隊列名稱
        const string QueueName = "justin.queue";
        public static void DirectAcceptExchange()
        {
            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);
                    while (true)
                    { 
                        BasicGetResult msgResponse = channel.BasicGet(QueueName, noAck: 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));
                        }

                        //BasicGetResult msgResponse2 = channel.BasicGet(QueueName, noAck: false);

                        ////process message ...

                        //channel.BasicAck(msgResponse2.DeliveryTag, multiple: false);
                        System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
                    }
                }
            }
        }

　　但是這種處理速度較慢，因為循環線程等待。高效的接收消息的方式可以使用EventingBasicConsumer進行消息接收處理，修改代碼內容如下：

        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, noAck: true, consumer: consumer);

                    //已過時用EventingBasicConsumer代替
                    //var consumer2 = new QueueingBasicConsumer(channel);
                    //channel.BasicConsume(QueueName, noAck: true, consumer: consumer);
                    //var msgResponse = consumer2.Queue.Dequeue(); //blocking
                    //var msgBody2 = Encoding.UTF8.GetString(msgResponse.Body);

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

　　但是有些時候，消費者同一時間沒有能力處理太多的業務，導致分配過來的隊列消息不能及時處理完成，這個時候，我們可以設置BasicQos屬性，告訴Broker同一時間將未處理完成的消息分配其他消費者，所以接收消息的地方需要略做修改，代碼如下：

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);
                Console.WriteLine(" [x] Done");
                //處理完成，告訴Broker可以服務端可以刪除消息，分配新的消息過來
                channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            //noAck設置false,告訴broker，發送消息之後，消息暫時不要刪除，等消費者處理完成再說
            channel.BasicConsume(QueueName, noAck: false, consumer: consumer);

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

　　<2> RabbitMQ的Topic類型Exchange

　　Producter 發送消息代碼：

        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){
            HostName ="192.168.1.8",UserName="hao",Password="abc123",Port= 5672
        };
        /// <summary>
        /// 路由名稱
        /// </summary>
        const string TopExchangeName = "topic.justin.exchange";

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

        public static void TopicExchangeSendMsg()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName, "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();
                    }
                }
            }
        }

　　Customer接收消息代碼：

        /// <summary>
        /// 連接配置
        /// </summary>
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {
            HostName = "192.168.1.8", UserName = "hao", Password = "abc123", Port = 5672
        };

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

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

        public static void TopicAcceptExchange()
        {
            using (IConnection conn = rabbitMqFactory.CreateConnection())
            {
                using (IModel channel = conn.CreateModel())
                {
                    channel.ExchangeDeclare(TopExchangeName, "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, noAck: false, consumer: consumer);

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

參考資料：

在 Windows 上安裝Rabbit MQ 指南（http://www.cnblogs.com/shanyou/p/4067250.html）

.NET 環境中使用RabbitMQ（http://www.cnblogs.com/yangecnu/p/4227535.html）

RabbitMQ Tutorial（http://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html）

源代碼下載

知道的越多，不知道的也就越多，多多學習！

RabbitMQ.Net 應用（1）

utf8 msg ssa 金融 count 不知道 six clu bind 概述　　MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。RabbitMQ是一個在AMQP基礎上完整的，可復用的企業消息系統。他遵循Mozilla

RabbitMQ.Net 應用（1）

RabbitMQ.Net 應用（1）

NET快取框架CacheManager在混合式開發框架中的應用（1）-CacheManager的介紹和使用基於C#的MongoDB資料庫開發應用（4）--Redis的安裝及使用

.NET快取框架CacheManager在混合式開發框架中的應用（1）-CacheManager的介紹和使用基於C#的MongoDB資料庫開發應用（4）--Redis的安裝及使用

如何使用DGBroker關閉redo應用（1）

ADO.NET教程（1）初識ado.net

Spring MVC 配置及應用（1）

Spring IOC 應用（1）管理物件

Revit二次開發高階應用（1）——利用wcf做簡單的族庫管理

Java反射應用（1）

RabbitMQ學習筆記（1）----RabbitMQ簡介與安裝

新手上手Tensorflow之手寫數字識別應用（1）

【資料結構】順序表的應用（1）

JavaWeb學習（十）Session的基本應用（1）

Storm安裝部署與應用（1）

Python的Django框架應用（1）

棧及其簡單應用（1）

RabbitMQ負載均衡（1）

高等數學：第六章定積分的應用（1）定積分的應用平面圖形的面積立體體積

稀疏矩陣的三元組表示的實現及應用（1）——建立稀疏矩陣三元組表示的演算法庫

第一個ASP.net網頁（1）

RabbitMQ.Net 應用（1）

相關推薦