RabbitMQ死信交换机

时间：2024-01-26 15:09:09

目录

1.死信交换机介绍

 2.TTL

3.延迟队列

4.消息堆积问题

5.惰性队列

 6.代码实战

1.死信交换机介绍

当一个队列中信息满足下列情况之一时，可以成为死信（dead letter）

（1）消费者使用basic.reject（RejectAndDontRequeueRecoverer处理策略）或bastic.nack（ImmediateRequeueMessageRecoverer处理策略）消费失败，requeue（重新入队）设置参数为false；

（2）消息是一个过期消息，过期无人消费；

（3）要投递的队列消息堆积满了，最早的消息成为死信；

如果该队列配置了dead-letter-exchange属性，指定交换机，队列设置dead-letter-routing-key属性，设置死信交换机与死信队列的RoutingKey，那么队列中的死信就会投递到这个交换机中，而这个交换机成为死信交换机（Dead Letter Exchange，简称DLX）。

 

 2.TTL

        TTL，也就是Time-To-Live。如果队列中的消息TTL结束仍未消费则会成为死信，TTL超时分为两种情况：

（1）消息所在的队列设置了存活时间；

（2）消息本身设置了存活时间；

 

3.延迟队列

利用ttl结合死信交换机，实现了消息发出后，消费者延迟收到消息的结果，这种消息队列成为延迟队列（Delay Queue）模式。

延迟队列使用的场景：

（1）用户下单后超过15分钟未支付取消支付；

（2）预约20分钟会议，到时间自动通知参会人员；

订单的超时处理：

生产者生产一条1分钟后超时的订单消息到正常交换机exchange中，但一分钟后仍未消费，消息会被投递到死信交换机dlx-exchange中，并发送到私信队列中，死信队列dlx-queue的消费者拿到消息后，根据消息去查询订单的状态，如果仍然是为支付状态，将订单状态更新为超时状态 

4.消息堆积问题

当生产者发送的消息的速度大于消费者消费的速度，就会导致队列中的消息堆积，直到达到消息的上限，最早接受的消息，可能会被丢弃，成为死信。

解决消费堆积的三种策略：

（1）增加消费者，增加消费速度

（2）在消费者内开启线程池加快消息处理速度

（3）扩大队列的容积

5.惰性队列

从RabbitMq的3.6.0版本开始增加了Lazy Queues的概念，也就是惰性队列。

惰性队列的特征如下：

（1）接受消息后直接存入磁盘而非内存；

（2）消费者消费消息才会中磁盘中将消息加载到内存；

（3）支持数百万的数据存储；

 6.代码实战

先写两个队列和分别自定义一个直连交换机 并且给交换机分别绑定一个队列

    //死信，延迟队列
    @Bean
    public Queue queueA(){
        Map<String, Object> config = new HashMap<>();
        //message在该队列queue的存活时间最大为20秒
        config.put("x-message-ttl", 20000);
        //x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
        config.put("x-dead-letter-exchange", "ExchangeB"); 
        //x-dead-letter-routing-key参数是给这个DLX指定路由键
        config.put("x-dead-letter-routing-key", "bb");
        return new Queue("queueA",true,false,false,config);
    }
    @Bean
    public DirectExchange ExchangeA(){
        return new DirectExchange("ExchangeA");
    }
    @Bean
    public Binding bindingA(){
        return BindingBuilder.bind(queueA()).to(ExchangeA()).with("aa");
    }

    @Bean
    public Queue queueB(){
        return new Queue("queueB");
    }
    @Bean
    public DirectExchange ExchangeB(){
        return new DirectExchange("ExchangeB");
    }
    @Bean
    public Binding bindingB(){
        return BindingBuilder.bind(queueB()).to(ExchangeB()).with("bb");
    }

 在控制类给交换机发送一条消息

    //死信交换机
    @RequestMapping("/send5")
    public String send5(){
        template.convertAndSend("ExchangeA","aa","4502520");
        return "????";
    }

写一个测试类进行测试

package com.example.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues="queueB")
public class ReceiverQB {
    @RabbitHandler
    public void process(String id){
        log.warn("QB接收到:" + id);
    }
}

 接着启动生产者和消费者，并在RabbitMQ中查看

 

注意：如果在启动消费者时报错，请先在页面*问，再重新启动消费者 

因为我们设置队列queue的存活时间最大为20秒，所以再隔20秒后消息会自动出来

 

 

确认消息（局部方法处理消息）

默认情况下消息消费者是自动 ack （确认）消息的，如果要手动 ack（确认）则需要修改确认模式为 manual

在消费者的yml文件中进行配置

server:
    port: 9999
spring:
    rabbitmq:
        host: 192.168.169.132
        password: 123456
        port: 5672
        username: spring
        virtual-host: my_vhost
        listener:
            simple:
                acknowledge-mode: manual
接着写一个测试类
package com.example.consumer;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues="queueA")
public class ReceiverQA {
    @RabbitHandler
    public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
        log.warn("QA接收到:" + id);
        channel.basicAck(tag,true);

    }
}

在页面中刷新上面的方法，可以在idea中看到

如果想要手动否认，拒绝消息

@RabbitHandler
    public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
        log.error("QA接收到:" + id);
        channel.basicReject(tag,true);
        Thread.sleep(1000);//延迟时间
    }

因为延迟时间只有1秒钟，而且需要重新进入队列，所以会一直循环，在循环到在我们上面设置的那个20秒钟后会进入QB

也可以拒绝该消息，消息会被丢弃，不会重回队列  

    @RabbitHandler
    public void process(String id, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
        log.error("QA接收到:" + id);
        channel.basicReject(tag,false);
        Thread.sleep(1000);//延迟时间
    }

这时候会直接进入QB 

 

相关文章