RabbitMQ----死信队列和延迟队列

Posted 2022-12-06 未来.....

RabbitMQ----死信队列和延迟队列

  先想一想，如果你有一个快递，一直尝试给你派送，但是一直联系不上你，签收不了，那么此时这个快递该如何处理？是再次尝试派送，还是处理掉？

  现在引入一个名词，死信，顾名思义：就是不能被消费的信息，字面意思可以这样理解，生产者(发送方)将信息传递到交换机，交换机将信息发送到队列中，消费者从队列中取出信息并消费掉，但是由于各种原因，导致队列中的某些信息不能被消费，这样的消息如果没有后续的处理，就变成了死信，有了死信信息，就有了死信的队列，也就有了死信交换机(Dead Letter Exchange)。

1、RabbitMQ----死信队列

1、1 消息成为死信的三种情况：

did

1. 队列消息长度到达限制；

2. 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

3. 原队列存在消息过期设置，消息到达超时时间未被消费；

1、2 如何给队列绑定死信交换机呢？

1. 给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key
   

1、3 代码块：

创建队列和交换机：

@Configuration
public class RabbitConfig 
    private final String EXCHANGE = "exchange";//普通交换机的名称
    private final String DEAD_EXCHANGE = "exchange";//死信交换机的名称
    private final String QUEUE = "queue";// 普通队列的名称
    private final String DEAD_QUEUE = "dead_queue";// 死信队列的名称


    /**
     * 普通队列
     */
    @Bean
    public Queue queue()
        return QueueBuilder
                .durable(QUEUE)
                .withArgument("x-message-ttl",5000)//过期时间
                .withArgument("x-max-length",10)//队列中最大条数
                .withArgument("x-dead-letter-exchange",DEAD_EXCHANGE)//绑定死信交换机
                .withArgument("x-dead-letter-routing-key","error")//死信交换机的key
                .build();
    

    /**
     * 死信队列
     */
    @Bean
    public Queue dead_queue()
        return QueueBuilder.durable(DEAD_EXCHANGE).build();
    

    /**
     * 普通交换机
     */
    @Bean
    public Exchange exchange()
        return ExchangeBuilder.directExchange(EXCHANGE).build();
    

    /**
     * 死信交换机
     */
    @Bean
    public Exchange dead_exchange()
        return ExchangeBuilder.directExchange(EXCHANGE).build();
    

    /**
     * 绑定普通
     */
    @Bean
    public Binding binding()
        return BindingBuilder.bind(queue()).to(exchange()).with("info").noargs();
    

    /**
     * 绑定死信
     */
    @Bean
    public Binding dead_binding()
        return BindingBuilder.bind(dead_queue()).to(dead_exchange()).with("error").noargs();
    

创建application.yml文件

spring:
  rabbitmq:
    host: 192.168.31.33
    port: 5672

创建测试类单元：

@SpringBootTest
public class Test01 

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testDeadQueue() 
        rabbitTemplate.convertAndSend("exchange", "info", "这条消息会因为超时而挂掉。");
    

运行后发现信息两个队列中都有信息，等待20秒，看看是否会到

此时默认过了20秒，此时发现普通队列中是没有信息了。

RabbitMQ----延迟队列

1.为什么需要使用延迟队列？适用于什么场景？

  订单下单之后30分钟后，如果用户没有付钱，则系统自动取消订单。
  这样类似的需求是我们经常会遇见的问题。最常用的方法是定期轮训数据库，设置状态。在数据量小的时候并没有什么大的问题，但是数据量一大轮训数据库的方式就会变得特别耗资源。当面对千万级、上亿级数据量时，本身写入的IO就比较高，导致长时间查询或者根本就查不出来。通过使用延迟队列来解决这种问题。

延迟队列就是给队列设置了一个过期时间，延迟队列一般都是配合TTL和死信队列来实现的。

上代码实现：
在原有项目上创建消费者子项目
创建消费者：

@Component
public class MyListener 
    @RabbitListener(queues = "dead_queue")
    public void listener(Message message, Channel channel)
        System.out.println("三十分钟没有处理订单，取消订单");
        try 
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
         catch (IOException e) 
            try 
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);
             catch (IOException ioException) 
                
            
        
    

主启动类：

@SpringBootApplication
public class ConsumerApp 
    public static void main(String[] args) 

        SpringApplication.run(com.ConsumerApp.class,args);
    

发送信息到交换机,创建交换机在上面，点这里跳转

@Test
    public void testDeadQueue() 
        rabbitTemplate.convertAndSend("exchange", "info", "下单成功。");
    

先开启消费者，启动消费者主启动类，在启动测试单元。
此时死信队列中是没有信息的。
假设等待5秒钟，看看信息是否被消费者消费掉了。

得到结果，信息被消费者得到了。