《RabbitMQ系列教程-第九章-RabbitMQ之死信队列》
程序员文章站
2022-03-20 22:54:04
RabbitMQ之死信队列9.1.1 简介死信:Dead Letter,缩写DL;当一条正常的消息变成为"死信"时,会被RabbitMQ进行特殊处理,如果配置了死信队列信息,那么该消息将会被丢进死信队列中,如果没有配置,则该消息将会被丢弃。消息什么情况下会变为死信?1)当使用 channel.basicNack 或 channel.basicReject 拒绝签收消息,并且此时requeue 属性被设置为false时;2)当消息设置了TTL,并且已经到达时间时;3)消息队列中的消息数量已经到达...
教程说明
- 本系列教程目录大纲:《RabbitMQ系列教程-目录大纲》
- 本系列教程配套代码:https://gitee.com/Horizon1024/rabbitmt.git(码云地址)
RabbitMQ之死信队列
9.1.1 简介
死信:Dead Letter
,缩写DL;当一条正常的消息变成为"死信"时,会被RabbitMQ进行特殊处理,如果配置了死信队列信息,那么该消息将会被丢进死信队列中,如果没有配置,则该消息将会被丢弃。
消息什么情况下会变为死信?
- 1)当使用
channel.basicNack
或channel.basicReject
拒绝签收消息,并且此时requeue
属性被设置为false
时; - 2)当消息设置了TTL,并且已经到达时间时;
- 3)消息队列中的消息数量已经到达上限时,其余的消息将成为死信;
当消息成为死信时,我们将其转发到一个专门处理死信的交换机上,这个交换机也称为死信交换机DLX
(Dead Letter Exchange
),死信被死信交换机重新转发到死信队列中,供消费者进行消费;
9.1.2 死信队列工作流程图:
死信交换机也是一个普通的交换机,只不过用于接收死信而已;
9.1.3 案例测试
根据上述图我们应该配置业务交换机、死信交换机、业务队列、死信队列;
9.1.3.1 配置spring.xml:
<!--
死信队列:
1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
3. 正常队列绑定死信交换机
设置两个参数:
* x-dead-letter-exchange:死信交换机名称
* x-dead-letter-routing-key:发送给死信交换机的routingkey
-->
<!-- 1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx) -->
<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
<!--3. 正常队列绑定死信交换机-->
<rabbit:queue-arguments>
<!--3.1 x-dead-letter-exchange:死信交换机名称-->
<entry key="x-dead-letter-exchange" value="exchange_dlx" />
<!--3.2 x-dead-letter-routing-key:发送给死信交换机的routingKey-->
<entry key="x-dead-letter-routing-key" value="dlx.abc" />
<!--4.1 设置队列的过期时间 ttl-->
<entry key="x-message-ttl" value="3000" value-type="java.lang.Integer" />
<!--4.2 设置队列的长度限制 max-length -->
<entry key="x-max-length" value="10" value-type="java.lang.Integer" />
</rabbit:queue-arguments>
</rabbit:queue>
<!-- 正常的交换机 -->
<rabbit:direct-exchange name="test_exchange_dlx">
<rabbit:bindings>
<rabbit:binding queue="test_queue_dlx" key="test_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- 2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx) -->
<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<!-- 死信交换机 -->
<rabbit:topic-exchange name="exchange_dlx">
<rabbit:bindings>
<rabbit:binding queue="queue_dlx" pattern="dlx.*"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>
9.1.3.2 测试代码:
生产者:
分别测试过期时间到期、长度超限、消息拒收,查看消息是否变化为死信;
/**
* 发送测试死信消息:
* 1. 过期时间
* 2. 长度限制
* 3. 消息拒收
*/
@Test
public void testDlx() {
//1. 测试过期时间,死信消息
// rabbitTemplate.convertAndSend("test_exchange_dlx","test_dlx","dlx.....");
//2. 测试长度限制后,消息死信
/* for (int i = 0; i < 20; i++) {
rabbitTemplate.convertAndSend("test_exchange_dlx","test_dlx","dlx....."+i);
}*/
//3. 测试消息拒收
rabbitTemplate.convertAndSend("test_exchange_dlx", "test_dlx", "dlx...");
}
消费者拒收:
package com.lscl.rabbitmq.listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
@Component
public class DlxListener implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//1.接收转换消息
System.out.println(new String(message.getBody()));
//2. 处理业务逻辑
System.out.println("处理业务逻辑...");
int i = 3/0;//出现错误
//3. 手动签收
channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();
System.out.println("出现异常,拒绝接受");
//4.拒绝签收,不重回队列 requeue=false
channel.basicNack(deliveryTag,true,false);
}
}
}
注册监听:
<rabbit:listener-container connection-factory="connectionFactory" >
<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"></rabbit:listener>
</rabbit:listener-container>
9.1.4 死信队列小结
存储死信的队列叫做死信队列,一条正常的消息有三种情况会变为死信;当消息变为死信时,可以根据一定的绑定规则配置到死信交换机上,将消息路由到死信队列中重新利用;
正常消息变为死信的条件:
- 1)消息被拒绝签收,并且没有重回队列
- 2)消息到达了过期时间
- 3)队列到达最大限制之后的消息都将变为死信
本文地址:https://blog.csdn.net/Bb15070047748/article/details/112191772