RabbitMQ延迟队列优化

最新推荐文章于 2024-03-23 16:51:18 发布

zhang__1234

最新推荐文章于 2024-03-23 16:51:18 发布

阅读量171

点赞数

文章标签： java-rabbitmq rabbitmq java

本文链接：https://blog.csdn.net/zhang__1234/article/details/128485719

版权

本文介绍如何在RabbitMQ中通过配置TTL队列实现消息延迟处理，包括不同队列的TTL设置、死信队列绑定及消费者监听，并探讨其优缺点。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

配置类

@Configuration
public class TtlQueueConfig {
    //    普通交换机名称
    public static final String X_EXCHANGE = "X";
    //    死信交换机名称
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    //    普通队列名称
    public static final String Queue_A = "QA";
    //    死信队列名称
    public static final String Queue_B = "QB";
    //    死信队列名称
    public static final String DEAD_LETTER_QUEUE = "QD";
//    普通队列的名称
    public static final String QUEUE_C="QC";

    //    声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    //   声明yExchange别名
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //     声明普通队列 TTL为10s钟
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> arguments = new HashMap<>();
//      设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
//      设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "YD");
//      设置TTL
        arguments.put("x-message-ttl", 10000);

        return QueueBuilder.durable(Queue_A).withArguments(arguments).build();
    }

    //    声明普通队列TTL 40S
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>();
//      设置死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
//      设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key", "YD");
//      设置TTL
        arguments.put("x-message-ttl", 40000);

        return QueueBuilder.durable(Queue_B).withArguments(arguments).build();
    }
//    死信队列
    @Bean("queueD")
    public Queue queueD(){
        return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
    }

//    声明QC
    @Bean("queueC")
    public Queue queueC(){
        HashMap<String, Object> arguments = new HashMap<>();
//        设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//        设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","YD");
        return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
    }
    @Bean
    public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }
//    绑定
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }
    //    绑定
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }
    @Bean
    public Binding queueBBindingY(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange){
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

死信队列消费者

@Slf4j
@Component
public class DeadLetterQueueConsumer {
//    接收消息
    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel){
        String msg=new String(message.getBody());
        log.info("当前时间:{},收到死信队列的消息:{}",new Date().toString(),msg);
    }
}

生产者

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        log.info("当前时间:{},发送一条消息给两个ttl队列:{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列" + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列" + message);
    }
    @GetMapping("/sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
        log.info("当前时间:{},发送一条时长:{}毫秒消息给QCttl队列:{}", new Date().toString(), ttlTime, message);
        rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
//    发送消息的时候 延迟时长
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
    }
}

application.properties文件

spring.rabbitmq.host=192.168.66.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

优点：
可以动态设置消息的延迟时间，不用每次新增一个需求就增加一个队列
缺点：
由于队列遵循先进先出的原则，如果延迟40s的消息先进队列，紧跟着又进去一个10s的，但是消息是否过期是在即将投递消息到消费者之前判定的，也就是RabbitMQ只会判断队列头部的消息是否过期，其他消息是否过期它是不知道的，所以由于先进去的是40s的，后面10s的才进去，就会造成，消费者消费的时候10s的也是跟着40s的算的，出现消息堆积的情况，虽然10s的已经过期，但是由于40s的先进去，10s的不会被消费掉，之后等40s的消费掉之后，10s的才会出来