1、消息被否定接收，消費(fèi)者使用basic.reject 或者 basic.nack并且requeue重回隊(duì)列屬性設(shè)為false

2、消息在隊(duì)列里的時(shí)間超過了該消息設(shè)置的過期時(shí)間(TTL)

3、消息隊(duì)列到達(dá)了它的最大長度，之后再收到的消息。

2、死信隊(duì)列的原理

當(dāng)一個(gè)消息在隊(duì)列里變?yōu)樗佬牛鼤恢匦聀ublish到綁定的死信隊(duì)列所對應(yīng)的exchange交換機(jī)上，這個(gè)exchange就為DLX。因此我們只需要在聲明正常的業(yè)務(wù)隊(duì)列時(shí)添加一個(gè)可選的"x-dead-letter-exchange"參數(shù)，值為死信交換機(jī)，死信就會被rabbitmq重新publish到配置的這個(gè)交換機(jī)上，我們接著監(jiān)聽這個(gè)死信交換機(jī)所綁定隊(duì)列就可以了。

下面看個(gè)案例： 正常情況，我們配置交換機(jī)-A，會為其綁定路由鍵-A,和路由鍵-A所映射的隊(duì)列-A,并設(shè)置這個(gè)隊(duì)列-A的死信交換機(jī)-B、死信路由鍵-B。

消息消費(fèi)者-A，監(jiān)聽隊(duì)列-A,基本上就沒死信交換機(jī)什么事了，特殊時(shí)候隊(duì)列滿了，或者接收后給的回執(zhí)是false才會進(jìn)到死信交換機(jī)-B。

現(xiàn)在我們把消費(fèi)者-A刪了，并且設(shè)置隊(duì)列-A里面消息過期時(shí)間，那么所有進(jìn)到隊(duì)列A里面消息就自然過期，然后被推給死信交換機(jī)，那么監(jiān)聽死信交換機(jī)-B的消息者-B,就可以得到過期后的消息了，就實(shí)現(xiàn)了延時(shí)功能。

3、死信交換機(jī)

死信交換機(jī)DLX實(shí)際上也是普通的交換機(jī)，說白的就是將死信(過期消息)路由到死信隊(duì)列的交換機(jī)。

4、死信隊(duì)列

死信隊(duì)列DLQ(Dead Letter Queue)實(shí)際上也是普通的隊(duì)列，只不過它存儲的是死信交換機(jī)路由過來的死信(過期消息)

三、代碼

1、給隊(duì)列綁定死信隊(duì)列

Map<String, Object> args = new HashMap<>(2);
args.put(x-dead-letter-exchange,死信隊(duì)列交換機(jī))
args.put(x-dead-letter-routing-key,死信routeKey)
args.put(x-message-ttl,隊(duì)列里面的消息最大存活時(shí)間) 
//申明一個(gè)普通隊(duì)列，并且給這個(gè)普通隊(duì)里綁定一個(gè)死信隊(duì)列。當(dāng)消息過期就會轉(zhuǎn)到死信隊(duì)列里面去。
Queue queue = QueueBuilder.durable(queue).withArguments(args).build();

2、設(shè)置過期時(shí)間

目前有兩種方式可以設(shè)置消息的TTL。

第一種是通過隊(duì)列的屬性設(shè)置，隊(duì)列中的所有消息都有相同的過期時(shí)間。

第二種方法是對消息本身進(jìn)行單獨(dú)設(shè)置，每條消息的TTL可以不同。

如果兩種方法同時(shí)設(shè)置，則TTL以兩者之間較小的那個(gè)數(shù)值為準(zhǔn)。

消息在隊(duì)列中的生存時(shí)間一旦超過設(shè)置的TTL值時(shí)就會進(jìn)到死信。

1、通過隊(duì)列屬性設(shè)置消息過期時(shí)間

Map<String, Object> arguments = Maps.newHashMap();
//設(shè)置消息發(fā)送到隊(duì)列中在被丟棄之前可以存活的時(shí)間，單位：毫秒
arguments.put("x-message-ttl", 5000);
//聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);

2、發(fā)送消息的時(shí)候，對消息本身設(shè)定過期時(shí)間

MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setExpiration("5000");//設(shè)置消息多久沒有被消費(fèi)在過期。
        message.getMessageProperties().setContentEncoding("UTF-8");
        return message;
    }
};
rabbitTemplate.convertAndSend(exchangeName, routingKey, msg, messagePostProcessor);

上面底層代碼是：

AMQP.BasicProperties.Builder properties = MessageProperties.PERSISTENT_TEXT_PLAIN.builder();
properties.expiration("5000");//設(shè)置消息多久沒有被消費(fèi)在過期。
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true, properties.build(), message.getBytes());

到此這篇關(guān)于關(guān)于Rabbitmq死信隊(duì)列及延時(shí)隊(duì)列的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Rabbitmq死信隊(duì)列延時(shí)隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！