- 死信队列DLX:DeadLetter Exchange(死信交换机),当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就是DLX。
- 消息成为死信的三种情况:
- 队列消息数量到达限制;比如给队列最大只能存储10条消息,当第11条消息进来的时候存不下了,第11条消息就被称为死信
- 消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false;
- 原队列存在消息过期设置,消息到达超时时间未被消费;
- 死信的处理方式:
- 丢弃,如果不是很重要,可以选择丢弃
- 记录死信入库,然后做后续的业务分析或处理
- 通过死信队列,由负责监听死信的应用程序进行处理
- 队列绑定死信交换机:
- 给队列设置参数: x-dead-letter-exchange 和 x-dead-letter-routing-key
- 过期时间代码demo:
- spring-rabbitmq-producer.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.hehe"></entry> <!--4.1 设置队列的过期时间 ttl--> <entry key="x-message-ttl" value="10000" 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:topic-exchange name="test_exchange_dlx"> <rabbit:bindings> <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding> </rabbit:bindings> </rabbit:topic-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 pattern="dlx.#" queue="queue_dlx"></rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange>
- 发送消息
/** * 发送测试死信消息: * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 */ @Test public void testDlx(){ //1. 测试过期时间,死信消息 rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); }
- spring-rabbitmq-producer.xml
- 长度限制代码实现
- 发送消息
/** * 发送测试死信消息: * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 */ @Test public void testDlx(){ //1. 测试过期时间,死信消息 //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); //2. 测试长度限制后,消息死信 for (int i = 0; i < 20; i++) { rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); } }
- 发送消息
- 测试消息拒收
- spring-rabbitmq-consumer.xml
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"> <!--<rabbit:listener ref="ackListener" queue-names="test_queue_confirm"></rabbit:listener>--> <!--<rabbit:listener ref="qosListener" queue-names="test_queue_confirm"></rabbit:listener>--> <!--定义监听器,监听正常队列--> <rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"></rabbit:listener> </rabbit:listener-container>
- 消费端拒收:
@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); } } }
- 发送消息:
/** * 发送测试死信消息 * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 */ @Test public void testDlx(){ //1. 测试过期时间,死信消息 //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); //2. 测试长度限制后,消息死信 // for (int i = 0; i < 20; i++) { // rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); // } //3. 测试消息拒收 rabbitTemplate.convertAndSend("test_exchange_dlx", "test.dlx.haha", "我是一条消息,我会死吗?"); }
- spring-rabbitmq-consumer.xml
- 死信队列小结:
- 死信交换机和死信队列和普通的没有区别
- 当消息成为死信后,如果该队列绑定了死信交换机,则消息会被死信交换机重新路由到死信队列
- 消息成为死信的三种情况:
- 队列消息长度(数量)到达限制;
- 消费者拒接消费消息,并且不重回队列;
- 原队列存在消息过期设置,消息到达超时时间未被消费;