.NET中RabbitMQ的使用

概述

　　MQ全稱為Message Queue, 消息隊(duì)列（MQ）是一種應(yīng)用程序?qū)?yīng)用程序的通信方法。RabbitMQ是一個(gè)在AMQP基礎(chǔ)上完整的，可復(fù)用的企業(yè)消息系統(tǒng)。他遵循Mozilla Public License開(kāi)源協(xié)議。AMQP(高級(jí)消息隊(duì)列協(xié)議) 是一個(gè)異步消息傳遞所使用的應(yīng)用層協(xié)議規(guī)范，作為線路層協(xié)議，而不是API（例如JMS），AMQP 客戶端能夠無(wú)視消息的來(lái)源任意發(fā)送和接受信息。AMQP的原始用途只是為金融界提供一個(gè)可以彼此協(xié)作的消息協(xié)議，而現(xiàn)在的目標(biāo)則是為通用消息隊(duì)列架構(gòu)提供通用構(gòu)建工具。因此，面向消息的中間件 （MOM）系統(tǒng)，例如發(fā)布/訂閱隊(duì)列，沒(méi)有作為基本元素實(shí)現(xiàn)。AMQP當(dāng)中有四個(gè)概念非常重要（一個(gè)虛擬主機(jī)持有一組交換機(jī)、隊(duì)列和綁定）：

virtual host，虛擬主機(jī)

exchange，交換機(jī)

queue，隊(duì)列

binding，綁定

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

Window下安裝RabbbitMQ

文件下載安裝

Rabbit MQ 是建立在強(qiáng)大的Erlang OTP平臺(tái)上，因此安裝Rabbit MQ的前提是安裝Erlang。通過(guò)下面兩個(gè)連接下載安裝3.2.3 版本：

下載并安裝?Erlang OTP For Windows?(vR16B03)

運(yùn)行安裝?Rabbit MQ Server Windows Installer?(v3.2.3)

默認(rèn)安裝的Rabbit MQ 監(jiān)聽(tīng)端口是5672。先安裝Erlang OTP后安裝RabbitMQ，安裝方式默認(rèn)即可，RabbitMQ可以勾選安裝后臺(tái)服務(wù)、服務(wù)啟動(dòng)和停止等操作。

激活Rabbit MQ's Management Plugin

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

rabbitmq-plugins enable rabbitmq_management

要重啟服務(wù)才能生效，可以執(zhí)行

net stop RabbitMQ && net start RabbitMQ

輸入網(wǎng)址，打開(kāi)監(jiān)控頁(yè)面： ?http://localhost:15672 (默認(rèn)賬號(hào)和密碼：guest 和guest)

配置RabbitMQ用戶權(quán)限

RabbitMQ是存在用戶權(quán)限的，默認(rèn)是guest 密碼也是guest，隸屬于Administrator管理員下。現(xiàn)需要配置新用戶和權(quán)限，繼續(xù)打開(kāi)CMD命令，cd到安裝目錄sbin下：

用戶操作指令：

::查詢服務(wù)狀態(tài)
rabbitmqctl status

::列舉虛擬主機(jī)列表
rabbitmqctl list_vhosts ::列舉用戶列表
rabbitmqctl list_users

:: 添加用戶和密碼 rabbitmqctl add_user hao abc123:: 設(shè)置權(quán)限 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第三方類庫(kù)，版本V3.6.5，高版本與.NET Framework 4.5有沖突，RabbitMQ Client地址

2、利用RabbitMQ Clinet類庫(kù)編碼(代碼內(nèi)容有注釋，此處不做詳細(xì)解釋，文章后有完整代碼的下載地址)

?

　　<1>RabbitMQ的direct類型Exchange

　　　Producter發(fā)送消息代碼：

/// <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";//隊(duì)列名稱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("***發(fā)送時(shí)間:{0}，發(fā)送完成，輸入exit退出消息發(fā)送",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";//隊(duì)列名稱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("***接收時(shí)間:{0}，消息內(nèi)容：{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));}}}}

　　

　　但是這種處理速度較慢，因?yàn)檠h(huán)線程等待。高效的接收消息的方式可以使用EventingBasicConsumer進(jìn)行消息接收處理，修改代碼內(nèi)容如下：

?

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("***接收時(shí)間:{0}，消息內(nèi)容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));};channel.BasicConsume(QueueName, noAck: true, consumer: consumer);//已過(guò)時(shí)用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();}}}

　　

　　但是有些時(shí)候，消費(fèi)者同一時(shí)間沒(méi)有能力處理太多的業(yè)務(wù)，導(dǎo)致分配過(guò)來(lái)的隊(duì)列消息不能及時(shí)處理完成，這個(gè)時(shí)候，我們可以設(shè)置BasicQos屬性，告訴Broker同一時(shí)間將未處理完成的消息分配其他消費(fèi)者，所以接收消息的地方需要略做修改，代碼如下：

?

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同一時(shí)間只處理一個(gè)消息//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("***接收時(shí)間:{0}，消息內(nèi)容：{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可以服務(wù)端可以刪除消息，分配新的消息過(guò)來(lái)channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);};//noAck設(shè)置false,告訴broker，發(fā)送消息之后，消息暫時(shí)不要?jiǎng)h除，等消費(fèi)者處理完成再說(shuō)channel.BasicConsume(QueueName, noAck: false, consumer: consumer);Console.WriteLine("按任意值，退出程序");Console.ReadKey();}} }

　　

　　<2> RabbitMQ的Topic類型Exchange

　　Producter 發(fā)送消息代碼：

/// <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";//隊(duì)列名稱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("***發(fā)送時(shí)間:{0}，發(fā)送完成，輸入exit退出消息發(fā)送", 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";//隊(duì)列名稱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("***接收時(shí)間:{0}，消息內(nèi)容：{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 環(huán)境中使用RabbitMQ（http://www.cnblogs.com/yangecnu/p/4227535.html）

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

源代碼下載

知道的越多，不知道的也就越多，多多學(xué)習(xí)！

轉(zhuǎn)載于:https://www.cnblogs.com/Jeely/p/10788484.html

總結(jié)

以上是生活随笔為你收集整理的.NET中RabbitMQ的使用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。