RabbitMQ学习--死信队列

Posted 2022-12-08 三笠·阿卡曼

死信队列

概念

死信，指的是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致** queue 中的某些消息无法被消费**，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
应用场景：为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

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

死信实战

代码架构图

演示消息TTL过期导致的消息进入死信队列

生产者

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/**
 * @author vleus
 * @date 2021年07月24日 22:23
 *
 * 死信队列之生产者生产消息
 */
public class Producer 

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

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

        Channel channel = RabbitMqUtils.getChannel();

        //发送延迟消息（死信消息），设置TTL（time to live）
        AMQP.BasicProperties basicProperties =
                new AMQP.BasicProperties()
                        .builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) 
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes(StandardCharsets.UTF_8));
        

    

消费者C1（消费正常队列的消息）

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

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

/**
 * @author vleus
 * @date 2021年07月24日 22:02
 * <p>
 * 死信队列实战
 * 正常消费者1
 */
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 Exception 

        //获取信道
        Channel channel = RabbitMqUtils.getChannel();

        //声明死信和普通交换机，类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        //声明普通队列
        Map<String, Object> argumentMap = new HashMap<>();
        //设置正常消息的过期时间
        argumentMap.put("x-message-ttl", 10000);
        //正常队列设置死信交换机是谁？
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //设置死信routingKey
        argumentMap.put("x-dead-letter-routing-key", "lisi");
        //设置正常队列的队列长度的限制
//        argumentMap.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, argumentMap);

        //声明死信队列(消息是从普通队列转发到死信队列)
        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("正常消费者C1等待接收消息");

        DeliverCallback deliverCallback = (consumerTag, message) -> 
            System.out.println("Consumer01接收的消息为: " + new String(message.getBody(), "UTF-8"));
        ;

        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> 
        );

    

消费者C2

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

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

/**
 * @author vleus
 * @date 2021年07月24日 22:02
 * <p>
 * 死信队列实战
 * 正常消费者1
 */
public class Consumer02 

    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

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

        //获取信道
        Channel channel = RabbitMqUtils.getChannel();

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

        DeliverCallback deliverCallback = (consumerTag, message) -> 
            System.out.println("Consumer02接收的消息为: " + new String(message.getBody(), "UTF-8"));
        ;

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> 
        );

    

演示消息超过正常队列长度进入死信队列

生产者

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/**
 * @author vleus
 * @date 2021年07月24日 22:23
 *
 * 死信队列之生产者生产消息
 */
public class Producer 

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

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

        Channel channel = RabbitMqUtils.getChannel();

        //发送延迟消息（死信消息），设置TTL（time to live）
//        AMQP.BasicProperties basicProperties =
//                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(StandardCharsets.UTF_8));
        

    

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

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

/**
 * @author vleus
 * @date 2021年07月24日 22:02
 * <p>
 * 死信队列实战
 * 正常消费者1
 */
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 Exception 

        //获取信道
        Channel channel = RabbitMqUtils.getChannel();

        //声明死信和普通交换机，类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        //声明普通队列
        Map<String, Object> argumentMap = new HashMap<>();
        //设置正常消息的过期时间
//        argumentMap.put("x-message-ttl", 10000);
        //正常队列设置死信交换机是谁？
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //设置死信routingKey
        argumentMap.put("x-dead-letter-routing-key", "lisi");
        //设置正常队列的队列长度的限制
        argumentMap.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, argumentMap);

        //声明死信队列(消息是从普通队列转发到死信队列)
        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("正常消费者C1等待接收消息");

        DeliverCallback deliverCallback = (consumerTag, message) -> 
            System.out.println("Consumer01接收的消息为: " + new String(message.getBody(), "UTF-8"));
        ;

        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> 
        );

    

消费者C2

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

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

/**
 * @author vleus
 * @date 2021年07月24日 22:02
 * <p>
 * 死信队列实战
 * 正常消费者1
 */
public class Consumer02 

    //死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

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

        //获取信道
        Channel channel = RabbitMqUtils.getChannel();

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

        DeliverCallback deliverCallback = (consumerTag, message) -> 
            System.out.println("Consumer02接收的消息为: " + new String(message.getBody(), "UTF-8"));
        ;

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> 
        );

    

演示消息被拒绝进入死信队列

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/**
 * @author vleus
 * @date 2021年07月24日 22:23
 *
 * 死信队列之生产者生产消息
 */
public class Producer 

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

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

        Channel channel = RabbitMqUtils.getChannel();

        //发送延迟消息（死信消息），设置TTL（time to live）
//        AMQP.BasicProperties basicProperties =
//                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(StandardCharsets.UTF_8));
        

    

消费者C1

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

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

/**
 * @author vleus
 * @date 2021年07月24日 22:02
 * <p>
 * 死信队列实战
 * 正常消费者1
 */
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 Exception 

        //获取信道
        Channel channel = RabbitMqUtils.getChannel();

        //声明死信和普通交换机，类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        //声明普通队列
        Map<String, Object> argumentMap = new HashMap<>();
        //设置正常消息的过期时间
//        argumentMap.put("x-message-ttl", 10000);
        //正常队列设置死信交换机是谁？
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //设置死信routingKey
        argumentMap.put("x-dead-letter-routing-key", "lisi");
        //设置正常队列的队列长度的限制
//        argumentMap.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, argumentMap);

        //声明死信队列(消息是从普通队列转发到死信队列)
      
以上是关于RabbitMQ学习--死信队列的主要内容，如果未能解决你的问题，请参考以下文章
 
 07-rabbitMQ-死信队列
 RabbitMQ学习（中）——交换机死信队列和延迟队列
 RabbitMQ——死信队列的三大来源应用举例
 RabbitMQ——死信队列的三大来源应用举例
 RabbitMQ学习09--死信队列(TTL过期)
 rabbitmq死信队列及延迟队列