RabbitMq高階特性之TTL 存活時間/過期時間通俗易懂超詳細【內含案例】

阿新 • • 發佈：2020-07-25

RabbitMq高階特性之TTL 存活時間/過期時間

前提

完成 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();
    }

}