《RabbitMQ系列教程-第九章-RabbitMQ之死信队列》-CFANZ编程社区

教程说明

本系列教程目录大纲：《RabbitMQ系列教程-目录大纲》

本系列教程配套代码：https://gitee.com/lvshuichangliu/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 死信队列工作流程图：

《RabbitMQ系列教程-第九章-RabbitMQ之死信队列》_spring

死信交换机也是一个普通的交换机，只不过用于接收死信而已；

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>