欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

RabbitMQ消息單獨與批量的TTL詳細介紹

 更新時間：2023年05月19日 11:45:29   作者：小白的救贖  
這篇文章主要介紹了RabbitMQ消息單獨與批量的TTL，TTL全名是Time To Live存活時間，表示當(dāng)消息由生產(chǎn)端存入MQ當(dāng)中的存活時間，當(dāng)時間到達的時候還未被消息就會被自動清除，感興趣的同學(xué)可以參考下文

TTL（Time To Live）存活時間。表示當(dāng)消息由生產(chǎn)端存入MQ當(dāng)中的存活時間，當(dāng)時間到達的時候還未被消息就會被自動清除。RabbitMQ可以對消息單獨設(shè)置過期時間也可以對整個隊列（并不是隊列，而是隊列中的消息）設(shè)置過期時間。

生產(chǎn)者端

目錄結(jié)構(gòu)

導(dǎo)入依賴

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

修改yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    #三個類型：none默認不開啟確認回調(diào) correlated開啟確認回調(diào)
    #simple也會確認回調(diào) 還會調(diào)用waitForConfirms()方法或waitForConfirmsOrDie()方法
    publisher-confirm-type: correlated # 開啟確認回調(diào)
    publisher-returns: true # 開啟退回回調(diào)

業(yè)務(wù)邏輯

隊列消息過期

第一段代碼即是定義交換機與隊列的名稱并使其進行綁定，僅是一個配置類的效果。第二段代碼就是生產(chǎn)者產(chǎn)生消息的方法，只需要在意for循環(huán)里面的邏輯即可。而圖1即是創(chuàng)建出來的隊列以及生產(chǎn)的10條消息，在10s會自動刪除。因為在配置類中已經(jīng)定義了TTL。

/**
 * 定義交換機與隊列的Bean 并且使之綁定
 */
@Component
public class RabbitMQConfig {
    public static final String TTL_EXCHANGE_NAME = "ttl_exchange_name";
    public static final String TTL_QUEUE_NAME = "ttl_queue_name";
    @Bean("ttlExchange")
    public Exchange ttlExchange(){
        return ExchangeBuilder.topicExchange(TTL_EXCHANGE_NAME).durable(true).build();
    }
    //配置隊列的時候順帶上ttl()方法 其內(nèi)部對MQ設(shè)置了參數(shù)"x-message-ttl"
    //注意這里的單位是毫秒 所以我寫的參數(shù)為10000毫秒即10秒
    @Bean("ttlQueue")
    public Queue ttlQueue(){
        return QueueBuilder.durable(TTL_QUEUE_NAME).ttl(10000).build();
    }
    @Bean
    public Binding ttl(@Qualifier("ttlExchange") Exchange exchange,
                       @Qualifier("ttlQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("test.#").noargs();
    }
}
@SpringBootTest
@RunWith(SpringRunner.class)
class RabbitmqProducerApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testTTL(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b) System.out.println("交換機成功接受到了消息");
                else System.out.println("消息失敗原因" + s);
            }
        });
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("隊列接受不到交換機的消息進行了失敗回調(diào)");
            }
        });
        // 以上代碼只是保證消息傳遞的可靠性 與TTL無關(guān)
        for(int i = 0; i < 10; ++i){
            rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld");
        }
    }
}

圖1

消息單獨過期

這里的配置類還是如上相同：隊列中的消息10s自動過期，再對其中一條消息進行處理就可以更好的明白這兩種過期的區(qū)別：其中一條消息設(shè)置了5s自動過期，通過圖2可以發(fā)現(xiàn)隊列中有11條消息，當(dāng)5s后變?yōu)?0條消息，再過了5s后就沒有消息。

@SpringBootTest
@RunWith(SpringRunner.class)
class RabbitmqProducerApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testTTL(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b) System.out.println("交換機成功接受到了消息");
                else System.out.println("消息失敗原因" + s);
            }
        });
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("隊列接受不到交換機的消息進行了失敗回調(diào)");
            }
        });
        // 以上代碼只是保證消息傳遞的可靠性 與TTL無關(guān)
        // 消息的后處理對象 設(shè)置一些消息的參數(shù)信息
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor(){
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");//設(shè)置消息對象5s后過期
                return message;
            }
        };
        //消息單獨5s過期
        rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld",messagePostProcessor);
        //隊列中的消息全體10s過期
        for(int i = 0; i < 10; ++i){
            rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld");
        }
    }
}

圖2

到這真的就以為結(jié)束了嗎，當(dāng)我把for循環(huán)的10條消息放到單獨過期的消息上面，發(fā)現(xiàn)了新大陸：一開始的消息也是如圖2所示為11條，但是但是，過了5s后并不會消除一條消息，而是過了10s后將11條全部刪除了。于是我猜測這個消息的存放隊列就好似一個棧，雖然先生產(chǎn)的消息的生存時間短，但是當(dāng)別的消息壓在自己頭上的時候是出不去的，而是必須等自己為棧頂元素才可以出棧！

到此這篇關(guān)于RabbitMQ消息單獨與批量的TTL詳細介紹的文章就介紹到這了,更多相關(guān)RabbitMQ消息的TTL內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

最新評論