C#實(shí)現(xiàn)rabbitmq 延遲隊(duì)列功能實(shí)例代碼
最近在研究rabbitmq,項(xiàng)目中有這樣一個(gè)場(chǎng)景:在用戶(hù)要支付訂單的時(shí)候,如果超過(guò)30分鐘未支付,會(huì)把訂單關(guān)掉。當(dāng)然我們可以做一個(gè)定時(shí)任務(wù),每個(gè)一段時(shí)間來(lái)掃描未支付的訂單,如果該訂單超過(guò)支付時(shí)間就關(guān)閉,但是在數(shù)據(jù)量小的時(shí)候并沒(méi)有什么大的問(wèn)題,但是數(shù)據(jù)量一大輪訓(xùn)數(shù)據(jù)庫(kù)的方式就會(huì)變得特別耗資源。當(dāng)面對(duì)千萬(wàn)級(jí)、上億級(jí)數(shù)據(jù)量時(shí),本身寫(xiě)入的IO就比較高,導(dǎo)致長(zhǎng)時(shí)間查詢(xún)或者根本就查不出來(lái),更別說(shuō)分庫(kù)分表以后了。除此之外,還有優(yōu)先級(jí)隊(duì)列,基于優(yōu)先級(jí)隊(duì)列的JDK延遲隊(duì)列,時(shí)間輪等方式。但如果系統(tǒng)的架構(gòu)中本身就有RabbitMQ的話(huà),那么選擇RabbitMQ來(lái)實(shí)現(xiàn)類(lèi)似的功能也是一種選擇。 我們項(xiàng)目中用到了rabbitmq,可以做一個(gè)延遲隊(duì)列完美的解決這個(gè)問(wèn)題。
rabbitmq本身不具有延時(shí)消息隊(duì)列的功能,但是可以通過(guò)TTL(Time To Live)、DLX(Dead Letter Exchanges)特性實(shí)現(xiàn)。其原理給消息設(shè)置過(guò)期時(shí)間,在消息隊(duì)列上為過(guò)期消息指定轉(zhuǎn)發(fā)器,這樣消息過(guò)期后會(huì)轉(zhuǎn)發(fā)到與指定轉(zhuǎn)發(fā)器匹配的隊(duì)列上,變向?qū)崿F(xiàn)延時(shí)隊(duì)列。利用rabbitmq的這種特性,應(yīng)該有了一個(gè)大概的思路。、
網(wǎng)上搜了一下 rabbitmq-delayed-message-exchange 這個(gè)插件也可以實(shí)現(xiàn)延遲隊(duì)列的功能。今天介紹的是如何用C#來(lái)實(shí)現(xiàn)。
首先了解一下TTL和DLX
消息的TTL(Time To Live)
消息的TTL就是消息的存活時(shí)間。RabbitMQ可以對(duì)隊(duì)列和消息分別設(shè)置TTL。對(duì)隊(duì)列設(shè)置就是隊(duì)列沒(méi)有消費(fèi)者連著的保留時(shí)間,也可以對(duì)每一個(gè)單獨(dú)的消息做單獨(dú)的設(shè)置。超過(guò)了這個(gè)時(shí)間,我們認(rèn)為這個(gè)消息就死了,稱(chēng)之為死信。如果隊(duì)列設(shè)置了,消息也設(shè)置了,那么會(huì)取小的。所以一個(gè)消息如果被路由到不同的隊(duì)列中,這個(gè)消息死亡的時(shí)間有可能不一樣(不同的隊(duì)列設(shè)置)。這里單講單個(gè)消息的TTL,因?yàn)樗攀菍?shí)現(xiàn)延遲任務(wù)的關(guān)鍵。
Dead Letter Exchanges
Exchage的概念在這里就不在贅述。一個(gè)消息在滿(mǎn)足如下條件下,會(huì)進(jìn)死信路由,記住這里是路由而不是隊(duì)列,一個(gè)路由可以對(duì)應(yīng)很多隊(duì)列。
1. 一個(gè)消息被Consumer拒收了,并且reject方法的參數(shù)里requeue是false。也就是說(shuō)不會(huì)被再次放在隊(duì)列里,被其他消費(fèi)者使用。
2. 上面的消息的TTL到了,消息過(guò)期了。
3. 隊(duì)列的長(zhǎng)度限制滿(mǎn)了。排在前面的消息會(huì)被丟棄或者扔到死信路由上。
Dead Letter Exchange其實(shí)就是一種普通的exchange,和創(chuàng)建其他exchange沒(méi)有兩樣。只是在某一個(gè)設(shè)置Dead Letter Exchange的隊(duì)列中有消息過(guò)期了,會(huì)自動(dòng)觸發(fā)消息的轉(zhuǎn)發(fā),發(fā)送到Dead Letter Exchange中去。
首先我建了兩個(gè)控制臺(tái)項(xiàng)目一個(gè)是生產(chǎn)者,一個(gè)是消費(fèi)者。
生產(chǎn)者代碼如下
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);//隊(duì)列上消息過(guò)期時(shí)間,應(yīng)小于隊(duì)列過(guò)期時(shí)間 dic.Add("x-dead-letter-exchange", "exchange-direct");//過(guò)期消息轉(zhuǎn)向路由 dic.Add("x-dead-letter-routing-key", "routing-delay");//過(guò)期消息轉(zhuǎn)向路由相匹配routingkey //創(chuàng)建一個(gè)名叫"zzhello"的消息隊(duì)列 channel.QueueDeclare(queue: "zzhello", durable: true, exclusive: false, autoDelete: false, arguments: dic); var message = "Hello World!"; var body = Encoding.UTF8.GetBytes(message); //向該消息隊(duì)列發(fā)送消息message channel.BasicPublish(exchange: "", routingKey: "zzhello", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } } } Console.ReadKey();
消費(fèi)者代碼如下:
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"); //回調(diào),當(dāng)consumer收到消息后會(huì)執(zhí)行該函數(shù) 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); //消費(fèi)隊(duì)列"hello"中的消息 channel.BasicConsume(queue: name, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } Console.ReadKey();
效果 :
在等待了12秒后消費(fèi)者等到了消息。
這樣我們就實(shí)現(xiàn)了延遲隊(duì)列的功能了。
以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持我們。
欄 目:C#教程
下一篇:ASP.NET C#中Application的用法教程
本文標(biāo)題:C#實(shí)現(xiàn)rabbitmq 延遲隊(duì)列功能實(shí)例代碼
本文地址:http://mengdiqiu.com.cn/a1/C_jiaocheng/5767.html
您可能感興趣的文章
- 01-10C#實(shí)現(xiàn)txt定位指定行完整實(shí)例
- 01-10WinForm實(shí)現(xiàn)仿視頻 器左下角滾動(dòng)新聞效果的方法
- 01-10C#實(shí)現(xiàn)清空回收站的方法
- 01-10C#實(shí)現(xiàn)讀取注冊(cè)表監(jiān)控當(dāng)前操作系統(tǒng)已安裝軟件變化的方法
- 01-10C#實(shí)現(xiàn)多線(xiàn)程下載文件的方法
- 01-10C#實(shí)現(xiàn)Winform中打開(kāi)網(wǎng)頁(yè)頁(yè)面的方法
- 01-10C#實(shí)現(xiàn)遠(yuǎn)程關(guān)閉計(jì)算機(jī)或重啟計(jì)算機(jī)的方法
- 01-10C#自定義簽名章實(shí)現(xiàn)方法
- 01-10C#文件斷點(diǎn)續(xù)傳實(shí)現(xiàn)方法
- 01-10winform實(shí)現(xiàn)創(chuàng)建最前端窗體的方法


