RabbitMQ--基础--10.2--死信队列

RabbitMQ–基础–10.2–死信队列

1、死信队列(DLX queue)

1. 当消息在一个队列中变成死信之后，它能重新被发送到另一个交换机中，这个交换机就是 死信交换机，绑定 死信交换机(DLX Exchange) 的队列就称之为死信队列
2. 死信队列同其他的队列一样都是普通的队列。
3. 在RabbitMQ中并没有特定的"死信队列"类型，而是通过配置，将其实现。
   1. 设置死信队列需要设置以下2个属性
      1. 交换机 x-dead-letter-exchange
      2. 路由键 x-dead-letter-routing-key

2、设置死信队列

1. 当我们在创建一个业务的交换机和队列的时候，可以配置参数，指明另一个队列为当前队列的死信队列。
2. 当消息"死信"后，会被自动路由到DLX Exchange的queue中
   

3、消息变成死信有这几种情况

1. 消费者对broker应答nack，并且消息禁止重回队列。
   1. 应答nack
      1. basic.reject or basic.nack
   2. 消息禁止重回队列
      1. requeue=false
2. 消费者对broker应答nack，允许重回队列，但是达到的retry重新入队的上限次数
3. 消息的TTL过期(Time To Live)：存活时间已经过期
4. Queue队列长度已达上限。
   1. 队列满，queue的"x-max-length"参数

4、死信队列的应用场景

1. 处理异常情况
   1. 重要的业务队列如果失败，就需要重新将消息用另一种业务逻辑处理。因为消息放到 另一个队列(A)里面，只要处理A队列的内容就行
2. 搭配 TTL 模拟延迟队列
3. 如果是正常的业务逻辑故意让消息中不合法的值失败，就不需要死信

5、代码

5.1、结构

5.2、代码架构图

5.3、死信队列的实现

1. TTL过期
2. 队列达到最大长度
3. 消息被拒

5.3.1、TTL过期

 package com.example.rabbitmq03.business.dead_queue.demo1;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 死信队列实战 生产者
 */
public class Producer {
    
    // 定义交换机名称
    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";

    // 设置RoutingKey
    public static final String NORMAL_RoutingKey = "zhangsan";
    public static final String DEAD_RoutingKey = "lisi";


    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();

        setConig(channel);


        // 开启发布确认
        channel.confirmSelect();
        
        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
        // 消息发布确认成功回调函数
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = map.headMap(deliveryTag);
                confirmed.clear();
            } else {
                map.remove(deliveryTag);
            }
            System.out.println("消息发布确认成功:" + deliveryTag);
        };
        
        // 消息发布确认失败回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("消息发布确认失败:" + deliveryTag);
        };
        
        // 准备确认监听器
        channel.addConfirmListener(ackCallback, nackCallback);
        
        String message = "Hello Dead Queue";
        
        // 死信消息 设置TTL时间(过期时间) 单位：ms，这里5秒就过期
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("5000").build();
        
        // 发送消息
        channel.basicPublish(NORMAL_EXCHANGE, NORMAL_RoutingKey, properties, message.getBytes());
        map.put(channel.getNextPublishSeqNo(), message);
        
        System.out.println("发出消息:" + message);
        
    }

    //配置交换机，队列，路由键信息
    public static void setConig(Channel channel) throws Exception {
        // 声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", DEAD_RoutingKey);
        
        // 声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        
        // 绑定交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_RoutingKey);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_RoutingKey);
    }
}

 package com.example.rabbitmq03.business.dead_queue.demo1;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.*;

import java.util.HashMap;
import java.util.Map;

/**
 * 死信队列实战: TTL过期 消费者01
 */
public class Consumer {
    // 定义队列名称
    public static final String NORMAL_QUEUE = "normal_queue";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        System.out.println("Consumer等待接收消息。。。");
        
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer接收到的消息:" + new String(message.getBody(), "UTF-8"));
            // 手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费失败");
        };
        // 消费消息
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
    }
}

package com.example.rabbitmq03.business.dead_queue.demo1;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

/**
 * 死信队列实战: TTL过期
 * 死信消费者
 */
public class DeadConsumer {

    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception{
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();

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

        DeliverCallback deliverCallback = ( consumerTag,  message) ->{
            System.out.println("DeadConsumer接收到的消息:"+new String(message.getBody(),"UTF-8"));
            //手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费失败");
        };
        //消费消息
        channel.basicConsume(DEAD_QUEUE,false,deliverCallback,cancelCallback);
    }
}

1. 先启动Producer
2. 再启动DeadConsumer
   

5.3.2、队列达到最大长度

package com.example.rabbitmq03.business.dead_queue.demo2;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;

/**
 * 死信队列实战 生产者
 */
public class Producer {
    
    // 定义交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange2";
    public static final String DEAD_EXCHANGE = "dead_exchange2";
    // 定义队列名称
    public static final String NORMAL_QUEUE = "normal_queue2";
    public static final String DEAD_QUEUE = "dead_queue2";
    
    // 设置RoutingKey
    public static final String NORMAL_RoutingKey = "zhangsan2";
    public static final String DEAD_RoutingKey = "lisi2";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        setConig(channel);
        
