.NET Core實(shí)現(xiàn)RabbitMQ消息隊(duì)列的示例代碼
RabbitMQ 是一個(gè)流行的消息隊(duì)列中間件,它允許應(yīng)用程序通過(guò)異步消息的方式進(jìn)行通信。RabbitMQ 支持 AMQP 協(xié)議,可以通過(guò)多種方式與應(yīng)用程序交互。在本教程中,我們將深入探討如何在 .NET Core 環(huán)境中使用 RabbitMQ 來(lái)實(shí)現(xiàn)消息隊(duì)列。我們將學(xué)習(xí)如何在生產(chǎn)者端發(fā)送消息,消費(fèi)者端接收消息,并確保消息的可靠性。
1. 安裝和配置 RabbitMQ
在開(kāi)始使用 RabbitMQ 之前,首先需要確保你的機(jī)器上已經(jīng)安裝并運(yùn)行 RabbitMQ??梢酝ㄟ^(guò)以下方式安裝 RabbitMQ:
使用 Docker 安裝 RabbitMQ
RabbitMQ 提供了官方的 Docker 鏡像,這使得在本地機(jī)器上運(yùn)行 RabbitMQ 非常簡(jiǎn)單。
docker pull rabbitmq:management docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:management
5672是 RabbitMQ 的默認(rèn)消息隊(duì)列端口。15672是 RabbitMQ 管理插件的 Web 界面端口。通過(guò)瀏覽器訪問(wèn)http://localhost:15672可以登錄 RabbitMQ 管理界面,默認(rèn)的用戶名和密碼都是guest。
安裝并啟動(dòng) RabbitMQ 后,您可以繼續(xù)進(jìn)行開(kāi)發(fā)。
2. 安裝 RabbitMQ 客戶端庫(kù)
在 .NET Core 中與 RabbitMQ 進(jìn)行交互,我們需要使用 RabbitMQ.Client NuGet 包??梢酝ㄟ^(guò)以下命令在項(xiàng)目中添加這個(gè)依賴:
dotnet add package RabbitMQ.Client
這個(gè)庫(kù)提供了與 RabbitMQ 服務(wù)進(jìn)行交互所需的所有工具。
3. 創(chuàng)建生產(chǎn)者(Producer)
生產(chǎn)者是負(fù)責(zé)將消息發(fā)送到 RabbitMQ 的應(yīng)用程序。它通過(guò)連接到 RabbitMQ 服務(wù)器、創(chuàng)建一個(gè)隊(duì)列和交換機(jī),將消息發(fā)布到隊(duì)列中。
創(chuàng)建消息生產(chǎn)者代碼
下面是一個(gè)基本的生產(chǎn)者示例代碼,展示了如何連接到 RabbitMQ,聲明隊(duì)列,并發(fā)送一條簡(jiǎn)單的消息:
using RabbitMQ.Client;
using System;
using System.Text;
class Program
{
static void Main(string[] args)
{
// 創(chuàng)建連接工廠
var factory = new ConnectionFactory() { HostName = "localhost" };
// 創(chuàng)建連接和通道
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
// 聲明一個(gè)隊(duì)列(確保隊(duì)列存在)
channel.QueueDeclare(queue: "hello_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
// 創(chuàng)建消息
string message = "Hello, RabbitMQ!";
var body = Encoding.UTF8.GetBytes(message);
// 發(fā)送消息到隊(duì)列
channel.BasicPublish(exchange: "", routingKey: "hello_queue", basicProperties: null, body: body);
Console.WriteLine(" [x] Sent {0}", message);
}
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
在上面的代碼中:
ConnectionFactory用來(lái)創(chuàng)建連接到 RabbitMQ 服務(wù)器的連接。QueueDeclare用來(lái)聲明一個(gè)隊(duì)列,確保隊(duì)列存在。如果隊(duì)列已經(jīng)存在,聲明將被忽略。BasicPublish用來(lái)將消息發(fā)送到隊(duì)列。
參數(shù)說(shuō)明:
queue: 隊(duì)列的名稱(此例中是hello_queue)。durable: 是否將隊(duì)列標(biāo)記為持久化。如果設(shè)置為true,即使 RabbitMQ 重啟,隊(duì)列也會(huì)存在。exclusive: 是否使隊(duì)列只對(duì)當(dāng)前連接可用。autoDelete: 是否在最后一個(gè)消費(fèi)者斷開(kāi)連接時(shí)自動(dòng)刪除隊(duì)列。
4. 創(chuàng)建消費(fèi)者(Consumer)
消費(fèi)者從隊(duì)列中獲取并處理消息。消費(fèi)者通常是另一個(gè)應(yīng)用程序,它會(huì)連接到 RabbitMQ,并持續(xù)地從隊(duì)列中取出消息進(jìn)行處理。
創(chuàng)建消息消費(fèi)者代碼
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
class Program
{
static void Main(string[] args)
{
// 創(chuàng)建連接工廠
var factory = new ConnectionFactory() { HostName = "localhost" };
// 創(chuàng)建連接和通道
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
// 聲明隊(duì)列,確保消費(fèi)者能夠連接到相同的隊(duì)列
channel.QueueDeclare(queue: "hello_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
// 創(chuàng)建消費(fèi)者對(duì)象
var consumer = new EventingBasicConsumer(channel);
// 消息處理邏輯
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
};
// 開(kāi)始消費(fèi)消息
channel.BasicConsume(queue: "hello_queue", autoAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
在上面的代碼中:
QueueDeclare用來(lái)確保消費(fèi)者連接到相同的隊(duì)列。EventingBasicConsumer是消費(fèi)者的實(shí)現(xiàn),用于異步接收消息。BasicConsume用于開(kāi)始消費(fèi)消息,autoAck設(shè)置為true,表示自動(dòng)確認(rèn)消息。
參數(shù)說(shuō)明:
autoAck: 如果設(shè)置為true,消費(fèi)者會(huì)自動(dòng)確認(rèn)消息。如果設(shè)置為false,需要手動(dòng)確認(rèn)消息。
5. 持久化消息
如果您希望在 RabbitMQ 重啟后保持消息的持久性,可以在生產(chǎn)者和消費(fèi)者中啟用消息的持久化。
消息持久化設(shè)置
在生產(chǎn)者端發(fā)送持久化消息:
// 設(shè)置消息持久化 var properties = channel.CreateBasicProperties(); properties.Persistent = true; // 設(shè)置消息為持久化 channel.BasicPublish(exchange: "", routingKey: "hello_queue", basicProperties: properties, body: body);
此外,聲明隊(duì)列時(shí)也需要設(shè)置 durable: true,確保隊(duì)列本身是持久化的。
channel.QueueDeclare(queue: "hello_queue", durable: true, exclusive: false, autoDelete: false, arguments: null);
6. 消息確認(rèn)機(jī)制
在消息傳遞過(guò)程中,為了確保消息被成功消費(fèi)并避免丟失,可以啟用消息確認(rèn)機(jī)制。在這種情況下,消費(fèi)者需要顯式確認(rèn)消息。
啟用手動(dòng)消息確認(rèn)
在消費(fèi)者端禁用自動(dòng)確認(rèn),并手動(dòng)確認(rèn)每條已成功處理的消息:
channel.BasicConsume(queue: "hello_queue", autoAck: false, consumer: consumer);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
// 手動(dòng)確認(rèn)消息
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
BasicAck 用于確認(rèn)消息已經(jīng)被處理。deliveryTag 是消息的標(biāo)識(shí)符,multiple 參數(shù)表示是否確認(rèn)多個(gè)消息。
7. 運(yùn)行和測(cè)試
- 啟動(dòng)消費(fèi)者應(yīng)用程序,確保它可以連接到 RabbitMQ 并等待消息。
- 啟動(dòng)生產(chǎn)者應(yīng)用程序,它將發(fā)送消息到 RabbitMQ 隊(duì)列。
- 消費(fèi)者將從隊(duì)列中接收到消息,并進(jìn)行處理。
如果一切配置正確,您將在控制臺(tái)中看到生產(chǎn)者發(fā)送的消息以及消費(fèi)者處理的消息。
8. 總結(jié)
通過(guò)本教程,我們學(xué)習(xí)了如何在 .NET Core 中使用 RabbitMQ 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的消息隊(duì)列系統(tǒng)。關(guān)鍵步驟包括:
- 安裝 RabbitMQ 客戶端庫(kù)。
- 在生產(chǎn)者中聲明隊(duì)列并發(fā)送消息。
- 在消費(fèi)者中聲明隊(duì)列并處理消息。
- 配置消息持久化和確認(rèn)機(jī)制,確保消息的可靠性。
RabbitMQ 是一個(gè)強(qiáng)大的消息隊(duì)列中間件,適用于各種需要解耦和異步通信的應(yīng)用程序。通過(guò)靈活的交換機(jī)和隊(duì)列配置,您可以實(shí)現(xiàn)不同的消息傳遞模式,以滿足不同的業(yè)務(wù)需求。
到此這篇關(guān)于.NET Core實(shí)現(xiàn)RabbitMQ消息隊(duì)列的示例代碼的文章就介紹到這了,更多相關(guān).NET Core RabbitMQ消息隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- 如何在一個(gè).NET?Core項(xiàng)目中使用RabbitMQ進(jìn)行即時(shí)消息管理
- .NET Core中RabbitMQ使用死信隊(duì)列的實(shí)現(xiàn)
- .Net?Core和RabbitMQ限制循環(huán)消費(fèi)的方法
- .NETCore基于RabbitMQ實(shí)現(xiàn)延時(shí)隊(duì)列的兩方法
- 運(yùn)用.net core中實(shí)例講解RabbitMQ高可用集群構(gòu)建
- 運(yùn)用.NetCore實(shí)例講解RabbitMQ死信隊(duì)列,延時(shí)隊(duì)列
- 運(yùn)用.net core中實(shí)例講解RabbitMQ
相關(guān)文章
關(guān)于EF的Code?First的使用以及踩坑記錄
這篇文章主要介紹了關(guān)于EF的Code?First的使用以及踩坑記錄,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-11-11
.NET Core 1.0創(chuàng)建Self-Contained控制臺(tái)應(yīng)用
這篇文章主要為大家詳細(xì)介紹了.NET Core 1.0創(chuàng)建Self-Contained控制臺(tái)應(yīng)用的相關(guān)資料,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-04-04
.Net RabbitMQ實(shí)現(xiàn)HTTP API接口調(diào)用
RabbitMQ Management插件還提供了基于RESTful風(fēng)格的HTTP API接口來(lái)方便調(diào)用。本文就主要介紹了.Net RabbitMQ實(shí)現(xiàn)HTTP API接口調(diào)用,感興趣的可以了解一下2021-06-06
.net core版 文件上傳/ 支持批量上傳拖拽及預(yù)覽功能(bootstrap fileinput上傳文件)
本篇內(nèi)容主要解決.net core中文件上傳的問(wèn)題 開(kāi)發(fā)環(huán)境:ubuntu+vscode.本文給大家介紹的非常詳細(xì),感興趣的朋友一起看看吧2017-03-03
Entity?Framework?Core關(guān)聯(lián)刪除
關(guān)聯(lián)刪除通常是一個(gè)數(shù)據(jù)庫(kù)術(shù)語(yǔ),用于描述在刪除行時(shí)允許自動(dòng)觸發(fā)刪除關(guān)聯(lián)行的特征;即當(dāng)主表的數(shù)據(jù)行被刪除時(shí),自動(dòng)將關(guān)聯(lián)表中依賴的數(shù)據(jù)行進(jìn)行刪除,或者將外鍵更新為NULL或默認(rèn)值。本文將為大家具體介紹一下Entity?Framework?Core關(guān)聯(lián)刪除,需要的可以參考一下2021-12-12
DropDownList獲取的SelectIndex一直為0的問(wèn)題
由于初始化判斷出錯(cuò)導(dǎo)致每次傳到服務(wù)器的時(shí)候會(huì)初始化一次,這就導(dǎo)致每次獲取DropDownList的SelectIndex的時(shí)候只能是02014-06-06
asp.net不同頁(yè)面間數(shù)據(jù)傳遞的多種方法
這篇文章主要介紹了asp.net不同頁(yè)面間數(shù)據(jù)傳遞的多種方法,包括使用QueryString顯式傳遞、頁(yè)面對(duì)象的屬性、cookie、Cache等9種方法2014-01-01

