.NET中RabbitMQ的使用
.NET中RabbitMQ的使用
概述
MQ全稱為Message Queue, 消息隊(duì)列(MQ)是一種應(yīng)用程序?qū)?yīng)用程序的通信方法。RabbitMQ是一個(gè)在AMQP基礎(chǔ)上完整的,可復(fù)用的企業(yè)消息系統(tǒng)。他遵循Mozilla Public License開源協(xié)議。AMQP(高級(jí)消息隊(duì)列協(xié)議) 是一個(gè)異步消息傳遞所使用的應(yīng)用層協(xié)議規(guī)范,作為線路層協(xié)議,而不是API(例如JMS),AMQP 客戶端能夠無視消息的來源任意發(fā)送和接受信息。AMQP的原始用途只是為金融界提供一個(gè)可以彼此協(xié)作的消息協(xié)議,而現(xiàn)在的目標(biāo)則是為通用消息隊(duì)列架構(gòu)提供通用構(gòu)建工具。因此,面向消息的中間件 (MOM)系統(tǒng),例如發(fā)布/訂閱隊(duì)列,沒有作為基本元素實(shí)現(xiàn)。AMQP當(dāng)中有四個(gè)概念非常重要(一個(gè)虛擬主機(jī)持有一組交換機(jī)、隊(duì)列和綁定):
更多理論性東西可以參考(在Windows上安裝Rabbit MQ 指南),針對(duì)隊(duì)列的講解相當(dāng)詳細(xì)
Window下安裝RabbbitMQ
文件下載安裝
Rabbit MQ 是建立在強(qiáng)大的Erlang OTP平臺(tái)上,因此安裝Rabbit MQ的前提是安裝Erlang。通過下面兩個(gè)連接下載安裝3.2.3 版本:
默認(rèn)安裝的Rabbit MQ 監(jiān)聽端口是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),打開CMD命令,cd到安裝目錄(..\rabbitmq_server-3.2.3\sbin)下,輸入下面的命令激活:
rabbitmq-plugins enable rabbitmq_management要重啟服務(wù)才能生效,可以執(zhí)行
net stop RabbitMQ && net start RabbitMQ輸入網(wǎng)址,打開監(jiān)控頁面: ?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ù)打開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第三方類庫,版本V3.6.5,高版本與.NET Framework 4.5有沖突,RabbitMQ Client地址
2、利用RabbitMQ Clinet類庫編碼(代碼內(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);//已過時(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í)間沒有能力處理太多的業(yè)務(wù),導(dǎo)致分配過來的隊(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ù)端可以刪除消息,分配新的消息過來channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);};//noAck設(shè)置false,告訴broker,發(fā)送消息之后,消息暫時(shí)不要?jiǎng)h除,等消費(fèi)者處理完成再說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的使用的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: python自带ide和pycharm哪
- 下一篇: python下载安装搭建