運(yùn)用.NetCore實(shí)例講解RabbitMQ死信隊(duì)列,延時(shí)隊(duì)列
一、死信隊(duì)列
描述:Q1隊(duì)列綁定了x-dead-letter-exchange(死信交換機(jī))為X2,x-dead-letter-routing-key(死信路由key)指向Q2(隊(duì)列2)
P(生產(chǎn)者)發(fā)送消息經(jīng)X1(交換機(jī)1)路由到Q1(隊(duì)列1),Q1的消息觸發(fā)特定情況,自動(dòng)把消息經(jīng)X2(交換機(jī)2)路由到Q2(隊(duì)列2),C(消費(fèi)者)直接消息Q2的消息。
特定情況有哪些呢:
- 1.消息被拒(basic.reject or basic.nack)并且沒有重新入隊(duì)(requeue=false);
- 2.當(dāng)前隊(duì)列中的消息數(shù)量已經(jīng)超過最大長(zhǎng)度(創(chuàng)建隊(duì)列時(shí)指定" x-max-length參數(shù)設(shè)置隊(duì)列最大消息數(shù)量)。
- 3.消息在隊(duì)列中過期,即當(dāng)前消息在隊(duì)列中的存活時(shí)間已經(jīng)超過了預(yù)先設(shè)置的TTL(Time To Live)時(shí)間;
這里演示情況1:
假如場(chǎng)景:Q1中隊(duì)列數(shù)據(jù)不完整,就算從新處理也會(huì)報(bào)錯(cuò),那就可以不ack,把這個(gè)消息轉(zhuǎn)到死信隊(duì)列另外處理。
生產(chǎn)者:
public static void SendMessage() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; //消息交換機(jī) string exchange = "direct-exchange"; //消息隊(duì)列 string queueName = "queue_a"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 創(chuàng)建消息交換機(jī) channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建消息隊(duì)列,并指定死信隊(duì)列 channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //設(shè)置當(dāng)前隊(duì)列的DLX(死信交換機(jī)) { "x-dead-letter-routing-key",dlxQueueName}, //設(shè)置DLX的路由key,DLX會(huì)根據(jù)該值去找到死信消息存放的隊(duì)列 }); //消息隊(duì)列綁定消息交換機(jī) channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; //發(fā)布消息 channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"向隊(duì)列:{queueName}發(fā)送消息:{message}"); } } }
消費(fèi)者:
public static void Consumer() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; //消息交換機(jī) string exchange = "direct-exchange"; //消息隊(duì)列 string queueName = "queue_a"; var connection = RabbitMQHelper.GetConnection(); { //創(chuàng)建信道 var channel = connection.CreateModel(); { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 創(chuàng)建消息交換機(jī) channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建消息隊(duì)列,并指定死信隊(duì)列 channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //設(shè)置當(dāng)前隊(duì)列的DLX { "x-dead-letter-routing-key",dlxQueueName}, //設(shè)置DLX的路由key,DLX會(huì)根據(jù)該值去找到死信消息存放的隊(duì)列 }); //消息隊(duì)列綁定消息交換機(jī) channel.QueueBind(queueName, exchange, routingKey: queueName); var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true); consumer.Received += (model, ea) => { //處理業(yè)務(wù) var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"隊(duì)列{queueName}消費(fèi)消息:{message},不做ack確認(rèn)"); //channel.BasicAck(ea.DeliveryTag, false); //不ack(BasicNack),且不把消息放回隊(duì)列(requeue:false) channel.BasicNack(ea.DeliveryTag, false, requeue: false); }; channel.BasicConsume(queueName, autoAck: false, consumer); } } }
消費(fèi)者加上channel.BasickNack()模擬消息處理不了,不ack確認(rèn)。
執(zhí)行結(jié)果:
RabbitMQ管理界面:
看到消息隊(duì)列為queue_a,特性有DLX(死信交換機(jī)),DLK(死信路由)。因?yàn)橄M(fèi)端不nack,觸發(fā)了死信,被轉(zhuǎn)發(fā)到了死信隊(duì)列dlx.queue。
二、延時(shí)隊(duì)列
延時(shí)隊(duì)列其實(shí)也是配合死信隊(duì)列一起用,其實(shí)就是上面死信隊(duì)列的第二中情況。給隊(duì)列添加消息過時(shí)時(shí)間(TTL),變成延時(shí)隊(duì)列。
簡(jiǎn)單的描述就是:P(生產(chǎn)者)發(fā)送消息到Q1(延時(shí)隊(duì)列),Q1的消息有過期時(shí)間,比如10s,那10s后消息過期就會(huì)觸發(fā)死信,從而把消息轉(zhuǎn)發(fā)到Q2(死信隊(duì)列)。
解決問題場(chǎng)景:像商城下單,未支付時(shí)取消訂單場(chǎng)景。下單時(shí)寫一條記錄入Q1,延時(shí)30分鐘后轉(zhuǎn)到Q2,消費(fèi)Q2,檢查訂單,支付則不做操作,沒支付則取消訂單,恢復(fù)庫存。
生產(chǎn)者代碼:
public static void SendMessage() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; //消息交換機(jī) string exchange = "direct-exchange"; //消息隊(duì)列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 創(chuàng)建消息交換機(jī) channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建消息隊(duì)列,并指定死信隊(duì)列,和設(shè)置這個(gè)隊(duì)列的消息過期時(shí)間為10s channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //設(shè)置當(dāng)前隊(duì)列的DLX(死信交換機(jī)) { "x-dead-letter-routing-key",dlxQueueName}, //設(shè)置DLX的路由key,DLX會(huì)根據(jù)該值去找到死信消息存放的隊(duì)列 { "x-message-ttl",10000} //設(shè)置隊(duì)列的消息過期時(shí)間 }); //消息隊(duì)列綁定消息交換機(jī) channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; //發(fā)布消息 channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{queueName}發(fā)送消息:{message}"); } } }
消費(fèi)者代碼:
public static void Consumer() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; var connection = RabbitMQHelper.GetConnection(); { //創(chuàng)建信道 var channel = connection.CreateModel(); { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true); consumer.Received += (model, ea) => { //處理業(yè)務(wù) var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{DateTime.Now},隊(duì)列{dlxQueueName}消費(fèi)消息:{message}"); channel.BasicAck(ea.DeliveryTag, false); }; channel.BasicConsume(dlxQueueName, autoAck: false, consumer); } } }
執(zhí)行代碼:
向延時(shí)隊(duì)列發(fā)送消息,監(jiān)聽死信隊(duì)列,發(fā)送和收到消息時(shí)間剛好是設(shè)置的10s。
RabbitMQ管理界面:
三、延時(shí)消息設(shè)置不同過期時(shí)間
上面的延時(shí)隊(duì)列能解決消息過期時(shí)間都是相同的場(chǎng)景,能不能解決消息的過期時(shí)間是不一樣的呢?
例如場(chǎng)景:機(jī)器人客服,為了更像人為操作,收到消息后要隨機(jī)3-10秒回復(fù)客戶。
- 1)隊(duì)列不設(shè)置TTL(消息過期時(shí)間),把過期時(shí)間設(shè)置在消息上。
生產(chǎn)者代碼:
public static void SendMessage() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; //消息交換機(jī) string exchange = "direct-exchange"; //消息隊(duì)列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 創(chuàng)建消息交換機(jī) channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建消息隊(duì)列,并指定死信隊(duì)列,和設(shè)置這個(gè)隊(duì)列的消息過期時(shí)間為10s channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //設(shè)置當(dāng)前隊(duì)列的DLX(死信交換機(jī)) { "x-dead-letter-routing-key",dlxQueueName}, //設(shè)置DLX的路由key,DLX會(huì)根據(jù)該值去找到死信消息存放的隊(duì)列 //{ "x-message-ttl",10000} //設(shè)置隊(duì)列的消息過期時(shí)間 }); //消息隊(duì)列綁定消息交換機(jī) channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message 10s后處理"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; properties.Expiration = "10000";//消息的有效期10s //發(fā)布消息,延時(shí)10s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{queueName}發(fā)送消息:{message},延時(shí):10s"); string message2 = "hello rabbitmq message 5s后處理"; var properties2 = channel.CreateBasicProperties(); properties2.Persistent = true; properties2.Expiration = "5000";//消息有效期5s //發(fā)布消息,延時(shí)5s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties2, body: Encoding.UTF8.GetBytes(message2)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{queueName}發(fā)送消息:{message2},延時(shí):5s"); } } }
消費(fèi)者代碼還是上面延時(shí)隊(duì)列的不變,先試下效果。
生產(chǎn)者向隊(duì)列中發(fā)送一條延時(shí)10s的消息再發(fā)一條延時(shí)5秒的消息,但消費(fèi)者卻先拿到延時(shí)10s的,再拿到延時(shí)5秒的,我想要的結(jié)果是先拿到延時(shí)5s的再拿到延時(shí)10s的,是什么原因呢。
原因是:隊(duì)列是先進(jìn)先出的,而RabbitMQ只會(huì)對(duì)首位第一條消息做檢測(cè),第一條沒過期,那么后面的消息就會(huì)阻塞住等待前面的過期。
解決辦法:增加一個(gè)消費(fèi)者對(duì)延時(shí)隊(duì)列消費(fèi),不ack,把第一條消息放到隊(duì)列尾部。一直讓消息在流動(dòng),這樣就能檢測(cè)到了。
- 2)新增消費(fèi)者代碼:
public static void SendMessage() { //死信交換機(jī) string dlxexChange = "dlx.exchange"; //死信隊(duì)列 string dlxQueueName = "dlx.queue"; //消息交換機(jī) string exchange = "direct-exchange"; //消息隊(duì)列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //創(chuàng)建死信交換機(jī) channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建死信隊(duì)列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信隊(duì)列綁定死信交換機(jī) channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 創(chuàng)建消息交換機(jī) channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //創(chuàng)建消息隊(duì)列,并指定死信隊(duì)列,和設(shè)置這個(gè)隊(duì)列的消息過期時(shí)間為10s channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //設(shè)置當(dāng)前隊(duì)列的DLX(死信交換機(jī)) { "x-dead-letter-routing-key",dlxQueueName}, //設(shè)置DLX的路由key,DLX會(huì)根據(jù)該值去找到死信消息存放的隊(duì)列 //{ "x-message-ttl",10000} //設(shè)置隊(duì)列的消息過期時(shí)間 }); //消息隊(duì)列綁定消息交換機(jī) channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message 10s后處理"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; properties.Expiration = "10000";//消息的有效期10s //發(fā)布消息,延時(shí)10s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{queueName}發(fā)送消息:{message},延時(shí):10s"); string message2 = "hello rabbitmq message 5s后處理"; var properties2 = channel.CreateBasicProperties(); properties2.Persistent = true; properties2.Expiration = "5000";//消息有效期5s //發(fā)布消息,延時(shí)5s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties2, body: Encoding.UTF8.GetBytes(message2)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{queueName}發(fā)送消息:{message2},延時(shí):5s"); } } }
執(zhí)行效果:
這會(huì)得到了想要的效果。
RabbitMQ管理界面:
四、延時(shí)消息用延時(shí)插件的方式實(shí)現(xiàn)
相比上面第三的延時(shí)消息,這里的插件方式會(huì)顯的更加簡(jiǎn)單,也推薦用這種。
因?yàn)檫@里只需要一個(gè)交換機(jī)和一個(gè)對(duì)隊(duì)列,生產(chǎn)者向隊(duì)列發(fā)送消息,會(huì)直接是延時(shí)才會(huì)到隊(duì)列。
安裝插件:
地址:https://www.rabbitmq.com/community-plugins.html
找到和自己RabbitMQ一樣的版本,下載下來上傳到Linux,或F12查看這個(gè)文件的地址,直接Linux上下載(這里用這種)
Linux下載插件:
#下載插件
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.9.0/rabbitmq_delayed_message_exchange-3.9.0.ez
已經(jīng)下載到Linux上
#把文件復(fù)制到rabbitmq docker容器下的plugins文件夾
docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins
#進(jìn)入rabbitmq docker容器
docker exec -it rabbitmq bash
#開啟插件:
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
做完上面這些在RabbitMQ管理界面可以看到多了一個(gè)延時(shí)消息的交換機(jī)。
插件裝好了,生產(chǎn)者代碼:
public static void SendMessage() { //延時(shí)消息交換機(jī) string delayExchange = "delay.exchange"; //延時(shí)消息隊(duì)列 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必須加 //創(chuàng)建延時(shí)交換機(jī),type類型為x-delayed-message channel.ExchangeDeclare(delayExchange, type: "x-delayed-message", durable: true, autoDelete: false,arguments: args); //創(chuàng)建延時(shí)消息隊(duì)列 channel.QueueDeclare(delayQueueName, durable: true, exclusive: false, autoDelete: false); //交換機(jī)綁定隊(duì)列 channel.QueueBind(delayQueueName, delayExchange, routingKey: delayQueueName); string message = "hello rabbitmq message 10s后處理"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; //延時(shí)時(shí)間從header賦值 Dictionary<string, object> headers = new Dictionary<string, object>(); headers.Add("x-delay", 10000); properties.Headers = headers; //發(fā)布消息,按時(shí)10s channel.BasicPublish(exchange: delayExchange, routingKey: delayQueueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{delayQueueName}發(fā)送消息:{message},延時(shí):10s"); string message2 = "hello rabbitmq message 5s后處理"; var properties2 = channel.CreateBasicProperties(); properties2.Persistent = true; //延時(shí)時(shí)間從header賦值 Dictionary<string, object> headers2 = new Dictionary<string, object>(); headers2.Add("x-delay", 5000); properties2.Headers = headers2; //發(fā)布消息,延時(shí)5s channel.BasicPublish(exchange: delayExchange, routingKey: delayQueueName, basicProperties: properties2, body: Encoding.UTF8.GetBytes(message2)); Console.WriteLine($"{DateTime.Now},向隊(duì)列:{delayQueueName}發(fā)送消息:{message2},延時(shí):5s"); } } }
消費(fèi)者代碼:
public static void DelayMessageConsumer() { //延時(shí)隊(duì)列 string queueName = "delay_queue"; var connection = RabbitMQHelper.GetConnection(); { //創(chuàng)建信道 var channel = connection.CreateModel(); { var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true); consumer.Received += (model, ea) => { //處理業(yè)務(wù) var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{DateTime.Now},接收到消息:{message}"); channel.BasicAck(ea.DeliveryTag, false); }; channel.BasicConsume(queueName, autoAck: false, consumer); } } }
執(zhí)行代碼:
RabbitMQ管理界面,只有一個(gè)隊(duì)列:
到此這篇關(guān)于運(yùn)用.NetCore實(shí)例講解RabbitMQ死信隊(duì)列,延時(shí)隊(duì)列的文章就介紹到這了,更多相關(guān).NetCore RabbitMQ死信隊(duì)列,延時(shí)隊(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)用.net core中實(shí)例講解RabbitMQ
- .NET Core實(shí)現(xiàn)RabbitMQ消息隊(duì)列的示例代碼
相關(guān)文章
ASP.NET2.0服務(wù)器控件之類型轉(zhuǎn)換器
ASP.NET2.0服務(wù)器控件之類型轉(zhuǎn)換器...2006-09-09DataGridView展開與收縮功能實(shí)現(xiàn)
我們今天將要講到DataGridView之行的展開與收縮,包括功能是如何實(shí)現(xiàn)的,感興趣的小伙伴們可以參考一下2015-09-09asp.net 請(qǐng)求輸入到輸出的全過程及httpHandler和httpModuler詳細(xì)介紹
看了幾篇講述httpHandler和HttpModuler的文章,雖然說沒有完全了解底層操作,但是我也算明白了一個(gè)請(qǐng)求從進(jìn)入IIS到最后輸出都經(jīng)歷了哪些過程,感興趣的朋友可以了解下2013-01-01