RabbitMQ消息單獨(dú)與批量的TTL詳細(xì)介紹
TTL(Time To Live)存活時(shí)間。表示當(dāng)消息由生產(chǎn)端存入MQ當(dāng)中的存活時(shí)間,當(dāng)時(shí)間到達(dá)的時(shí)候還未被消息就會(huì)被自動(dòng)清除。RabbitMQ可以對(duì)消息單獨(dú)設(shè)置過(guò)期時(shí)間也可以對(duì)整個(gè)隊(duì)列(并不是隊(duì)列,而是隊(duì)列中的消息)設(shè)置過(guò)期時(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
#三個(gè)類型:none默認(rèn)不開(kāi)啟確認(rèn)回調(diào) correlated開(kāi)啟確認(rèn)回調(diào)
#simple也會(huì)確認(rèn)回調(diào) 還會(huì)調(diào)用waitForConfirms()方法或waitForConfirmsOrDie()方法
publisher-confirm-type: correlated # 開(kāi)啟確認(rèn)回調(diào)
publisher-returns: true # 開(kāi)啟退回回調(diào)
業(yè)務(wù)邏輯
隊(duì)列消息過(guò)期
第一段代碼即是定義交換機(jī)與隊(duì)列的名稱并使其進(jìn)行綁定,僅是一個(gè)配置類的效果。第二段代碼就是生產(chǎn)者產(chǎn)生消息的方法,只需要在意for循環(huán)里面的邏輯即可。而圖1即是創(chuàng)建出來(lái)的隊(duì)列以及生產(chǎn)的10條消息,在10s會(huì)自動(dòng)刪除。因?yàn)樵谂渲妙愔幸呀?jīng)定義了TTL。
/** * 定義交換機(jī)與隊(duì)列的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(); } //配置隊(duì)列的時(shí)候順帶上ttl()方法 其內(nèi)部對(duì)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("交換機(jī)成功接受到了消息"); 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("隊(duì)列接受不到交換機(jī)的消息進(jìn)行了失敗回調(diào)"); } }); // 以上代碼只是保證消息傳遞的可靠性 與TTL無(wú)關(guān) for(int i = 0; i < 10; ++i){ rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld"); } } }
圖1
消息單獨(dú)過(guò)期
這里的配置類還是如上相同:隊(duì)列中的消息10s自動(dòng)過(guò)期,再對(duì)其中一條消息進(jìn)行處理就可以更好的明白這兩種過(guò)期的區(qū)別:其中一條消息設(shè)置了5s自動(dòng)過(guò)期,通過(guò)圖2可以發(fā)現(xiàn)隊(duì)列中有11條消息,當(dāng)5s后變?yōu)?0條消息,再過(guò)了5s后就沒(méi)有消息。
@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("交換機(jī)成功接受到了消息"); 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("隊(duì)列接受不到交換機(jī)的消息進(jìn)行了失敗回調(diào)"); } }); // 以上代碼只是保證消息傳遞的可靠性 與TTL無(wú)關(guān) // 消息的后處理對(duì)象 設(shè)置一些消息的參數(shù)信息 MessagePostProcessor messagePostProcessor = new MessagePostProcessor(){ @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000");//設(shè)置消息對(duì)象5s后過(guò)期 return message; } }; //消息單獨(dú)5s過(guò)期 rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld",messagePostProcessor); //隊(duì)列中的消息全體10s過(guò)期 for(int i = 0; i < 10; ++i){ rabbitTemplate.convertAndSend(RabbitMQConfig.TTL_EXCHANGE_NAME,"test.heHe","HelloWorld"); } } }
圖2
到這真的就以為結(jié)束了嗎,當(dāng)我把for循環(huán)的10條消息放到單獨(dú)過(guò)期的消息上面,發(fā)現(xiàn)了新大陸:一開(kāi)始的消息也是如圖2所示為11條,但是但是,過(guò)了5s后并不會(huì)消除一條消息,而是過(guò)了10s后將11條全部刪除了。于是我猜測(cè)這個(gè)消息的存放隊(duì)列就好似一個(gè)棧,雖然先生產(chǎn)的消息的生存時(shí)間短,但是當(dāng)別的消息壓在自己頭上的時(shí)候是出不去的,而是必須等自己為棧頂元素才可以出棧!
到此這篇關(guān)于RabbitMQ消息單獨(dú)與批量的TTL詳細(xì)介紹的文章就介紹到這了,更多相關(guān)RabbitMQ消息的TTL內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
SpringBoot引入Redis報(bào)Redis?command?timed?out兩種異常情況
這篇文章主要給大家介紹了關(guān)于SpringBoot引入Redis報(bào)Redis?command?timed?out兩種異常情況的相關(guān)資料,文中通過(guò)示例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-08-08Maven之導(dǎo)入thymeleaf依賴飄紅問(wèn)題及解決
這篇文章主要介紹了Maven之導(dǎo)入thymeleaf依賴飄紅問(wèn)題及解決方案,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-08-08redis實(shí)現(xiàn)多進(jìn)程數(shù)據(jù)同步工具代碼分享
這篇文章主要介紹了使用redis實(shí)現(xiàn)多進(jìn)程數(shù)據(jù)同步工具的代碼,大家參考使用吧2014-01-01java實(shí)現(xiàn)Runnable接口適合資源的共享
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)Runnable接口適合資源的共享,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-07-07詳解如何使用tldb數(shù)據(jù)庫(kù)的java客戶端
這篇文章主要為大家介紹了如何使用tldb數(shù)據(jù)庫(kù)的java客戶端過(guò)程示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-09-09Awaitility同步異步工具實(shí)戰(zhàn)示例詳解
這篇文章主要為大家介紹了Awaitility同步異步工具實(shí)戰(zhàn)示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-08-08SpringBoot整合Redis使用RedisTemplate和StringRedisTemplate
Spring?Boot?Data(數(shù)據(jù))?Redis?中提供了RedisTemplate和StringRedisTemplate,其中StringRedisTemplate是RedisTemplate的子類,兩個(gè)方法基本一致。本文介紹了SpringBoot整合Redis使用RedisTemplate和StringRedisTemplate的方法,需要的可以參考一下2022-12-12