閱讀排行
- 1C語(yǔ)言 while語(yǔ)句的用法詳解
- 2java 實(shí)現(xiàn)簡(jiǎn)單圣誕樹(shù)的示例代碼(圣誕
- 3利用C語(yǔ)言實(shí)現(xiàn)“百馬百擔(dān)”問(wèn)題方法
- 4C語(yǔ)言中計(jì)算正弦的相關(guān)函數(shù)總結(jié)
- 5c語(yǔ)言計(jì)算三角形面積代碼
- 6什么是 WSH(腳本宿主)的詳細(xì)解釋
- 7C++ 中隨機(jī)函數(shù)random函數(shù)的使用方法
- 8正則表達(dá)式匹配各種特殊字符
- 9C語(yǔ)言十進(jìn)制轉(zhuǎn)二進(jìn)制代碼實(shí)例
- 10C語(yǔ)言查找數(shù)組里數(shù)字重復(fù)次數(shù)的方法
本欄相關(guān)
- 01-10C#通過(guò)反射獲取當(dāng)前工程中所有窗體并
- 01-10關(guān)于A(yíng)SP網(wǎng)頁(yè)無(wú)法打開(kāi)的解決方案
- 01-10WinForm限制窗體不能移到屏幕外的方法
- 01-10WinForm繪制圓角的方法
- 01-10C#實(shí)現(xiàn)txt定位指定行完整實(shí)例
- 01-10WinForm實(shí)現(xiàn)仿視頻 器左下角滾動(dòng)新
- 01-10C#停止線(xiàn)程的方法
- 01-10C#實(shí)現(xiàn)清空回收站的方法
- 01-10C#通過(guò)重寫(xiě)Panel改變邊框顏色與寬度的
- 01-10C#實(shí)現(xiàn)讀取注冊(cè)表監(jiān)控當(dāng)前操作系統(tǒng)已
隨機(jī)閱讀
- 01-11Mac OSX 打開(kāi)原生自帶讀寫(xiě)NTFS功能(圖文
- 01-10使用C語(yǔ)言求解撲克牌的順子及n個(gè)骰子
- 08-05DEDE織夢(mèng)data目錄下的sessions文件夾有什
- 01-11ajax實(shí)現(xiàn)頁(yè)面的局部加載
- 01-10delphi制作wav文件的方法
- 08-05織夢(mèng)dedecms什么時(shí)候用欄目交叉功能?
- 08-05dedecms(織夢(mèng))副欄目數(shù)量限制代碼修改
- 01-10SublimeText編譯C開(kāi)發(fā)環(huán)境設(shè)置
- 04-02jquery與jsp,用jquery
- 01-10C#中split用法實(shí)例總結(jié)