作者:海向
來源:https://www.cnblogs.com/haixiang/p/10905189.html
死信隊列
死信隊列:沒有被及時消費的消息存放的隊列。
消息沒有被及時消費的原因:
- a.消息被拒絕(basic.reject/ basic.nack)並且不再重新投遞 requeue=false
- b.TTL(time-to-live) 消息超時未消費
- c.達到最大隊列長度
實現死信隊列步驟
首先需要設置死信隊列的 exchange 和 queue,然后進行綁定:
Exchange: dlx.exchange
Queue: dlx.queue
RoutingKey: # 代表接收所有路由 key
然后我們進行正常聲明交換機、隊列、綁定,只不過我們需要在普通隊列加上一個參數即可: arguments.put("x-dead-letter-exchange",' dlx.exchange' )
這樣消息在過期、requeue失敗、 隊列在達到最大長度時,消息就可以直接路由到死信隊列!
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class DlxProducer {
public static void main(String[] args) throws Exception {
//設置連接以及創建 channel 湖綠
String exchangeName = "test_dlx_exchange";
String routingKey = "item.update";
String msg = "this is dlx msg";
//我們設置消息過期時間,10秒后再消費 讓消息進入死信隊列
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
.deliveryMode(2)
.expiration("10000")
.build();
channel.basicPublish(exchangeName, routingKey, true, properties, msg.getBytes());
System.out.println("Send message : " + msg);
channel.close();
connection.close();
}
}
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class DlxConsumer {
public static void main(String[] args) throws Exception {
//創建連接、創建channel忽略 內容可以在上面代碼中獲取
String exchangeName = "test_dlx_exchange";
String queueName = "test_dlx_queue";
String routingKey = "item.#";
//必須設置參數到 arguments 中
Map<String, Object> arguments = new HashMap<String, Object>();
arguments.put("x-dead-letter-exchange", "dlx.exchange");
channel.exchangeDeclare(exchangeName, "topic", true, false, null);
//將 arguments 放入隊列的聲明中
channel.queueDeclare(queueName, true, false, false, arguments);
//一般不用代碼綁定,在管理界面手動綁定
channel.queueBind(queueName, exchangeName, routingKey);
//聲明死信隊列
channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
channel.queueDeclare("dlx.queue", true, false, false, null);
//路由鍵為 # 代表可以路由到所有消息
channel.queueBind("dlx.queue", "dlx.exchange", "#");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
//6. 設置 Channel 消費者綁定隊列
channel.basicConsume(queueName, true, consumer);
}
}
總結
DLX也是一個正常的 Exchange,和一般的 Exchange 沒有區別,它能在任何的隊列上被指定,實際上就是設置某個隊列的屬性。
當這個隊列中有死信時,RabbitMQ 就會自動的將這個消息重新發布到設置的 Exchange 上去,進而被路由到另一個隊列。可以監聽這個隊列中消息做相應的處理。
近期熱文推薦:
1.Java 15 正式發布, 14 個新特性,刷新你的認知!!
2.終於靠開源項目弄到 IntelliJ IDEA 激活碼了,真香!
3.我用 Java 8 寫了一段邏輯,同事直呼看不懂,你試試看。。
覺得不錯,別忘了隨手點贊+轉發哦!