6. 死信队列

6.1. 死信的概念

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景:为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

6.2. 死信的来源

消息 TTL 过期消息存活时间过期

队列达到最大长度(队列满了，无法再添加数据到 mq 中)

消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false. 拒绝应答不放在队列中也会在死信中

6.3. 死信实战

6.3.1. 代码架构图

6.3.2. 消息 TTL 过期

生产者代码

public class Producer {
private static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitMqUtils.getChannel()) { channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//设置消息的 TTL 时间 newAMQP.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);
}
}
}
}

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 死信队列  生产者
 */
public class Producer {
    //普通交换机 名称
    public static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
       //死信消息 设置TTl  单位是ms 100000ms=10s
        AMQP.BasicProperties basicProperties =
                new AMQP.BasicProperties()
                        .builder().expiration("1000").build();
        for (int i = 0; i < 11; i++) {
            String message="info"+i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
        }
    }
}

消费者 C1 代码(启动之后关闭该消费者模拟其接收不到消息)

public class Consumer01 {
//普通交换机名称
private static final String NORMAL_EXCHANGE = "normal_exchange";
//死信交换机名称
private static final String DEAD_EXCHANGE = "dead_exchange"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel();
//声明死信和普通交换机 类型为 direct channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
//声明死信队列
String deadQueue = "dead-queue"; 
channel.queueDeclare(deadQueue, false, false, false, null);
//死信队列绑定死信交换机与 routingkey channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
//正常队列绑定死信队列信息
Map<String, Object> params = new HashMap<>();
//正常队列设置死信交换机 参数 key 是固定值 params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
//正常队列设置死信 routing-key 参数 key 是固定值 params.put("x-dead-letter-routing-key", "lisi");
String normalQueue = "normal-queue";
channel.queueDeclare(normalQueue, false, false, false, params); 
channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> 
{ String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Consumer01 接收到消息"+message);
};
channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> { });
}
}

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.HashMap;
import java.util.Map;
/**
 * 死信队列
 * 消费者 1
 * 大小写切换 ctrl+shirft+u
 */
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  1.交换机的名字  2.交换机的类型
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明普通交换机 类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        /**
         *          * 生成一个队列
         *          * 1.队列名称
         *          * 2.队列中的消息是否持久化  默认消息存储在内存中 否不持久化 （持久化放在磁盘中）
         *          *3.该队列是否只供一个消费者进行消费 是否共享  true 可以多个消费者消费  默认false,只能一个消费者消费
         *          4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true自动删除 false不删除
         *          5.其他参数
         *
         */
        //声明死信队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
        //声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        //过期时间   10s=100000ms
        // arguments.put("x-message-ttl",100000);
        //正常队列设置死信交换机        1.值  2.死信交换机名字
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        //声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
        //绑定普通的交换机与偶有普通的队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        //绑定死信的交换机与偶有死信的队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        System.out.println("等待接收消息.......");
        DeliverCallback deliverCallback = (var1, var2) -> {
            System.out.println("Consumer01" + new String(var2.getBody(), "utf-8"));
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, var1 -> {
        });
    }
}

消费者 C2 代码(以上步骤完成后启动 C2 消费者它消费死信队列里面的消息)

public class Consumer02 {
private static final String DEAD_EXCHANGE = "dead_exchange"; 
public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); 
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); 
String deadQueue = "dead-queue";
channel.queueDeclare(deadQueue, false, false, false, null); 
channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
System.out.println("等待接收死信队列消息 ........... "); 
DeliverCallback deliverCallback = (consumerTag, delivery) ->
{ String message = new String(delivery.getBody(), "UTF-8"); 
System.out.println("Consumer02 接收死信队列的消息" + message);
};
channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> { });
}
}

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.HashMap;
import java.util.Map;
/**
 * 死信队列
 * 消费者 1
 * 大小写切换 ctrl+shirft+u
 */
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("等待接收消息.......");
        DeliverCallback deliverCallback = (var1, var2) -> {
            System.out.println("Consumer02" + new String(var2.getBody(), "utf-8"));
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, var1 -> {
        });
    }
}

6.3.3. 队列达到最大长度

1. 消息生产者代码去掉 TTL 属性

public class Producer {
private static final String NORMAL_EXCHANGE = "normal_exchange"; 
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitMqUtils.getChannel())
{ 
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//该信息是用作演示队列个数限制
for (int i = 1; i <11 ; i++) {
 String message="info"+i;
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
 System.out.println("生产者发送消息:"+message);
}
}
}
}

2. C1 消费者修改以下代码(启动之后关闭该消费者模拟其接收不到消息

注意此时需要把原先队列删除因为参数改变了

C2 消费者代码不变(启动 C2 消费者)

6.3.4. 消息被拒

消息生产者代码同上生产者一致
C1 消费者代码(启动之后关闭该消费者模拟其接收不到消息)

public class Consumer01 {
//普通交换机名称
private static final String NORMAL_EXCHANGE = "normal_exchange";
//死信交换机名称
private static final String DEAD_EXCHANGE = "dead_exchange"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel();
//声明死信和普通交换机 类型为 direct channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
//声明死信队列
String deadQueue = "dead-queue"; channel.queueDeclare(deadQueue, false, false, false, null);
//死信队列绑定死信交换机与 routingkey channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
//正常队列绑定死信队列信息
Map<String, Object> params = new HashMap<>();
//正常队列设置死信交换机 参数 key 是固定值 params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
//正常队列设置死信 routing-key 参数 key 是固定值 params.put("x-dead-letter-routing-key", "lisi"); String normalQueue = "normal-queue";
channel.queueDeclare(normalQueue, false, false, false, params); channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
System.out.println("等待接收消息 ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); if(message.equals("info5")){
System.out.println("Consumer01 接收到消息"   +                      message  +                                                                                       "并拒绝签收该消息");
//requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
}else {
System.out.println("Consumer01 接收到消息"+message); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> { });
}
}

C2 消费者代码不变

启动消费者 1 然后再启动消费者 2

上面3个部分内容代码整理如下：

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.HashMap;
import java.util.Map;
/**
 * 死信队列
 * 消费者 1
 * 大小写切换 ctrl+shirft+u
 */
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  1.交换机的名字  2.交换机的类型
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明普通交换机 类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        /**
         *          * 生成一个队列
         *          * 1.队列名称
         *          * 2.队列中的消息是否持久化  默认消息存储在内存中 否不持久化 （持久化放在磁盘中）
         *          *3.该队列是否只供一个消费者进行消费 是否共享  true 可以多个消费者消费  默认false,只能一个消费者消费
         *          4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true自动删除 false不删除
         *          5.其他参数
         *
         */
        //声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        //过期时间   10s=100000ms
       //  arguments.put("x-message-ttl",100000);
        //正常队列设置死信交换机        1.值  2.死信交换 机名字
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "lisi");
        //设置正常队列长度的限制
        //arguments.put("x-max-length",6);
        //声明普通队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false,arguments);
        //声明死信队列
        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("等待接收消息.......");
        DeliverCallback deliverCallback = (var1, var2) -> {
           String msg= new String(var2.getBody(), "utf-8");
           if (msg.equals("info5")){
               System.out.println("Consumer01接受的消息是"+msg+"消息是被拒绝的");
               // 1.获取到值 2.不放回普通队列
               channel.basicReject(var2.getEnvelope().getDeliveryTag(),false );
           }else {
               System.out.println("Consumer01接受的消息是" +msg);
               channel.basicReject(var2.getEnvelope().getDeliveryTag(),false );
           }
        };
     //   channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, var1 -> {});
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, var1 -> {});
    }
}

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.HashMap;
import java.util.Map;
/**
 * 死信队列
 * 消费者 1
 * 大小写切换 ctrl+shirft+u
 */
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("等待接收消息.......");
        DeliverCallback deliverCallback = (var1, var2) -> {
            System.out.println("Consumer02" + new String(var2.getBody(), "utf-8"));
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, var1 -> {
        });
    }
}

生产者

package com.atguigu.rabbitmq.eight;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 死信队列  生产者
 */
public class Producer {
    //普通交换机 名称
    public static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
       //死信消息 设置TTl  单位是ms 100000ms=10s  过期时间
//        AMQP.BasicProperties basicProperties =
//                new AMQP.BasicProperties()
//                        .builder().expiration("1000").build();
        for (int i = 1; i < 11; i++) {
            String message="info"+i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
          //  channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
            System.out.println("生产者发送消息"+message);
        }
    }
}

RibbitMQ学习笔记之死信队列

6. 死信队列

6.1. 死信的概念

6.2. 死信的来源

6.3. 死信实战

6.3.1. 代码架构图

6.3.2. 消息 TTL 过期

6.3.3. 队列达到最大长度

1. 消息生产者代码去掉 TTL 属性

2. C1 消费者修改以下代码(启动之后关闭该消费者模拟其接收不到消息

6.3.4. 消息被拒

热门文章

最新文章

相关电子书

RibbitMQ学习笔记之死信队列

6. 死信队列

6.1. 死信的概念

6.2. 死信的来源

6.3. 死信实战

6.3.1. 代码架构图

6.3.2. 消息 TTL 过期

6.3.3. 队列达到最大长度

1. 消息生产者代码去掉 TTL 属性

2. C1 消费者修改以下代码(启动之后关闭该消费者 模拟其接收不到消息

6.3.4. 消息被拒

热门文章

最新文章

相关电子书

2. C1 消费者修改以下代码(启动之后关闭该消费者模拟其接收不到消息