RabbitMq之死信队列_死信队列的作用

死信队列

1. 死信的概念

死信，顾名思义就是死掉的信息（无法被消费的信息），一般来说，就是生产者将消息发送给broker（exchange+queue），但由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有被处理，就形成了死信，有死信自然就有死信队列，来存放死信。

2. 死信的来源

1.消息TTL时间过期
2.队列达到最大长度（无法再接收消息）
3.消息被拒绝 (比如 ： basic.reject 或 basic.nack)并且 requeue=false

3. 应用场景

为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息
消费发生异常时，将消息投入死信队列中。
比如说：
1.用户下单后未在指定时间类支付，此订单就会失效
2.一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，当发生异常时，当然不能每次通过日志来获取原消息，然后让运维帮忙重新投递消息。这时候我们就可以通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息。

5. 死信来源的三种代码演示

（这里三种来源都整合到一起了，看注释，演示效果是第3种消息被拒绝的情况）
生产者

package com.ra.sx;

import com.ra.lx.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

/**
 * Map<String, Object> headers = new HashMap<String, Object>();
 *             headers.put("arg1", "aaa");
 *  AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
 *                     .deliveryMode(2) // 传送方式
 *                     .contentEncoding("UTF-8") // 编码方式
 *                     .expiration("10000") // 过期时间
 *                     .headers(headers) //自定义属性
 *                     .build();
 *
 *
 *
 *                     String contentType, //消息内容的类型
 * String contentEncoding, //消息内容的编码格式
 * Map<String,Object> headers,//header类型的交换机可以用到
 * Integer deliveryMode,//消息持久化 1 不持久化 2 持久化
 * Integer priority,//优先级
 * String correlationId, //关联id
 * String replyTo,//通常用于命名回调队列
 * String expiration,//设置过期消息过期时间
 * String messageId, //消息id
 * Date timestamp, //消息的时间戳
 * String type,  //类型
 * String userId, //用户ID
 * String appId, //应用程序id
 * String clusterId //集群id
 */
public class Producer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();


        //1.设置死信过期时间  AMQP.BasicProperties 提供了一个构造器，可以通过builder() 来设置一些属性；
        //AMQP.BasicProperties properties =
        //new AMQP.BasicProperties().builder().expiration("10000").build();

        for (int i = 0; i <10 ; i++) {
            String message = "info ->"+i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
        }

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

消费者

package com.ra.sx;

import com.ra.lx.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;


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

public class Comsumer01 {

    private static final String NORMAL_EXCHANGE = "normal_exchange";

    private static final String DEAD_EXCHANGE = "dead_exchange";

    private static final String NOMAL_QUEUE = "normal_queue";

    private static final String DEAD_QUEUE = "dead_queue";

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

        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        Map<String,Object>arguments = new HashMap();
        //正常队列设置死信交换机 参数 key 是固定值
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常队列设置死信 routing-key 参数 key 是固定值
        arguments.put("x-dead-letter-routing-key", "lisi");

        //设计正常队列里面存放的消息长度 超过6 存入死信
        // arguments.put("x-max-length",6);
        /**
         * 死信队列的信息  是 由正常队列 转发的 ， 所以 正常队列要设置参数  告诉 要转给 死信队列
         */
        channel.queueDeclare(NOMAL_QUEUE,false,false,false,arguments);

        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);

        System.out.println("等待接收消息.....");
        channel.queueBind(NOMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        DeliverCallback deliverCallback = ( consumerTag,  message)->{
            String msg = new String(message.getBody(), "UTF-8");
            if("info ->5".equals(msg)){
                System.out.println("Consumer01接收的消息是："+msg+"此消息是被C1拒绝的");
                //message.getEnvelope().getDeliveryTag() 获取消息的标签 , false是不放回队列
                // channel.basicReject 拒绝此条消息
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else {
                System.out.println("Consumer01接收的消息是："+msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            }

            System.out.println("接收的消息 :" + new String(message.getBody(),"UTF-8"));
        };

        //开启手动应答
        channel.basicConsume(NOMAL_QUEUE,false,deliverCallback,consumerTag -> {});
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

消费者2

package com.ra.sx;

import com.ra.lx.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

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

public class Comsumer02 {


    private static final String DEAD_QUEUE = "dead_queue";

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



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


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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

5. 总结

1.业务消息被投入交换机再通过路由发送到业务队列
2.消费者消费业务队列的消息，由于处理过程中发生异常，将消息投递的死信交换机中
3.死信交换机将消息投入相应的死信队列
4.死信队列的消费者消费死信消息