死信的概念

死信：顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息
消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时
间未支付时自动失效。

死信的来源

• 消息 TTL 过期
• 队列达到最大长度(队列满了，无法再添加数据到 mq 中)
• 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

死信实战

消息TTL过期 TTl：time to live

生产者代码

public static void setMessageOverdueProducer() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();

        // 死信消息 设置TTL时间  TTl：time to live
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").build();

        // 循环次数
        long length = 11L;

        for (int i = 1; i < length; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "normal", properties, message.getBytes(StandardCharsets.UTF_8));
        }
    }

普通消费者代码

/**
 * @author zjh
 *
 * 普通消费者
 */
public class ConsumerOne {

    /**
     * 普通交换机
     */
    private static final String NORMAL_EXCHANGE = "normal_exchange";

    /**
     * 死信交换机
     */
    private static final String DEAD_EXCHANGE = "dead_exchange";

    /**
     * 普通队列名称
     */
    private static final String NORMAL_QUEUE = "normal_queue";

    /**
     * 死信队列名称
     */
    private static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();

        // 声明死信和普通交换机 类型为DIRECT
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明普通队列 类型为DIRECT
        Map<String, Object> arguments = new HashMap<>(8);
        // 正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信routingKey
        arguments.put("x-dead-letter-routing-key", "dead");

        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

        // 声明死信 类型为DIRECT
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 绑定普通交换机与普通队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "normal");
        // 绑定死信交换机与死信队列
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "dead");
        System.out.println("等待接收消息...");

        // 接收消息回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("ConsumerOne接收消息是：" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback,
                consumerTag -> {});
    }
}

死信消费者代

/**
 * @author zjh
 *
 * 死信队列 消费者
 */
public class ConsumerTwo {

    /**
     * 死信队列名称
     */
    private static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();

        System.out.println("等待接收消息...");


        channel.basicConsume(DEAD_QUEUE, true, (consumerTag, message) -> System.out.println("ConsumerTwo接收消息是：" + new String(message.getBody(), StandardCharsets.UTF_8)),
                consumerTag -> {});
    }
}

测试方法步骤

1. 首先先启动 ConsumerOne 进行声明交换机和队列

2. 声明好后将 ConsumerOne 停止，因为测试消费TTL过期设置了10s，所以将消费者停止 消息无人消费就会进入死信队列

3. 启动 Producer 发送消息，可以看到正常队列有十条消息等待消费

4. 等待十秒钟可以看到十条消息都进入死信队列了

5. 在启动 ConsumerTwo 消费死信队列

标签：代码,语言,系列,编程,严谨,数据,结构,数据结构,抽象 来源：

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。