RabbitMq高级特性之TTL 存活时间/过期时间 通俗易懂 超详细 内含案例

Posted 2020-12-18 beixuan

RabbitMq高级特性之TTL 存活时间/过期时间

每条消息设置过期时间

整个 Queue 队列设置过期时间

前提

1. 完成 RabbitMq高级特性之消费端限流

一、每条消息设置过期时间

1.更改ProducerTest.java文件

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class producerTest {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Test
    public void test(){
        
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //设置消息 3秒后过期
                message.getMessageProperties().setExpiration("3000");
                return message;
            }
        };
    
        String routingKey = "item.insert";
    
        int count = 1;
        while (count <= 9){
            String message = "发送第"+count+"条消息";
            //log.debug("路由键:{}",routingKey);
            if (count % 3 == 0){
                //给其中的第3 6 9条消息添加过期时间
                rabbitTemplate.convertAndSend(RabbitConfig.DIRECT_EXCHANGE,"",message,messagePostProcessor);
            } else {
                rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE_NAME,routingKey,message);
            }
            count++;
        }
        log.debug("发送成功");
    }
}

二、整个 Queue 队列设置过期时间

1.更改RabbitMqConfig.java文件

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMq 配置类
 */
@Configuration
public class RabbitMqConfig {
    private static final String TOPIC_EXCHANGE_NAME = "topic_exchange";
    private static final String TOPIC_QUEUE_NAME = "topic_queue";

    /**
     * 创建 交换机
     * @return
     */
    @Bean
    public Exchange itemTopicExchange(){
        return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE_NAME).build();
    }

    /**
     * 创建 队列
     * @return
     */
    @Bean
    public Queue itemQueue(){
        //QueueBuilder.durable(TOPIC_QUEUE_NAME).withArgument("x-message-ttl",3000).build();
        //与下句代码 效果一致 写一个就可以
        return QueueBuilder.durable(TOPIC_QUEUE_NAME).ttl(3000).build();
    }

    /**
     * 绑定 交换机与队列
     * @param exchange
     * @param queue
     * @return
     */
    @Bean
    public Binding itemQueueExchange(@Qualifier("itemTopicExchange") Exchange exchange, @Qualifier("itemQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();
    }

}

三、测试

直接运行ProducerTest.java单元测试即可,为了达到自动过期,消费端不可启动

四、小结

两种模式可同时起作用,哪个时间短,就会执行哪个

Queue级别的过期时间一到,会自动丢掉消息,不会等到消费者来取消息.

Message级别的消息过期后,不会自动抹去,而是等到消费者获取消息时进行判断是否过期,如过期则丢掉.