        // 开启发布确认
        channel.confirmSelect();
        
        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
        // 消息发布确认成功回调函数
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = map.headMap(deliveryTag);
                confirmed.clear();
            } else {
                map.remove(deliveryTag);
            }
            System.out.println("消息发布确认成功:" + deliveryTag);
        };
        
        // 消息发布确认失败回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("消息发布确认失败:" + deliveryTag);
        };
        
        // 准备确认监听器
        channel.addConfirmListener(ackCallback, nackCallback);
        
        for (int i = 0; i < 11; i++) {
            String message = "Hello Dead Queue" + i;
            // 发送消息
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_RoutingKey, null, message.getBytes());
            map.put(channel.getNextPublishSeqNo(), message);
            System.out.println("发出消息:" + message);
        }
        
    }
    
    // 配置交换机，队列，路由键信息
    public static void setConig(Channel channel) throws Exception {
        // 声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", DEAD_RoutingKey);
        // 设置正常队列的长度
        arguments.put("x-max-length", 6);
        // 声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        
        // 绑定交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_RoutingKey);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_RoutingKey);
        
    }
}

package com.example.rabbitmq03.business.dead_queue.demo2;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

/**
 * 死信队列实战: TTL过期
 * 死信消费者
 */
public class DeadConsumer {

    public static final String DEAD_QUEUE = "dead_queue2";

    public static void main(String[] args) throws Exception{
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();

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

        DeliverCallback deliverCallback = ( consumerTag,  message) ->{
            System.out.println("DeadConsumer接收到的消息:"+new String(message.getBody(),"UTF-8"));
            //手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消息消费失败");
        };
        //消费消息
        channel.basicConsume(DEAD_QUEUE,false,deliverCallback,cancelCallback);
    }
}

package com.example.rabbitmq03.business.dead_queue.demo2;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

/**
 * 死信队列实战: TTL过期 消费者01
 */
public class Consumer {
    // 定义队列名称
    public static final String NORMAL_QUEUE = "normal_queue2";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        System.out.println("Consumer等待接收消息。。。");
        
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer接收到的消息:" + new String(message.getBody(), "UTF-8"));
            // 手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费失败");
        };
        // 消费消息
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
    }
}

5.3.3、消息被拒

package com.example.rabbitmq03.business.dead_queue.demo3;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

/**
 * 死信队列实战: TTL过期 消费者01
 */
public class Consumer {
    // 定义队列名称
    public static final String NORMAL_QUEUE = "normal_queue3";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        System.out.println("Consumer等待接收消息。。。");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), "UTF-8");
            if (msg.equals("Hello Dead Queue1")) {
                // 拒绝
                System.out.println("Consumer接收到的消息:" + msg + ",此消息被拒绝");
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            } else {
                System.out.println("Consumer接收到的消息:" + msg);
                // 手动应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费失败");
        };
        // 消费消息
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
    }
}

package com.example.rabbitmq03.business.dead_queue.demo3;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;

/**
 * 死信队列实战: TTL过期 死信消费者
 */
public class DeadConsumer {
    
    public static final String DEAD_QUEUE = "dead_queue3";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        System.out.println("DeadConsumer等待接收消息。。。");
        
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("DeadConsumer接收到的消息:" + new String(message.getBody(), "UTF-8"));
            // 手动应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消息消费失败");
        };
        // 消费消息
        channel.basicConsume(DEAD_QUEUE, false, deliverCallback, cancelCallback);
    }
}

package com.example.rabbitmq03.business.dead_queue.demo3;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import com.example.rabbitmq03.business.confirm.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;

/**
 * 死信队列实战 生产者
 */
public class Producer {
    
    // 定义交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange3";
    public static final String DEAD_EXCHANGE = "dead_exchange3";
    // 定义队列名称
    public static final String NORMAL_QUEUE = "normal_queue3";
    public static final String DEAD_QUEUE = "dead_queue3";
    
    // 设置RoutingKey
    public static final String NORMAL_RoutingKey = "zhangsan3";
    public static final String DEAD_RoutingKey = "lisi3";
    
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        Connection connection = RabbitMqUtil.getConnection("生产者");
        // 2. 通过连接获取通道 Channel
        Channel channel = connection.createChannel();
        
        setConig(channel);
        
        // 开启发布确认
        channel.confirmSelect();
        
        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
        // 消息发布确认成功回调函数
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = map.headMap(deliveryTag);
                confirmed.clear();
            } else {
                map.remove(deliveryTag);
            }
            System.out.println("消息发布确认成功:" + deliveryTag);
        };
        
        // 消息发布确认失败回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("消息发布确认失败:" + deliveryTag);
        };
        
        // 准备确认监听器
        channel.addConfirmListener(ackCallback, nackCallback);
        
        for (int i = 0; i < 5; i++) {
            String message = "Hello Dead Queue" + i;
            // 发送消息
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_RoutingKey, null, message.getBytes());
            map.put(channel.getNextPublishSeqNo(), message);
            System.out.println("发出消息:" + message);
        }
        
    }
    
    // 配置交换机，队列，路由键信息
    public static void setConig(Channel channel) throws Exception {
        // 声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", DEAD_RoutingKey);
        // // 设置正常队列的长度
        // arguments.put("x-max-length", 6);
        // 声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        
        // 绑定交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_RoutingKey);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_RoutingKey);
        
        // 声明普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 声明死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        
    }
}