快捷導(dǎo)航

RabbitMQ中的死信隊(duì)列(Dead Letter Exchanges)詳解

更新時(shí)間：2023年12月12日 09:29:54 作者：warybee

這篇文章主要介紹了RabbitMQ中的死信隊(duì)列(Dead Letter Exchanges)詳解,當(dāng)RabbitMQ出現(xiàn)死信,可能會(huì)導(dǎo)致業(yè)務(wù)邏輯錯(cuò)誤,比如下訂單后修改庫存操作,在下單后因?yàn)槟撤N原因,發(fā)送的消息未被簽收,這時(shí)庫存數(shù)據(jù)會(huì)出現(xiàn)不一致,需要的朋友可以參考下

RabbitMQ死信隊(duì)列

1. 介紹

當(dāng)消息在一個(gè)隊(duì)列中變?yōu)樗佬藕螅恢匦掳l(fā)送到另一個(gè)Exchange。

2. 在什么情況下會(huì)出現(xiàn)死信

消息未被簽收，在消費(fèi)端使用了 basic.reject 或 basic.nack ，并且requeue設(shè)置為false
消息過期（TTL）
消息隊(duì)列達(dá)到了最大長(zhǎng)度

3. 實(shí)際應(yīng)用

當(dāng)RabbitMQ出現(xiàn)死信，可能會(huì)導(dǎo)致業(yè)務(wù)邏輯錯(cuò)誤，比如下訂單后修改庫存操作，在下單后因?yàn)槟撤N原因，發(fā)送的消息未被簽收，這時(shí)庫存數(shù)據(jù)會(huì)出現(xiàn)不一致。

有死信隊(duì)列之后我們就可以監(jiān)聽死信隊(duì)列，來處理業(yè)務(wù)邏輯。

3.1 死信隊(duì)列設(shè)置

聲明隊(duì)列，添加參數(shù)x-dead-letter-exchange

Map<String, Object> agruments = new HashMap<String, Object>();
agruments.put("x-dead-letter-exchange", "dlx.exchange");
//這個(gè)agruments屬性，要設(shè)置到聲明隊(duì)列上
channel.queueDeclare(queueName, true, false, false, agruments);

死信隊(duì)列，是一個(gè)普通的Exchange和queue，需要設(shè)置死信Exchange和queue，并進(jìn)行綁定

/要進(jìn)行死信隊(duì)列的聲明:
		channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
		channel.queueDeclare("dlx.queue", true, false, false, null);
		//可以匹配任意routeKey
		channel.queueBind("dlx.queue", "dlx.exchange", "#");

4 代碼實(shí)現(xiàn)

生產(chǎn)端

public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("127.0.0.1");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		Connection connection = connectionFactory.newConnection();
		Channel channel = connection.createChannel();
		String exchange = "test_dlx_exchange";
		String routingKey = "dlx.test";
		String msg = "RabbitMQ DLX Message test";
		for(int i =0; i<1; i ++){
			AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
					.deliveryMode(2)
					.contentEncoding("UTF-8")
					.expiration("10000") //過期時(shí)間為1秒
					.build();
			channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
		}
	}

消費(fèi)端

public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		Connection connection = connectionFactory.newConnection();
		Channel channel = connection.createChannel();
		// 這就是一個(gè)普通的交換機(jī) 和 隊(duì)列 以及路由
		String exchangeName = "test_dlx_exchange";
		String routingKey = "dlx.#";
		String queueName = "test_dlx_queue";
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		Map<String, Object> agruments = new HashMap<String, Object>();
		agruments.put("x-dead-letter-exchange", "dlx.exchange");
		//這個(gè)agruments屬性，要設(shè)置到聲明隊(duì)列上
		channel.queueDeclare(queueName, true, false, false, agruments);
		channel.queueBind(queueName, exchangeName, routingKey);
		//要進(jìn)行死信隊(duì)列的聲明:
		channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
		channel.queueDeclare("dlx.queue", true, false, false, null);
		//可以匹配任意routeKey
		channel.queueBind("dlx.queue", "dlx.exchange", "#");
	   channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("接收到消息：："+new String(body));
                }
            });
	}

運(yùn)行以上代碼，在交換機(jī)（Exchange）中會(huì)多出一個(gè)名為dlx.exchange 類型為topic的交換機(jī)，隊(duì)列中也有一個(gè)dlx.queue 隊(duì)列。

到此這篇關(guān)于RabbitMQ中的死信隊(duì)列(Dead Letter Exchanges)詳解的文章就介紹到這了,更多相關(guān)RabbitMQ死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: