如何用.NETCore操作RabbitMQ
什么是RabbitMQ？
RabbitMQ是由erlang語(yǔ)言開(kāi)發(fā)的一個(gè)基于A(yíng)MQP(Advanced Message Queuing Protocol)協(xié)議的企業(yè)級(jí)消息隊(duì)列中間件?？蓪?shí)現(xiàn)隊(duì)列，訂閱/發(fā)布，路由，通配符等工作模式。
為什么要使用RabbitMQ?
- 異步處理：比如發(fā)送郵件，發(fā)送短信等不需要等待處理結(jié)果的操作
- 應(yīng)用解耦：比如下單成功后，通知倉(cāng)庫(kù)發(fā)貨，不需要等待倉(cāng)庫(kù)回應(yīng)，通過(guò)消息隊(duì)列去通知倉(cāng)庫(kù)，降低應(yīng)用間耦合程序，可并行開(kāi)發(fā)兩個(gè)功能模塊
- 流量削鋒：在搶購(gòu)或者其他的活動(dòng)頁(yè)，服務(wù)處于爆發(fā)式請(qǐng)求狀態(tài)，如果直連數(shù)據(jù)庫(kù)，數(shù)據(jù)庫(kù)容易被拖垮。搶購(gòu)商品也容易出現(xiàn)庫(kù)存超賣(mài)的情況。通過(guò)隊(duì)列可有效解決該問(wèn)題。
- 日志處理：在單機(jī)中，日志直接寫(xiě)入到文件目錄中，但是在分布式應(yīng)用中，日志需要有統(tǒng)一的處理機(jī)制，可通過(guò)消息隊(duì)列統(tǒng)一由某個(gè)消費(fèi)端做處理。
- 消息通信：如生產(chǎn)端和消費(fèi)端可通過(guò)隊(duì)列進(jìn)行異步通信
如何安裝RabbitMQ?
Windows端
1.安裝erlang語(yǔ)言運(yùn)行環(huán)境
https://erlang.org/download/otp_win64_23.2.exe
下載后直接下一步即可
2.安裝RabbitMQ
https://www.rabbitmq.com/install-windows.html
直接點(diǎn)擊安裝下一步即可按章
3.安裝RabbitMQ的Web管理平臺(tái)
RabbitMQ的管理平臺(tái)是通過(guò)插件的形式使用，需要手動(dòng)啟用管理平臺(tái)
在Windows下，RabbitMQ默認(rèn)被安裝到C:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.14 下。
打開(kāi)sbin ，在cmd或者powershell中執(zhí)行
rabbitmq-plugins.bat enable rabbitmq_management
安裝完成后，瀏覽器打開(kāi) http://localhost:15672/ 即可看到RabbitMQ的管理界面。輸入默認(rèn)賬號(hào)密碼 guest 成功登錄。
Linux環(huán)境安裝
1.Ubuntu:https://www.rabbitmq.com/install-debian.html
2.Centos:https://www.rabbitmq.com/install-rpm.html
RabbitMQ的基本概念
生產(chǎn)者
發(fā)送消息的端
消費(fèi)者
獲取消息并處理的端
Connection
一個(gè)終端連接。每一個(gè)Connection都可以在RabbitMQ后臺(tái)看到
Channel
Channel是建立在Connection上的一個(gè)虛擬通信管道。一般情況下，往消息隊(duì)列中寫(xiě)入多條消息，為了不每條消息都建立一個(gè)TCP連接，所以RabbitMQ的做法是多條消息可以公用一個(gè)Connection，大大提高M(jìn)Q的負(fù)載能力。
Exchange
Exchange是一個(gè)虛擬交換機(jī)。每一條消息都必須要通過(guò)交換機(jī)才能能進(jìn)入對(duì)應(yīng)的隊(duì)列，可以理解為網(wǎng)絡(luò)設(shè)備中的交換機(jī)，是一個(gè)意思。
Queue
Queue是一個(gè)存儲(chǔ)消息的內(nèi)部對(duì)象，所有的Rabbit MQ消息都存儲(chǔ)在Queue中。生產(chǎn)者所生產(chǎn)的消息會(huì)存儲(chǔ)在Queue中，消費(fèi)者獲取的消息也是從Queue中獲取。
如何在.NET Core中使用RabbitMQ?
nuget安裝
dotnet add package RabbitMQ.Client
創(chuàng)建生產(chǎn)者
const string QUEUENAME = "HELLO_MQ";
//創(chuàng)建連接對(duì)象工廠(chǎng)
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672, //RabbitMQ默認(rèn)的端口
};
while (true)
{
using var conn = factory.CreateConnection();
var chanel = conn.CreateModel();
chanel.QueueDeclare(QUEUENAME, true, false, false);
Console.WriteLine("輸入生產(chǎn)內(nèi)容：");
var input = Console.ReadLine();
chanel.BasicPublish("", QUEUENAME, null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
}
在循環(huán)中，輸入一個(gè)值，按下enter，即可推送一條消息到隊(duì)列。
也可以直接在RabbitMQ的管理后臺(tái)查看
可以看到我們發(fā)送的消息已經(jīng)被RabbitMQ存儲(chǔ)在Queue中了。只等某個(gè)幸運(yùn)的消費(fèi)者前來(lái)消費(fèi)。
創(chuàng)建消費(fèi)者
const string QUEUENAME = "HELLO_MQ";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var chanel = conn.CreateModel();
chanel.QueueDeclare(QUEUENAME, true, false, false);
EventingBasicConsumer consumer = new EventingBasicConsumer(chanel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
chanel.BasicAck(e.DeliveryTag, true); //收到回復(fù)后，RabbitMQ會(huì)直接在隊(duì)列中刪除這條消息
};
chanel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("啟動(dòng)成功");
Console.ReadLine();
啟動(dòng)成功后，consumer的Received方法，會(huì)收到一條來(lái)自MQ的消息，
如果處理完成后，不調(diào)用chennel的BasicAck方法，那么這條消息依然會(huì)存在，下次有消費(fèi)者出現(xiàn)，會(huì)再次推送給消費(fèi)者。
簡(jiǎn)單的RabbitMQ Hello World到這里就算完成了。接下來(lái)就是稍微高級(jí)一點(diǎn)的應(yīng)用
RabbitMQ的工作模式
Work Queue 工作隊(duì)列模式
工作隊(duì)列模式的意思就是一個(gè)生產(chǎn)者對(duì)應(yīng)多個(gè)消費(fèi)者。RabbitMQ會(huì)使用輪詢(xún)?nèi)ソo每個(gè)消費(fèi)者發(fā)送消息。
publish/subscribe
發(fā)布訂閱模式是屬于比較用多的一種。
發(fā)布訂閱，是由交換機(jī)發(fā)布消息給多個(gè)隊(duì)列。多個(gè)隊(duì)列再對(duì)應(yīng)多個(gè)消費(fèi)者。
發(fā)布訂閱模式對(duì)應(yīng)的交換機(jī)類(lèi)型的fanout。
消費(fèi)者
A
const string QUEUENAME = "HELLO_MQ_B";
const string TESTEXCHANGE = "TESTEXCHANGE";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var channel = conn.CreateModel();
//定義隊(duì)列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定義交換機(jī)
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
//綁定隊(duì)列到交換機(jī)
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
channel.BasicAck(e.DeliveryTag, true); //收到回復(fù)后，RabbitMQ會(huì)直接在隊(duì)列中刪除這條消息
};
channel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("啟動(dòng)成功");
Console.ReadLine();
B
const string QUEUENAME = "HELLO_MQ";
const string TESTEXCHANGE = "TESTEXCHANGE";
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672,
};
var conn = factory.CreateConnection();
var channel = conn.CreateModel();
//定義隊(duì)列
channel.QueueDeclare(QUEUENAME, true, false, false);
//定義交換機(jī)
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
//綁定隊(duì)列到交換機(jī)
channel.QueueBind(QUEUENAME, TESTEXCHANGE, "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (a, e) =>
{
Console.WriteLine($"{DateTime.Now.ToString()}接收到消息:" + Encoding.Default.GetString(e.Body.ToArray()));
channel.BasicAck(e.DeliveryTag, true); //收到回復(fù)后，RabbitMQ會(huì)直接在隊(duì)列中刪除這條消息
};
channel.BasicConsume(QUEUENAME, false, consumer);
Console.WriteLine("啟動(dòng)成功");
Console.ReadLine();
生產(chǎn)者
const string QUEUENAME = "HELLO_MQ";
const string QUEUENAME_B = "HELLO_MQ_B";
const string TESTEXCHANGE = "TESTEXCHANGE";
//創(chuàng)建連接對(duì)象工廠(chǎng)
var factory = new ConnectionFactory()
{
UserName = "guest",
Password = "guest",
HostName = "localhost",
Port = 5672, //RabbitMQ默認(rèn)的端口
};
using var conn = factory.CreateConnection();
while (true)
{
var channel = conn.CreateModel();
//定義交換機(jī)
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Fanout, true, false);
Console.WriteLine("輸入生產(chǎn)內(nèi)容：");
var input = Console.ReadLine();
channel.BasicPublish(TESTEXCHANGE,"", null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
}
在生產(chǎn)者運(yùn)行成功后，RabbitMQ后臺(tái)會(huì)出現(xiàn)一個(gè)交換機(jī)，點(diǎn)擊交換機(jī)會(huì)看到交換機(jī)下綁定了兩個(gè)隊(duì)列
從生產(chǎn)者發(fā)送消息到隊(duì)列，兩個(gè)消費(fèi)者會(huì)同時(shí)收到消息
routing模式
routing模式對(duì)應(yīng)的交換機(jī)類(lèi)型是direct，和發(fā)布訂閱模式的區(qū)別在于：routing模式下，可以指定一個(gè)routingkey，用于區(qū)分消息
生產(chǎn)者
var channel = conn.CreateModel();
//定義交換機(jī)
channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false);
//綁定隊(duì)列到交換機(jī)
Console.WriteLine("輸入生產(chǎn)內(nèi)容：");
var input = Console.ReadLine();
channel.BasicPublish(TESTEXCHANGE, "INFO", null, Encoding.Default.GetBytes("hello rabbitmq:" + input));
消費(fèi)者 A
//定義隊(duì)列 channel.QueueDeclare(QUEUENAME, true, false, false); //定義交換機(jī) channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false); //綁定隊(duì)列到交換機(jī) channel.QueueBind(QUEUENAME, TESTEXCHANGE, "INFO");
消費(fèi)者 B
//定義隊(duì)列 channel.QueueDeclare(QUEUENAME, true, false, false); //定義交換機(jī) channel.ExchangeDeclare(TESTEXCHANGE, ExchangeType.Direct, true, false); //綁定隊(duì)列到交換機(jī) channel.QueueBind(QUEUENAME, TESTEXCHANGE, "ERROR");
綁定成功后，發(fā)送消息，消費(fèi)者A可以收到消息，消費(fèi)者B無(wú)法收到消息。
如果遇到指定routingKey生產(chǎn)一條消息，結(jié)果 AB消費(fèi)者都收到的情況。建議在RabbitMQ后臺(tái)的交換機(jī)下看一下綁定的Queue是否重復(fù)綁定了多個(gè)routingKey.
topic通配符模式
在通配符模式下，RabbitMQ使用模糊匹配來(lái)決定把消息推送給哪個(gè)生產(chǎn)者。通配符有兩個(gè)符號(hào)來(lái)匹配routingKey
1.*匹配一個(gè)字符 如:*.qq.com 可匹配 1.qq.com
2.#匹配一個(gè)或者多個(gè)字符。 如：*.qq.com 可匹配 1.qq.com或者1111.qq.com
其他的操作基本和routing模式一樣。
header模式
header模式是把routingkey放到header中.取消掉了routingKey。并使用一個(gè)字典傳遞 K、V的方式來(lái)匹配。
比如同時(shí)要給用戶(hù)發(fā)送郵件和短信，可直接通過(guò)header的鍵值對(duì)來(lái)匹配綁定的值，把消息傳遞給發(fā)短信和郵件的生產(chǎn)者.
以上就是如何用.NETCore操作RabbitMQ的詳細(xì)內(nèi)容，更多關(guān)于.NETCore 操作RabbitMQ的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
c#使用nsoup解析html亂碼解決方法分享 nsoup教程
NSoup是JSoup的Net移植版本。使用方法基本一致。如果項(xiàng)目涉及HTML的處理，強(qiáng)烈推薦NSoup。但是遺憾的是NSoup默認(rèn)的編碼是UTF-8，處理中文有亂碼,下面給出二種解決方法2014-01-01
C#實(shí)現(xiàn)實(shí)時(shí)監(jiān)控文件夾變化
在開(kāi)發(fā)各種應(yīng)用程序時(shí),我們經(jīng)常需要對(duì)文件系統(tǒng)中的文件或文件夾進(jìn)行實(shí)時(shí)監(jiān)測(cè),下面就跟隨小編一起來(lái)看看具體如何使用C#實(shí)現(xiàn)這一功能吧2024-03-03
在類(lèi)庫(kù)或winform項(xiàng)目中打開(kāi)另一個(gè)winform項(xiàng)目窗體的方法
這篇文章主要介紹了在類(lèi)庫(kù)或winform項(xiàng)目中打開(kāi)另一個(gè)winform項(xiàng)目窗體的方法,可以實(shí)現(xiàn)Winform項(xiàng)目間窗體的調(diào)用,在進(jìn)行Winform項(xiàng)目開(kāi)發(fā)中非常具有實(shí)用價(jià)值,需要的朋友可以參考下2014-11-11
這篇文章主要介紹了C#中Equality和Identity淺析,本文先是講解了Equality和Identity的定義,同時(shí)講解了判斷兩個(gè)對(duì)象等價(jià)性的4種方法,需要的朋友可以參考下2014-12-12
Unity UI或3D場(chǎng)景實(shí)現(xiàn)跟隨手機(jī)陀螺儀的晃動(dòng)效果
這篇文章主要介紹了Unity UI或3D場(chǎng)景實(shí)現(xiàn)跟隨手機(jī)陀螺儀的晃動(dòng)效果，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-11-11
.NET中的async和await關(guān)鍵字使用及Task異步調(diào)用實(shí)例
這篇文章主要介紹了.NET中的async和await關(guān)鍵字使用及Task異步調(diào)用實(shí)例,本文還包含了取消執(zhí)行和顯示進(jìn)度的例子,需要的朋友可以參考下2014-07-07
C#引用類(lèi)型和值類(lèi)型的介紹與實(shí)例
這篇文章主要介紹了C#引用類(lèi)型和值類(lèi)型，有需要的朋友可以參考一下2013-12-12
C#實(shí)現(xiàn)航班預(yù)訂系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了C#實(shí)現(xiàn)航班預(yù)訂系統(tǒng)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-05-05
C# 撒列實(shí)現(xiàn)關(guān)鍵字過(guò)濾的實(shí)例
C# 撒列實(shí)現(xiàn)關(guān)鍵字過(guò)濾的實(shí)例，需要的朋友可以參考一下2013-04-04
這篇文章主要介紹了詳解如何在C#/.NET Core中使用責(zé)任鏈模式，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-05-05