快捷導(dǎo)航

RabbitMQ消息有效期與死信的處理過(guò)程

更新時(shí)間：2022年03月01日 10:30:08 作者：lss0555

利用DLX，當(dāng)消息在一個(gè)隊(duì)列中變成死信?(dead?message)?之后，它能被重新publish到另一個(gè)Exchange，這個(gè)Exchange就是DLX，本文重點(diǎn)給大家介紹RabbitMQ消息有效期與死信的相關(guān)知識(shí)，感興趣的朋友跟隨小編一起看看吧

一.前言

RabbitMQ的TTL全稱(chēng)為T(mén)ime-To-Live，表示的是消息的有效期。消息如果在隊(duì)列中一直沒(méi)有被消費(fèi)并且存在時(shí)間超過(guò)了TTL，消息就會(huì)變成了"死信" (Dead Message)，后續(xù)無(wú)法再被消費(fèi)了。如果不設(shè)置TTL，則表示此消息永久有效（默認(rèn)消息是不會(huì)失效的）。如果將TTL設(shè)為0，則表示如果消息不能被立馬消費(fèi)則會(huì)被立即丟掉，這個(gè)特性可以部分替代RabbitMQ3.0以前支持的immediate參數(shù)，之所以所部分代替，是應(yīng)為immediate參數(shù)在投遞失敗會(huì)有basic.return方法將消息體返回（這個(gè)功能可以利用死信隊(duì)列來(lái)實(shí)現(xiàn)）。

設(shè)置TTL有兩種方式：

隊(duì)列有效期：是聲明隊(duì)列的時(shí)候，在隊(duì)列的屬性中設(shè)置，這樣該隊(duì)列中的消息都會(huì)有相同的有效期
消息有效期：發(fā)送消息時(shí)給消息設(shè)置屬性，可以為每條消息都設(shè)置不同的TTL

如果兩種方式都設(shè)置了，則以設(shè)置的較小的為準(zhǔn)。

區(qū)別：如果聲明隊(duì)列時(shí)設(shè)置了有效期，則消息過(guò)期了就會(huì)被刪掉；如果是發(fā)消息時(shí)設(shè)置的有效期，消息過(guò)期了也不會(huì)被立馬刪掉，因?yàn)檫@時(shí)消息是否過(guò)期是在要投遞給消費(fèi)者時(shí)判斷的。

二.設(shè)置消息有效期

1.設(shè)置隊(duì)列的有效期TTL

定義隊(duì)列的方法如下：

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                                 Map<String, Object> arguments) throws IOException;

該方法的arguments參數(shù)可以設(shè)置隊(duì)列的屬性，屬性名為x-message-ttl，單位為毫秒。后臺(tái)添加的話(huà)如下：

在這里插入圖片描述

代碼中設(shè)置如下：

Map<String, Object> arguments= new HashMap<String , Object>();
arguments.put("x-message-ttl " , 10000);//10秒鐘  單位為毫秒
channel.queueDeclare(queueName , durable , exclusive , autoDelete , arguments) ;

命令行模式來(lái)設(shè)置：

rabbitmqctl set_policy TTL ".*" '{"message-ttl":100000}' --apply-to queues

通過(guò)HTTP接口調(diào)用：

$ curl -i -u guest:guest -H "content-type:application/json"  -XPUT -d'{"auto_delete":false,"durable":true,"arguments":{"x-message-ttl": 100000}}' 
http://ip:15672/api/queues/{vhost}/{queuename}

2.設(shè)置隊(duì)列的有效期Expire

有效期Expire可以讓隊(duì)列在指定時(shí)間內(nèi) “未被使用” 的話(huà)會(huì)自動(dòng)過(guò)期刪除，未使用的意思是 queue 上沒(méi)有任何 consumer，queue 沒(méi)有被重新聲明，并且在過(guò)期時(shí)間段內(nèi)未調(diào)用過(guò) basic.get 命令。該方式可用于，例如，RPC-style 的回復(fù) queue, 其中許多queue 會(huì)被創(chuàng)建出來(lái)，但是卻從未被使用。

服務(wù)器會(huì)確保在過(guò)期時(shí)間到達(dá)后 queue 被刪除，但是不保證刪除的動(dòng)作有多么的及時(shí)。在服務(wù)器重啟后，持久化的queue 的超時(shí)時(shí)間將重新計(jì)算。 x-expires 參數(shù)值以毫秒為單位，并且服從和 x-message-ttl 一樣的約束條件，且不能設(shè)置為 0 。所以，如果該參數(shù)設(shè)置為 10000 ，則表示該 queue 如果在 10s之內(nèi)未被使用則會(huì)被刪除。

代碼如下：

Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-expires", 10000); 
channel.queueDeclare("queue", false, false, false, args);

3.通過(guò)發(fā)送消息時(shí)設(shè)置有效期

發(fā)送消息的方法如下：

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;

在該方法的props參數(shù)可以設(shè)置其有效期：

       Map<String, Object> headers = new HashMap<String, Object>();
                        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                                .deliveryMode(2) // 消息持久
                                .contentEncoding("UTF-8") // 編碼方式
                                .contentType("text/plain")
                                .expiration("100000")
                                .headers(headers)
                                .build();
      channel.basicPublish("", queueName, properties, message.getBytes());

