RabbitMQ死信队列

1. 死信队列

死信队列是一种特殊的消息队列,用来存储无法被正常消费的消息，常被用来实现延迟处理，异常消息处理等，提高了系统的可伸缩性和容错性，能够应对高并发和异常消息。
死信队列中的消息被称为死信消息,用来分发死信消息的交换机被称为死信交换机(Dead Letter Exchange,DLX)。
死信队列在实际项目中的应用场景有很多如:

• 订单超时未支付，将此消息放入死信队列中，等待后续处理(延迟等待)
• 消息消费失败将消息放入死信队列中进行重试(消息重试机制)
  

2.正常消息成为死信消息的条件

• 消息到了过期时间仍然未被消费者消费
• 队列已满无法保存新消息
• 消息被拒绝消费且未设置重新放入队列
  

3.消费者1

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/** * @Description: 死信队列消费者1 * @Author: hong * @Date: 2024-01-17 21:04 * @Version: 1.0 **/
public class Consumer1 {
    
    //正常交换机名称
    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 Exception {
    
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //正常队列绑定死信队列信息
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        map.put("x-dead-letter-routing-key", "lisi");
        //声明正常队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        System.out.println("Consumer1等待接收消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
    
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
    
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);

    }
}

4.生产者

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

/** * @Description: 死信队列消息生产者 * @Author: hong * @Date: 2024-01-17 20:49 * @Version: 1.0 **/
public class Producer {
    
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
    
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);

        //设置消息的 TTL 时间 10s
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").build();

        //该信息是用作演示队列个数限制
        for (int i = 1; i < 11; i++) {
    
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }
    }

}

启动消费者1后马上关闭消费者1,模拟消费者1接收不到消息,再启动生产者


生产者发送10条正常队列中有10条消息

10s后正常队列中的消息由于没有消费者消费进入死信队列中

5.消费者2

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/** * @Description: 死信队列-死信消费者 * @Author: hong * @Date: 2024-01-17 21:31 * @Version: 1.0 **/
public class Consumer2 {
    
    public static final String DEAD_EXCHANGE = "dead_exchange";
    public static final String DEAD_QUEUE = "dead_queue";
    public static void main(String[] args) throws Exception {
    
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");


        System.out.println("Consumer2等待接收死信消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
    
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
    
        };

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
    }
}

启动消费者2,使其消费死信队列中的消息

6.队列达到最大长度

以上代码是TTL,队列达到最大长度只要将上述代码稍微改动一下即可

6.1.注释掉生产者代码中的ttl部分

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

/** * @Description: 死信队列消息生产者 * @Author: hong * @Date: 2024-01-17 20:49 * @Version: 1.0 **/
public class Producer {
    
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
    
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);

        //设置消息的 TTL 时间 10s
// AMQP.BasicProperties properties = new AMQP.BasicProperties()
// .builder().expiration("10000").build();

        //该信息是用作演示队列个数限制
        for (int i = 1; i < 11; i++) {
    
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());
            System.out.println("生产者发送消息:" + message);
        }
    }
}

6.2.消费者1代码中加最大长度

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/** * @Description: 死信队列消费者1 * @Author: hong * @Date: 2024-01-17 21:04 * @Version: 1.0 **/
public class Consumer1 {
    
    //正常交换机名称
    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 Exception {
    
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //正常队列绑定死信队列信息
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        map.put("x-dead-letter-routing-key", "lisi");
        map.put("x-max-length",8);
        //声明正常队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        System.out.println("Consumer1等待接收消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
    
            System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };

        CancelCallback cancelCallback = var -> {
    
        };

        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);

    }
}

启动消费者1后立马关闭，模拟队列已满

7.消息被拒绝

7.1消费者1

package com.hong.rabbitmq9;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/** * @Description: 死信队列消费者1 * @Author: hong * @Date: 2024-01-17 21:04 * @Version: 1.0 **/
public class Consumer1 {
    
    //正常交换机名称
    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 Exception {
    
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");

        //正常队列绑定死信队列信息
        Map<String,Object> map = new HashMap<>();
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        map.put("x-dead-letter-routing-key", "lisi");
// map.put("x-max-length",8);
        //声明正常队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");

        System.out.println("Consumer1等待接收消息:");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
    
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            if("info3".equals(msg) || "info6".equals(msg) || "info9".equals(msg)){
    
                System.out.println("被Consumer1拒绝的消息:" + msg);
                //拒绝消息并且没有重新入队
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            }
            System.out.println( "Consumer1接收的消息:" + msg);
        };

        CancelCallback cancelCallback = var -> {
    
        };
        //手动应答
        channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);
    }
}

启动消费者1后不关闭再启动生产者



info3,info6,info9进入死信队列
启动消费者2将其消费