運用.NetCore例項講解RabbitMQ死信佇列,延時佇列
目錄
- 一、死信佇列
- 二、延時佇列
- 三、延時訊息設定不同過期時間
- 四、延時訊息用延時外掛的方式實現
一、死信佇列
描述:Q1佇列綁定了x-dead-letter-exchange(死信交換機)為X2,x-dead-letter-routing-key(死信路由key)指向Q2(佇列2)
P(生產者)傳送訊息經X1(交換機1)路由到Q1(佇列1),Q1的訊息觸發特定情況,自動把訊息經X2(交換機2)路由到Q2(佇列2),C(消費者)直接訊息Q2的訊息。
特定情況有哪些呢:
- 1.訊息被拒(basic.reject or basic.nack)並且沒有重新入隊(requeue=false);
- 2.當前佇列中的訊息數量已經超過最大長度(建立佇列時指定" x-max-length引數設定佇列最大訊息數量)。
- 3.訊息在佇列中過期,即當前訊息在佇列中的存活時間已經超過了預先設定的TTL(Time To Live)時間;
這裡演示情況1:
假如場景:Q1中佇列資料不完整,就算從新處理也會報錯,那就可以不ack,把這個訊息轉到死信佇列另外處理。
生產者:
public static void SendMessage() { //死信交換機 string dlxexChange = "dlx.exchange"; //死信佇列 string dlxQueueName = "dlx.queue"; //訊息交換機 string exchange = "direct-exchange"; //訊息佇列 string queueName = "queue_a"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //建立死信交換機 channel.ExchangeDeclare(dlxexChange,type: ExchangeType.Direct,durable: true,autoDelete: false); //建立死信佇列 channel.QueueDeclare(dlxQueueName,exclusive: false,autoDelete: false); //死信佇列繫結死信交換機 channel.QueueBind(dlxQueueName,dlxexChange,routingKey: dlxQueueName); // 建立訊息交換機 channel.ExchangeDeclare(exchange,autoDelete: false); //建立訊息佇列,並指定死信佇列 channel.QueueDeclare(queueName,autoDelete: false,arguments: new Dictionary<string,object> { { "x-dead-letter-exchange",dlxexChange},//設定當前佇列的DLX(死信交換機) { "x-dead-letter-routing-key",dlxQueueName},//設定DLX的路由key,DLX會根據該值去找到死信訊息存放的佇列 }); //訊息佇列繫結訊息交換機 channel.QueueBind(queueName,exchange,routingKey: queueName); string message = "hello rabbitmq message"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; //釋出訊息 channel.BasicPublish(exchange: exchange,routingKey: queueName,basicProperties: properties,body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"向佇列:{queueName}傳送訊息:{message}"); } } }
消費者:
public static void Consumer() { //死信交換機 string dlxexChange = "dlx.exchange"; //死信佇列 string dlxQueueName = "dlx.queue"; //訊息交換機 string exchange = "direct-exchange"; //訊息佇列 string queueName = "queue_a"; var connection = RabbitMQHelper.GetConnection(); { //建立通道 var channel = connection.CreateModel(); { //建立死信交換機 channel.ExchangeDeclare(dlxexChange,//設定當前佇列的DLX { "x-dead-letter-routing-key",routingKey: queueName); var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0,prefetchCount: 1,global: true); consumer.Received += (model,ea) => { //處理業務 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"佇列{queueName}消費訊息:{message},不做ack確認"); //channel.BasicAck(ea.DeliveryTag,false); //不ack(BasicNack),且不把訊息放回佇列(requeue:false) channel.BasicNack(ea.DeliveryTag,false,requeue: false); }; channel.BasicConsume(queueName,autoAck: false,consumer); } } }
消費者加上channel.BasickNack()模擬訊息處理不了,不ack確認。
執行結果:
RabbitMQ管理介面:
看到訊息佇列為queue_a,特性有DLX(死信交換機),DLK(死信路由)。因為消費端不nack,觸發了死信,被轉發到了死信佇列dlx.queue。
二、延時佇列
延時佇列其實也是配合死信佇列一起用,其實就是上面死信佇列的第二中情況。給佇列新增訊息過時時間(TTL),變成延時佇列。
簡單的描述就是:P(生產者)傳送訊息到Q1(延時佇列),Q1的訊息有過期時間,比如10s,那10s後訊息過期就會觸發死信,從而把訊息轉發到Q2(死信佇列)。
解決問題場景:像商城下單,未支付時取消訂單場景。下單時寫一條記錄入Q1,延時30分鐘後轉到Q2,消費Q2,檢查訂單,支付則不做操作,沒支付則取消訂單,恢復庫存。
生產者程式碼:
public static void SendMessage() { //死信交換機 string dlxexChange = "dlx.exchange"; //死信佇列 string dlxQueueName = "dlx.queue"; //訊息交換機 string exchange = "direct-exchange"; //訊息佇列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //建立死信交換機 channel.ExchangeDeclare(dlxexChange,並指定死信佇列,和設定這個佇列的訊息過期時間為10s channel.QueueDeclare(queueName,//設定DLX的路由key,DLX會根據該值去找到死信訊息存放的佇列 { "x-message-ttl",10000} //設定佇列的訊息過期時間 }); //訊息佇列繫結訊息交換機 channel.QueueBind(queueName,body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向佇列:{queueName}傳送訊息:{message}"); } } }
消費者程式碼:
public static void Consumer() { //死信交換機 string dlxexChange = "dlx.exchange"; //死信佇列 string dlxQueueName = "dlx.queue"; var connection = RabbitMQHelper.GetConnection(); { //建立通道 var channel = connection.CreateModel(); { //建立死信交換機 channel.ExchangeDeclare(dlxexChange,routingKey: dlxQueueName); var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0,ea) => { //處理業務 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{DateTime.Now},佇列{dlxQueueName}消費訊息:{message}"); channel.BasicAck(ea.DeliveryTag,false); }; channel.BasicConsume(dlxQueueName,consumer); } } }
執行程式碼:
向延時佇列傳送訊息,監聽死信佇列,傳送和收到訊息時間剛好是設定的10s。
RabbitMQ管理介面:
三、延時訊息設定不同過期時間
上面的延時佇列能解決訊息過期時間都是相同的場景,能不能解決訊息的過期時間是不一樣的呢?
例如場景:機器人客服,為了更像人為操作,收到訊息後要隨機3-10秒回覆客戶。
- 1)佇列不設定TTL(訊息過期時間),把過期時間設定在訊息上。
生產者程式碼:
public static void SendMessage() { //死信交換機 string dlxexChange = "dlx.exchange"; //死信佇列 string dlxQueueName = "dlx.queue"; //訊息交換機 string exchange = "direct-exchange"; //訊息佇列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //建立死信交換機 channel.ExchangeDeclare(dlxexChange,//設定DLX的路由key,DLX會根據該值去找到死信訊息存放的佇列 //{ "x-message-ttl",routingKey: queueName); string message = "hello rabbitmq message 10s後處理"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; properties.Expiration = "10000";//訊息的有效期10s //釋出訊息,延時10s channel.BasicPublish(exchange: exchange,body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向佇列:{queueName}傳送訊息:{message},延時:10s"); string message2 = "hello rabbitmq message 5s後處理"; var properties2 = channel.CreateBasicProperties(); properties2.Persistent = true; properties2.Expiration = "5000";//訊息有效期5s //釋出訊息,延時5s channel.BasicPublish(exchange: exchange,basicProperties: properties2,body: Encoding.UTF8.GetBytes(message2)); Console.WriteLine($"{DateTime.Now},向佇列:{queueName}傳送訊息:{message2},延時:5s"); } } }
消費者程式碼還是上面延時佇列的不變,先試下效果。
生產者向佇列中傳送一條延時10s的訊息再發一條延時5秒的訊息,但消費者卻先拿到延時10s的,再拿到延時5秒的,我想要的結果是先拿到延時5s的再拿到延時10s的,是什麼原因呢。
原因是:佇列是先進先出的,而RabbitMQ只會對首位第一條訊息做檢測,第一條沒過期,那麼後面的訊息就會阻塞住等待前面的過期。
解決辦法:增加一個消費者對延時佇列消費,不ack,把第一條訊息放到佇列尾部。一直讓訊息在流動,這樣就能檢測到了。
- 2)新增消費者程式碼:
public static void SendMessage()
{
//死信交換機
string dlxexChange = "dlx.exchange";
//死信佇列
string dlxQueueName = "dlx.queue";
//訊息交換機
string exchange = "direct-exchange";
//訊息佇列
string queueName = "delay_queue";
using (var connection = RabbitMQHelper.GetConnection())
{
using (var channel = connection.CreateModel())
{
//建立死信交換機
channel.ExchangeDeclare(dlxexChange,autoDelete: false);
//建立死信佇列
客棧 channel.QueueDeclare(dlxQueueName,延時:5s");
}
}
}
執行效果:
這會得到了想要的效果。
RabbitMQ管理介面:
四、延時訊息用延時外掛的方式實現
相比上面第三的延時訊息,這裡的外掛方式會顯的更加簡單,也推薦用這種。
因為這裡只需要一個交換機和一個對佇列,生產者向佇列傳送訊息,會直接是延時才會到佇列。
安裝外掛:
地址:https://www.rabbitmq.com/community-plugins.html
找到和自己RabbitMQ一樣的版本,下載下來上傳到,或F12檢視這個檔案的地址,直接Linux上下載(這裡用這種)
Linux下載外掛:
#下載外掛
wget https://.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.9.0/rabbitmq_delayed_message_exchange-3.9.0.ez
已經下載到Linux上
#把檔案複製到rabbitmq 容器下的plugins資料夾
docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins
#進入rabbitmq docker容器
docker exec -it rabbitmq bash
#開啟外掛:
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
做完上面這些在RabbitMQ管理介面可以看到多了一個延時訊息的交換機。
外掛裝好了,生產者程式碼:
public static void SendMessage()
{
//延時訊息交換機
string delayExchange = "delay.exchange";
//延時訊息佇列
string delayQueueName = "delay_queue";
using (var connection = RabbitMQHelper.GetConnection())
{
using (var channel = connection.CreateModel())
{
Dictionary<string,object> args = new Dictionary<string,object>();
args.Add("x-delayed-type","direct"); //x-delayed-type必須加
//建立延時交換機,type型別為x-delayed-message
channel.ExchangeDeclare(delayExchange,type: "x-delayed-message",arguments: args);
//建立延時訊息佇列
www.cppcns.com channel.QueueDeclare(delayQueueName,autoDelete: false);
//交換機繫結佇列
channel.QueueBind(delayQueueName,delayExchange,routingKey: delayQueueName);
string message = "hello rabbitmq message 10s後處理";
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
//延時時間從header賦值
Dictionary<string,object> headers = new Dictionary<string,object>();
headers.Add("x-delay",10000);
properties.Headers = headers;
//釋出訊息,按時10s
channel.BasicPublish(exchange: delayExchange,routingKey: delayQueueName,向佇列:{delayQueueName}傳送訊息:{message},延時:10s");
string message2 = "hello rabbitmq message 5s後處理";
var properties2 = channel.CreateBasicProperties();
properties2.Persistent = true;
//延時時間從header賦值
Dictionary<string,object> headers2 = new Dictionary<string,object>();
headers2.Add("x-delay",5000);
properties2.Headers = headers2;
//釋出訊息,延時5s
channel.BasicPublish(exchange: delayExchange,向佇列:{delayQueueName}傳送訊息:{message2},延時:5s");
}
}
}
消費者程式碼:
public static void DelayMessageConsumer() { //延時佇列 string queueName = "delay_queue"; var connection = RabbitMQHelper.GetConnection(); { //建立通道 var channel = connection.CreateModel(); { var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0,ea) => { //處理業務 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{DateTime.Now},接收到訊息:{message}"); channel.BasicAck(ea.DeliveryTag,false); }; channel.BasicConsume(queueName,延時佇列" loading="lazy" src="http://img1.cppcns.com/images/2021/202109/e4t2jqir5em.png" />RabbitMQ管理介面,只有一個佇列:
到此這篇關於運用.NetCore例項講解RabbitMQ死信佇列,延時佇列的文章就介紹到這了,更多相關.NetCore RabbitMQ死信佇列,延時佇列內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們!