RabbitMQ消息有效期與死信的處理過程是什么

這篇文章主要講解了“RabbitMQ消息有效期與死信的處理過程是什么”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“RabbitMQ消息有效期與死信的處理過程是什么”吧！

一.前言

RabbitMQ的TTL全稱為Time-To-Live，表示的是消息的有效期。消息如果在隊列中一直沒有被消費并且存在時間超過了TTL，消息就會變成了"死信" (Dead Message)，后續無法再被消費了。如果不設置TTL，則表示此消息永久有效（默認消息是不會失效的）。如果將TTL設為0，則表示如果消息不能被立馬消費則會被立即丟掉，這個特性可以部分替代RabbitMQ3.0以前支持的immediate參數，之所以所部分代替，是應為immediate參數在投遞失敗會有basic.return方法將消息體返回（這個功能可以利用死信隊列來實現）。

設置TTL有兩種方式：

• 隊列有效期：是聲明隊列的時候，在隊列的屬性中設置，這樣該隊列中的消息都會有相同的有效期

• 消息有效期：發送消息時給消息設置屬性，可以為每條消息都設置不同的TTL

如果兩種方式都設置了，則以設置的較小的為準。

• 區別：如果聲明隊列時設置了有效期，則消息過期了就會被刪掉；如果是發消息時設置的有效期，消息過期了也不會被立馬刪掉，因為這時消息是否過期是在要投遞給消費者時判斷的。

二.設置消息有效期

1.設置隊列的有效期TTL

定義隊列的方法如下：

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

該方法的arguments參數可以設置隊列的屬性，屬性名為x-message-ttl，單位為毫秒。后臺添加的話如下：

代碼中設置如下：

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

命令行模式來設置：

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

通過HTTP接口調用：

$ 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.設置隊列的有效期Expire

有效期Expire可以讓隊列在指定時間內 “未被使用” 的話會自動過期刪除，未使用的意思是 queue 上沒有任何 consumer，queue 沒有被重新聲明，并且在過期時間段內未調用過 basic.get 命令。該方式可用于，例如，RPC-style 的回復 queue, 其中許多queue 會被創建出來，但是卻從未被使用。

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

代碼如下：

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

3.通過發送消息時設置有效期

發送消息的方法如下：

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

在該方法的props參數可以設置其有效期：

       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());

通過HTTPAPI 接口設置：

$ 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

三.死信交換機DLX

介紹

• 死信隊列：DLX，dead-letter-exchange

• 利用DLX，當消息在一個隊列中變成死信 (dead message) 之后，它能被重新publish到另一個Exchange，這個Exchange就是DLX

消息變成死信幾種情況

• 消息被拒絕(Basic.Reject/Basic.Nack) ，井且設置requeue 參數為false

• 消息過期

• 隊列達到最大長度

死信處理過程

• DLX也是一個正常的Exchange，和一般的Exchange沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性。

• 當這個隊列中有死信時，RabbitMQ就會自動的將這個消息重新發布到設置的Exchange上去，進而被路由到另一個隊列。

• 可以監聽這個隊列中的消息做相應的處理。

用途

通過監控消費死信隊列中消息，來觀察和分析數據。
結合TTL實現延遲隊列（比如下單超過多長時間自動關閉）

使用

代碼如下：

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

實例

public static void main(String[] args) throws Exception {
		Connection connection = ConnectionUtil.getConnection();
		Channel channel = connection.createChannel();
		//聲明一個交換機，做死信交換機用
		channel.exchangeDeclare("dlx_exchange", "topic", true, false, null);
		//聲明一個隊列，做死信隊列用
		channel.queueDeclare("dlx_queue", true, false, false, null);
		//隊列綁定到交換機上
		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);//設置消息有效期1秒，過期后變成私信消息，然后進入DLX
		arguments.put("x-dead-letter-exchange" , "dlx_exchange");//設置DLX
		arguments.put("x-dead-letter-routing-key" , "dlx.test");//設置DLX的路由鍵
		//為隊列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, ("測試死信消息").getBytes());
		System.out.println("發送消息時間:"+ConnectionUtil.formatDate(new Date()));
		channel.close();
		connection.close();
	}

說明：

申明死信隊列dlx_queue的綁定如下，與死信交換機dlx_exchange（topic類型）進行綁定，routing key為"dlx.*"
申明隊列normal_queue，與交換機normal_exchange（fanout類型）進行綁定

執行流程：

• 消息發送到交換機normal_exchange，然后路由到隊列normal_queue上

• 因為隊列normal_queue沒有消費者，消息過期后成為死信消息

• 死信消息攜帶設置的x-dead-letter-routing-key=dlx.test進入到死信交換機dlx_exechage

• dlx_exechage與dlx_queue綁定的routing key為"dlx.*"，死信消息的路由鍵dlx.test符合該規則被路由到dlx.queue上面。

感謝各位的閱讀，以上就是“RabbitMQ消息有效期與死信的處理過程是什么”的內容了，經過本文的學習后，相信大家對RabbitMQ消息有效期與死信的處理過程是什么這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！