通過(guò)HTTPAPI 接口設(shè)置：

$ curl -i -u guest:guest -H "content-type:application/json"  -XPOST -d
'{"properties":{"expiration":"100000"},"routing_key":"routingkey","payload":"bodys","payload_encoding":"string"}'  
http://localhost:15672/api/exchanges/{vhost}/{exchangename}/publish

三.死信交換機(jī)DLX

介紹

死信隊(duì)列：DLX，dead-letter-exchange
利用DLX，當(dāng)消息在一個(gè)隊(duì)列中變成死信 (dead message) 之后，它能被重新publish到另一個(gè)Exchange，這個(gè)Exchange就是DLX

消息變成死信幾種情況

消息被拒絕(Basic.Reject/Basic.Nack) ，井且設(shè)置requeue 參數(shù)為false
消息過(guò)期
隊(duì)列達(dá)到最大長(zhǎng)度

死信處理過(guò)程

DLX也是一個(gè)正常的Exchange，和一般的Exchange沒(méi)有區(qū)別，它能在任何的隊(duì)列上被指定，實(shí)際上就是設(shè)置某個(gè)隊(duì)列的屬性。
當(dāng)這個(gè)隊(duì)列中有死信時(shí)，RabbitMQ就會(huì)自動(dòng)的將這個(gè)消息重新發(fā)布到設(shè)置的Exchange上去，進(jìn)而被路由到另一個(gè)隊(duì)列。
可以監(jiān)聽(tīng)這個(gè)隊(duì)列中的消息做相應(yīng)的處理。

用途

通過(guò)監(jiān)控消費(fèi)死信隊(duì)列中消息，來(lái)觀察和分析數(shù)據(jù)。
結(jié)合TTL實(shí)現(xiàn)延遲隊(duì)列（比如下單超過(guò)多長(zhǎng)時(shí)間自動(dòng)關(guān)閉）

使用

代碼如下：

channel.exchangeDeclare("dlx_exchange" , "direct"); //創(chuàng)建DLX: dlx_exchange
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange" , "dlx_exchange ");//設(shè)置死信交換機(jī)
args.put("x-dead-letter-routing-key" , "dlx-routing-key");//設(shè)置DLX的路由鍵(可以不設(shè)置)
channel.queueDeclare("myqueue" , false , false , false , args);

實(shí)例

public static void main(String[] args) throws Exception {
		Connection connection = ConnectionUtil.getConnection();
		Channel channel = connection.createChannel();
		//聲明一個(gè)交換機(jī)，做死信交換機(jī)用
		channel.exchangeDeclare("dlx_exchange", "topic", true, false, null);
		//聲明一個(gè)隊(duì)列，做死信隊(duì)列用
		channel.queueDeclare("dlx_queue", true, false, false, null);
		//隊(duì)列綁定到交換機(jī)上
		channel.queueBind("dlx_queue", "dlx_exchange", "dlx.*");
		
		channel.exchangeDeclare("normal_exchange", "fanout", true, false, null);
		Map<String, Object> arguments=new HashMap<String, Object>();
		arguments.put("x-message-ttl" , 1000);//設(shè)置消息有效期1秒，過(guò)期后變成私信消息，然后進(jìn)入DLX
		arguments.put("x-dead-letter-exchange" , "dlx_exchange");//設(shè)置DLX
		arguments.put("x-dead-letter-routing-key" , "dlx.test");//設(shè)置DLX的路由鍵
		//為隊(duì)列normal_queue 添加DLX
		channel.queueDeclare("normal_queue", true, false, false, arguments);
		channel.queueBind("normal_queue", "normal_exchange", "");
		channel.basicPublish("normal_exchange", "", MessageProperties.PERSISTENT_TEXT_PLAIN, ("測(cè)試死信消息").getBytes());
		System.out.println("發(fā)送消息時(shí)間:"+ConnectionUtil.formatDate(new Date()));
		channel.close();
		connection.close();
	}

說(shuō)明：

申明死信隊(duì)列dlx_queue的綁定如下，與死信交換機(jī)dlx_exchange（topic類(lèi)型）進(jìn)行綁定，routing key為"dlx.*"
申明隊(duì)列normal_queue，與交換機(jī)normal_exchange（fanout類(lèi)型）進(jìn)行綁定

執(zhí)行流程：

消息發(fā)送到交換機(jī)normal_exchange，然后路由到隊(duì)列normal_queue上
因?yàn)殛?duì)列normal_queue沒(méi)有消費(fèi)者，消息過(guò)期后成為死信消息
死信消息攜帶設(shè)置的x-dead-letter-routing-key=dlx.test進(jìn)入到死信交換機(jī)dlx_exechage
dlx_exechage與dlx_queue綁定的routing key為"dlx.*"，死信消息的路由鍵dlx.test符合該規(guī)則被路由到dlx.queue上面。

參考：

https://www.jianshu.com/p/986ee5eb78bc

https://blog.csdn.net/u012988901/article/details/88958654

到此這篇關(guān)于RabbitMQ之消息有效期與死信的文章就介紹到這了,更多相關(guān)RabbitMQ消息有效期內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: