C#實現rabbitmq 延遲隊列功能實例代碼。本站提示廣大學習愛好者:(C#實現rabbitmq 延遲隊列功能實例代碼)文章只能為提供參考,不一定能成為您想要的結果。以下是C#實現rabbitmq 延遲隊列功能實例代碼正文
最近在研究rabbitmq,項目中有這樣一個場景:在用戶要支付訂單的時候,如果超過30分鐘未支付,會把訂單關掉。當然我們可以做一個定時任務,每個一段時間來掃描未支付的訂單,如果該訂單超過支付時間就關閉,但是在數據量小的時候並沒有什麼大的問題,但是數據量一大輪訓數據庫的方式就會變得特別耗資源。當面對千萬級、上億級數據量時,本身寫入的IO就比較高,導致長時間查詢或者根本就查不出來,更別說分庫分表以後了。除此之外,還有優先級隊列,基於優先級隊列的JDK延遲隊列,時間輪等方式。但如果系統的架構中本身就有RabbitMQ的話,那麼選擇RabbitMQ來實現類似的功能也是一種選擇。 我們項目中用到了rabbitmq,可以做一個延遲隊列完美的解決這個問題。
rabbitmq本身不具有延時消息隊列的功能,但是可以通過TTL(Time To Live)、DLX(Dead Letter Exchanges)特性實現。其原理給消息設置過期時間,在消息隊列上為過期消息指定轉發器,這樣消息過期後會轉發到與指定轉發器匹配的隊列上,變向實現延時隊列。利用rabbitmq的這種特性,應該有了一個大概的思路。、
網上搜了一下 rabbitmq-delayed-message-exchange 這個插件也可以實現延遲隊列的功能。今天介紹的是如何用C#來實現。
首先了解一下TTL和DLX
消息的TTL(Time To Live)
消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設置TTL。對隊列設置就是隊列沒有消費者連著的保留時間,也可以對每一個單獨的消息做單獨的設置。超過了這個時間,我們認為這個消息就死了,稱之為死信。如果隊列設置了,消息也設置了,那麼會取小的。所以一個消息如果被路由到不同的隊列中,這個消息死亡的時間有可能不一樣(不同的隊列設置)。這裡單講單個消息的TTL,因為它才是實現延遲任務的關鍵。
Dead Letter Exchanges
Exchage的概念在這裡就不在贅述。一個消息在滿足如下條件下,會進死信路由,記住這裡是路由而不是隊列,一個路由可以對應很多隊列。
1. 一個消息被Consumer拒收了,並且reject方法的參數裡requeue是false。也就是說不會被再次放在隊列裡,被其他消費者使用。
2. 上面的消息的TTL到了,消息過期了。
3. 隊列的長度限制滿了。排在前面的消息會被丟棄或者扔到死信路由上。
Dead Letter Exchange其實就是一種普通的exchange,和創建其他exchange沒有兩樣。只是在某一個設置Dead Letter Exchange的隊列中有消息過期了,會自動觸發消息的轉發,發送到Dead Letter Exchange中去。
首先我建了兩個控制台項目一個是生產者,一個是消費者。
生產者代碼如下
var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "test", Password = "test" }; using (var connection = factory.CreateConnection()) { while (Console.ReadLine() != null) { using (var channel = connection.CreateModel()) { Dictionary<string, object> dic = new Dictionary<string, object>(); dic.Add("x-expires", 30000); dic.Add("x-message-ttl", 12000);//隊列上消息過期時間,應小於隊列過期時間 dic.Add("x-dead-letter-exchange", "exchange-direct");//過期消息轉向路由 dic.Add("x-dead-letter-routing-key", "routing-delay");//過期消息轉向路由相匹配routingkey //創建一個名叫"zzhello"的消息隊列 channel.QueueDeclare(queue: "zzhello", durable: true, exclusive: false, autoDelete: false, arguments: dic); var message = "Hello World!"; var body = Encoding.UTF8.GetBytes(message); //向該消息隊列發送消息message channel.BasicPublish(exchange: "", routingKey: "zzhello", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } } } Console.ReadKey();
消費者代碼如下:
var factory = new ConnectionFactory() { HostName = "127.0.01", UserName = "test", Password = "test" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.ExchangeDeclare(exchange: "exchange-direct", type: "direct"); string name = channel.QueueDeclare().QueueName; channel.QueueBind(queue: name, exchange: "exchange-direct", routingKey: "routing-delay"); //回調,當consumer收到消息後會執行該函數 var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine(ea.RoutingKey); Console.WriteLine(" [x] Received {0}", message); }; //Console.WriteLine("name:" + name); //消費隊列"hello"中的消息 channel.BasicConsume(queue: name, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } Console.ReadKey();
效果 :
在等待了12秒後消費者等到了消息。
這樣我們就實現了延遲隊列的功能了。
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持。