[大数据]RabbitMQ学习（六）--死信队列

死信队列

概念

死信，指的是无法被消费的消息，字面意思可以这样理解，一般来说，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);

        //声明死信队列(消息是从普通队列转发到死信队列)
        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) -> {
            String msg = new String(message.getBody(), "UTF-8");
            if (msg.equals("info5")) {
                System.out.println("Consumer01接收的消息为: " + msg + ",此消息被拒绝");
                //消费者拒绝消息，false表示放到死信队列里
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else{
                System.out.println("Consumer01接收的消息为: " + msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }
        };

        //开启手动应答才有拒绝消息的机会
        channel.basicConsume(NORMAL_QUEUE, false, 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 -> {
        });

    }
}