【尚硅谷 RabbitMQ】5、图文详解死信队列

1、概述

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

为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中用户在商城下单成功并点击去支付后，在指定时间未支付时自动失效 2、产生的原因

消息TTL过期队列达到最大长度（队列满了，无法再添加数据到mq中）消息被拒绝（basicReject 或 basicNack）并且 requeue = false 3、实战 1）代码架构图

2）消息TTL过期

消费者1

package com.tuwer.rabbitmq.deadletter;

import com.rabbitmq.client.*;
import com.tuwer.utils.RabbitMqUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


public class Consumer01 {
    
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";
    
    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException {
        // 工具类
        RabbitMqUtils mqUtils = new RabbitMqUtils();

        // 获取通道
        Channel channel = mqUtils.getChannel(
                "192.168.19.101",
                5672,
                "admin",
                "admin",
                "/",
                "消费者01");

        // 声明交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明普通队列：指定需要转发到死信队列的参数
        Map arguments = new HashMap<>();
        // 过期时间(单位毫秒)：过期时间建议在生产者发送时设置，更灵活
        //arguments.put("x-message-ttl", 10000);
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信队列RoutingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 普通队列绑定普通交换机
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");

        // 死信队列绑定死信交换机
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

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

        // 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("接收的信息：【" + new String(message.getBody()) + "】");
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("未接收的信息：" + consumerTag);
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, cancelCallback);
    }
}

生产者

package com.tuwer.rabbitmq.deadletter;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.tuwer.utils.RabbitMqUtils;

import java.io.IOException;
import java.util.concurrent.TimeUnit;


public class Producer {
    
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) {
        // 工具类
        RabbitMqUtils mqUtils = new RabbitMqUtils();

        // 获取通道
        Channel channel = mqUtils.getChannel(
                "192.168.19.101",
                5672,
                "admin",
                "admin",
                "/",
                "生产者");

        // 发送消息
        int n = 10;
        String message = "";
        try {
            // 设置消息ttl
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties()
                    .builder().expiration("10000").build();

            for (int i = 0; i < n; i++) {
                message = "message-" + (i + 1);
                channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", basicProperties, message.getBytes());
                System.out.println("【"+message+"】已发送！");
                // 休眠1秒
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            // 关闭
            mqUtils.close();
        }

    }
}

消费者2

package com.tuwer.rabbitmq.deadletter;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.tuwer.utils.RabbitMqUtils;

import java.io.IOException;


public class Consumer02 {
    
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException {
        // 工具类
        RabbitMqUtils mqUtils = new RabbitMqUtils();

        // 获取通道
        Channel channel = mqUtils.getChannel(
                "192.168.19.101",
                5672,
                "admin",
                "admin",
                "/",
                "消费者02");

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

        // 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("接收的信息：【" + new String(message.getBody()) + "】");
        };
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("未接收的信息：" + consumerTag);
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
    }
}

测试

3）队列达到最大长度

消费者

// 只需要修改这一处
// 设置普通队列的长度
arguments.put("x-max-length",6);

生产者

测试

4）消息被拒绝

消息者1

// ...

// 取消队列长度
// 设置普通队列的长度
//arguments.put("x-max-length",6);

// ...

// 接收消息
DeliverCallback deliverCallback = (consumerTag, message) -> {

    // 消息编号
    long tag = message.getEnvelope().getDeliveryTag();
    if(tag == 5){
        // 拒收消息，并且不重新入列
        channel.basicReject(tag, false);
        System.out.println("信息：【" + new String(message.getBody()) + "】被C1拒收了！");
    }else{
        // 确认接收，不批量确认
        channel.basicAck(tag, false);
        System.out.println("接收的信息：【" + new String(message.getBody()) + "】");
    }
};
CancelCallback cancelCallback = consumerTag -> {
    System.out.println("未接收的信息：" + consumerTag);
};
// 接收消息：不自动确认
channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
// ...

生产者、消费者2同上

测试

先删除之前的交换机、队列

【尚硅谷 RabbitMQ】5、图文详解 死信队列

Java相关栏目本月热门文章

【尚硅谷 RabbitMQ】5、图文详解死信队列