詳解RabbitMQ中延遲隊(duì)列結(jié)合業(yè)務(wù)場景的使用
消息進(jìn)入隊(duì)列后不會(huì)立即被消費(fèi),只有到達(dá)指定時(shí)間后才會(huì)被消費(fèi)。業(yè)務(wù)場景就是支付時(shí)間內(nèi)未支付就清除訂單或者用戶注冊一段時(shí)間后發(fā)短信問候。在這里想說的是這只是一種思想,并不是真正的一種用法,這種思想所需要的用法就是用上消息TTL存活時(shí)間以及死信隊(duì)列來實(shí)現(xiàn)。
生產(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
publisher-returns: true # 開啟退回回調(diào)
#三個(gè)類型:none默認(rèn)不開啟確認(rèn)回調(diào) correlated開啟確認(rèn)回調(diào)
#simple也會(huì)確認(rèn)回調(diào) 還會(huì)調(diào)用waitForConfirms()方法或waitForConfirmsOrDie()方法
publisher-confirm-type: correlated # 開啟確認(rèn)回調(diào)
業(yè)務(wù)邏輯
@Component public class RabbitMQConfig { public static final String EXCHANGE_NAME = "order_exchange_name"; public static final String QUEUE_NAME = "order_queue_name"; public static final String DELAY_EXCHANGE_NAME = "delay_exchange_name"; public static final String DELAY_QUEUE_NAME = "delay_queue_name"; @Bean("orderExchange") public Exchange testExchange(){ return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build(); } @Bean("delayExchange") public Exchange deadExchange(){ return ExchangeBuilder.topicExchange(DELAY_EXCHANGE_NAME).durable(true).build(); } //訂單隊(duì)列綁定延遲交換機(jī)并且?guī)下酚涉I @Bean("orderQueue") public Queue testQueue(){ return QueueBuilder.durable(QUEUE_NAME).deadLetterExchange(DELAY_EXCHANGE_NAME) .deadLetterRoutingKey("order.delay.user").build(); } @Bean("delayQueue") public Queue deadQueue(){ return QueueBuilder.durable(DELAY_QUEUE_NAME).build(); } @Bean public Binding link(@Qualifier("orderExchange") Exchange exchange, @Qualifier("orderQueue") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs(); } @Bean public Binding deadLink(@Qualifier("delayExchange") Exchange exchange, @Qualifier("delayQueue") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("order.delay.#").noargs(); } }
@SpringBootTest @RunWith(SpringRunner.class) class RabbitmqProducerApplicationTests { @Autowired private RabbitTemplate rabbitTemplate; @Test void testProducer() throws InterruptedException { 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); } }); // 設(shè)置交換機(jī)處理失敗消息的模式 // true:消息到達(dá)不了隊(duì)列時(shí) 會(huì)將消息重新返回給生產(chǎn)者 false:消息到達(dá)不了隊(duì)列直接丟棄 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)"); } }); // 以上代碼均是為了保證消息的可靠性傳遞 // 對消息進(jìn)行后置處理 設(shè)置其過期時(shí)間為10s MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("10000"); return message; } }; // 下單成功發(fā)送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.delay.user","ikun書籍", messagePostProcessor); } }
測試結(jié)果
消費(fèi)者端
目錄結(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>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> </dependency> </dependencies>
修改yml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
simple:
acknowledge-mode: manual # 開啟手動(dòng)確認(rèn)
業(yè)務(wù)邏輯
@Slf4j @Component public class OrderListener implements ChannelAwareMessageListener { @RabbitListener(queues = "delay_queue_name") // 監(jiān)聽的是死信隊(duì)列 @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { // 這里要注意在監(jiān)聽這條消息前 肯定會(huì)有接口點(diǎn)擊支付去更改支付狀態(tài)的業(yè)務(wù)邏輯 此處我是做了10s的訂單業(yè)務(wù)而已 log.info("您在時(shí)間為:{},時(shí)有一條訂單為:{}", LocalDateTime.now().minusSeconds(10), new String(message.getBody())); // 下面開始接受訂單消息邏輯 log.info("將訂單id傳入數(shù)據(jù)庫查詢訂單支付字段"); log.info("字段為支付成功狀態(tài)就手動(dòng)確認(rèn)簽收"); log.info("字段為未支付狀態(tài)就取消訂單并且回滾事務(wù)"); channel.basicAck(deliveryTag,false);// 僅確認(rèn)本次消息 } catch (Exception e){ log.info("出現(xiàn)異常 拒絕簽收消息 并且不重回隊(duì)列"); channel.basicNack(deliveryTag,false,false); } } }
測試結(jié)果
到此這篇關(guān)于詳解RabbitMQ中延遲隊(duì)列結(jié)合業(yè)務(wù)場景的使用的文章就介紹到這了,更多相關(guān)RabbitMQ延遲隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Java實(shí)現(xiàn)圖片旋轉(zhuǎn)、指定圖像大小和水平翻轉(zhuǎn)
這篇文章主要為大家詳細(xì)介紹了Java實(shí)現(xiàn)圖像旋轉(zhuǎn),指定圖像大小,水平翻轉(zhuǎn)圖像,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-02-02Java實(shí)現(xiàn)簡易圖書借閱系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了Java實(shí)現(xiàn)簡易圖書借閱系統(tǒng),文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2022-03-03java+jsp+struts2實(shí)現(xiàn)發(fā)送郵件功能
這篇文章主要為大家詳細(xì)介紹了java+jsp+struts2實(shí)現(xiàn)發(fā)送郵件功能,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-03-03java編程實(shí)現(xiàn)郵件定時(shí)發(fā)送的方法
這篇文章主要介紹了java編程實(shí)現(xiàn)郵件定時(shí)發(fā)送的方法,涉及Java基于定時(shí)器實(shí)現(xiàn)計(jì)劃任務(wù)的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-11-11Java數(shù)據(jù)結(jié)構(gòu)之鏈表詳解
本篇文章我們將講解一種新型的數(shù)據(jù)結(jié)構(gòu)—鏈表,鏈表是一種使用廣泛的通用數(shù)據(jù)結(jié)構(gòu),它可以用來作為實(shí)現(xiàn)棧,隊(duì)列等數(shù)據(jù)結(jié)構(gòu)的基礎(chǔ).文中有非常詳細(xì)的介紹,需要的朋友可以參考下2021-05-05Java使用線程池批量處理數(shù)據(jù)操作具體流程
這篇文章主要給大家介紹了關(guān)于Java使用線程池批量處理數(shù)據(jù)操作的相關(guān)資料,Java多線程編程中線程池是一個(gè)非常重要的概念,線程池可以提高線程的復(fù)用率和任務(wù)調(diào)度的效率,尤其是當(dāng)需要查詢大批量數(shù)據(jù)時(shí),需要的朋友可以參考下2023-06-06Java(springboot) 讀取txt文本內(nèi)容代碼實(shí)例
這篇文章主要介紹了Java(springboot) 讀取txt文本內(nèi)容代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-02-02MyBatisPlus使用${ew.customSqlSegment}別名問題解決
在使用MyBatisPlus進(jìn)行連表查詢時(shí),可能遇到因${ew.customSqlSegment}無法加別名的問題,本文就來介紹一下如何解決,感興趣的可以了解一下2024-10-10