SpringBoot+RabbitMQ?實(shí)現(xiàn)死信隊(duì)列的示例
前言
死信:無法被消費(fèi)的消息,稱為死信。
如果死信一直留在隊(duì)列中,會(huì)導(dǎo)致一直被消費(fèi),卻從不消費(fèi)成功。
所以我們專門開辟了一個(gè)來存放死信的隊(duì)列,叫死信隊(duì)列(DLX,dead-letter-exchange
)。
死信的幾種來源:
- 消息 TTL 過期(time to live,存活時(shí)間,可以用在限時(shí)支付消息)
- 隊(duì)列達(dá)到最大長度(隊(duì)列滿了,無法路由到該隊(duì)列)
- 消息被拒絕(
basic.reject / basic.nack
),并且requeue = false
環(huán)境準(zhǔn)備配置
準(zhǔn)備 MQ 的隊(duì)列和環(huán)境:
- 正常交換機(jī)
- 正常隊(duì)列(最長隊(duì)列 5) ---- 正常消費(fèi)者,拒絕消息
- ttl 隊(duì)列(過期時(shí)間 60 秒) ---- 沒有消費(fèi)者
- 死信交換機(jī)
- 死信隊(duì)列
Properties 配置
spring.application.name=spring-boot-rabbitmq spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=admin #手動(dòng)模式 spring.rabbitmq.listener.simple.acknowledge-mode=manual
Config配置
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; import java.util.Map; /** * @author wxd * @version V1.0 * @description DeadConfig * @date 2022/6/13 13:55 **/ @Configuration public class DeadConfig { /* 正常配置 **********************************************************************************************************/ /** * 正常交換機(jī),開啟持久化 */ @Bean DirectExchange normalExchange() { return new DirectExchange("normalExchange", true, false); } @Bean public Queue normalQueue() { // durable: 是否持久化,默認(rèn)是false,持久化隊(duì)列:會(huì)被存儲(chǔ)在磁盤上,當(dāng)消息代理重啟時(shí)仍然存在,暫存隊(duì)列:當(dāng)前連接有效 // exclusive: 默認(rèn)也是false,只能被當(dāng)前創(chuàng)建的連接使用,而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級(jí)高于durable // autoDelete: 是否自動(dòng)刪除,當(dāng)沒有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列,該隊(duì)列會(huì)自動(dòng)刪除。 Map<String, Object> args = deadQueueArgs(); // 隊(duì)列設(shè)置最大長度 args.put("x-max-length", 5); return new Queue("normalQueue", true, false, false, args); } @Bean public Queue ttlQueue() { Map<String, Object> args = deadQueueArgs(); // 隊(duì)列設(shè)置消息過期時(shí)間 60 秒 args.put("x-message-ttl", 60 * 1000); return new Queue("ttlQueue", true, false, false, args); } @Bean Binding normalRouteBinding() { return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normalRouting"); } @Bean Binding ttlRouteBinding() { return BindingBuilder.bind(ttlQueue()).to(normalExchange()).with("ttlRouting"); } /* 死信配置 **********************************************************************************************************/ /** * 死信交換機(jī) */ @Bean DirectExchange deadExchange() { return new DirectExchange("deadExchange", true, false); } /** * 死信隊(duì)列 */ @Bean public Queue deadQueue() { return new Queue("deadQueue", true, false, false); } @Bean Binding deadRouteBinding() { return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("deadRouting"); } /** * 轉(zhuǎn)發(fā)到 死信隊(duì)列,配置參數(shù) */ private Map<String, Object> deadQueueArgs() { Map<String, Object> map = new HashMap<>(); // 綁定該隊(duì)列到私信交換機(jī) map.put("x-dead-letter-exchange", "deadExchange"); map.put("x-dead-letter-routing-key", "deadRouting"); return map; }
參數(shù):
參數(shù)名 | 作用 |
---|---|
x-message-tti(Time-To-Live) | 發(fā)送到隊(duì)列的消息在丟棄之前可以存活多長時(shí)間(毫秒) |
x-max-length | 限制隊(duì)列最大長度(新增后擠出最早的),單位個(gè)數(shù) |
x-expires | 隊(duì)列沒有訪問超時(shí)時(shí),自動(dòng)刪除(包含沒有消費(fèi)的消息),單位毫秒 |
x-max-length-bytes | 限制隊(duì)列最大容量 |
x-dead-letter-exchange | 死信交換機(jī),將刪除/過期的數(shù)據(jù),放入指定交換機(jī) |
x-dead-letter-routing-key | 死信路由,將刪除/過期的數(shù)據(jù),放入指定routingKey |
x-max-priority | 隊(duì)列優(yōu)先級(jí) |
x-queue-mode | 對(duì)列模式,默認(rèn)lazy(將數(shù)據(jù)放入磁盤,消費(fèi)時(shí)放入內(nèi)存) |
x-queue-master-locator | 鏡像隊(duì)列 |
正常隊(duì)列消費(fèi)
package com.wxd.controller; import com.alibaba.fastjson.JSONObject; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import java.util.HashMap; import java.util.Map; import java.util.UUID; /** * @author wxd * @version V1.0 * @description HelloController * @date 2022/6/13 13:59 **/ @RestController public class HelloController { @Autowired private RabbitTemplate rabbitTemplate; /** * 正常消息隊(duì)列,隊(duì)列最大長度5 */ @GetMapping("/normalQueue") public String normalQueue() { Map<String, Object> map = new HashMap<>(2); map.put("messageId", String.valueOf(UUID.randomUUID())); map.put("data", System.currentTimeMillis() + ", 正常隊(duì)列消息,最大長度 5"); rabbitTemplate.convertAndSend("normalExchange", "normalRouting", map, new CorrelationData()); return JSONObject.toJSONString(map); } }
MQ控制臺(tái):
消息 TTL 過期隊(duì)列自動(dòng)消費(fèi)
消息的TTL 指的是消息的存活時(shí)間,我們可以通過設(shè)置消息的TTL或者隊(duì)列的TTL來實(shí)現(xiàn)。
消息的TTL :對(duì)于設(shè)置了過期時(shí)間屬性(expiration)的消息,消息如果在過期時(shí)間內(nèi)沒被消費(fèi),會(huì)過期 隊(duì)列的TTL :對(duì)于設(shè)置了過期時(shí)間屬性(x-message-ttl)的隊(duì)列,所有路由到這個(gè)隊(duì)列的消息,都會(huì)設(shè)置上這個(gè)過期時(shí)間。
代碼實(shí)現(xiàn):
/** * 消息 TTL, time to live */ @GetMapping("/ttlToDead") public String ttlToDead() { Map<String, Object> map = new HashMap<>(); map.put("messageId", String.valueOf(UUID.randomUUID())); map.put("data", System.currentTimeMillis() + ", ttl隊(duì)列消息"); rabbitTemplate.convertAndSend("normalExchange", "ttlRouting", map, new CorrelationData()); return JSONObject.toJSONString(map); }
發(fā)送后:
過期后:
Demo 中只是為了方便,代碼中盡量使用 消息TTL,不要用 隊(duì)列TTL。
死信隊(duì)列消費(fèi)
import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.Map; /** * @author wxd * @version V1.0 * @description DeadConsumer * @date 2022/6/13 14:56 **/ @Component @RabbitListener(queues = "deadQueue") public class DeadConsumer { @RabbitHandler public void process(Map<String, Object> message, Channel channel, Message mqMsg) throws IOException { System.out.println("死信隊(duì)列收到消息 : " + message.toString()); channel.basicAck(mqMsg.getMessageProperties().getDeliveryTag(), false); } }
測(cè)試結(jié)果:
死信隊(duì)列收到消息 : {data=1655109039693, deadQueue隊(duì)列消息, messageId=9712be7c-ee04-4f8f-af00-b73b5cdcd086}
死信隊(duì)列收到消息 : {data=1655109040100, deadQueue隊(duì)列消息, messageId=76dc4ebb-5f61-4c2e-8dff-c6607f22b112}
死信隊(duì)列收到消息 : {data=1655109040358, deadQueue隊(duì)列消息, messageId=ca45aa49-6e3d-4547-a622-c53110e208a4}
死信隊(duì)列收到消息 : {data=1655109040611, deadQueue隊(duì)列消息, messageId=fc3d2065-4acd-48bb-bf85-6e85b2e3ae72}
死信隊(duì)列收到消息 : {data=1655109040877, deadQueue隊(duì)列消息, messageId=a1874299-8773-4069-8b07-c9a644df0787}
死信隊(duì)列收到消息 : {data=1655109041133, deadQueue隊(duì)列消息, messageId=dd6bc562-8454-424d-8a6a-84c2379c0970}
到此這篇關(guān)于SpringBoot+RabbitMQ 實(shí)現(xiàn)死信隊(duì)列的示例的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMQ 死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- SpringBoot整合RabbitMQ處理死信隊(duì)列和延遲隊(duì)列
- 如何利用rabbitMq的死信隊(duì)列實(shí)現(xiàn)延時(shí)消息
- 深入分析RabbitMQ中死信隊(duì)列與死信交換機(jī)
- 關(guān)于SpringBoot整合RabbitMQ實(shí)現(xiàn)死信隊(duì)列
- 關(guān)于Rabbitmq死信隊(duì)列及延時(shí)隊(duì)列的實(shí)現(xiàn)
- Springboot結(jié)合rabbitmq實(shí)現(xiàn)的死信隊(duì)列
- RabbitMQ之死信隊(duì)列深入解析
- springboot中RabbitMQ死信隊(duì)列的實(shí)現(xiàn)示例
- SpringBoot整合RabbitMQ實(shí)現(xiàn)延遲隊(duì)列和死信隊(duì)列
- springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)
相關(guān)文章
使用PageHelper插件實(shí)現(xiàn)Service層分頁
這篇文章主要為大家詳細(xì)介紹了使用PageHelper插件實(shí)現(xiàn)Service層分頁,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2022-04-04java利用StringTokenizer分割字符串的實(shí)現(xiàn)
利用java.util.StringTokenizer的方法,可以將一個(gè)字符串拆分為一系列的標(biāo)記,本文就來介紹一下java利用StringTokenizer分割字符串的實(shí)現(xiàn),感興趣的可以了解一下2023-10-10Mybatis-plus中IService接口的基本使用步驟
Mybatis-plus是一個(gè)Mybatis的增強(qiáng)工具,它提供了很多便捷的方法來簡(jiǎn)化開發(fā),IService是Mybatis-plus提供的通用service接口,封裝了常用的數(shù)據(jù)庫操作方法,包括增刪改查等,下面這篇文章主要給大家介紹了關(guān)于Mybatis-plus中IService接口的基本使用步驟,需要的朋友可以參考下2023-06-06springboot實(shí)戰(zhàn)權(quán)限管理功能圖文步驟附含源碼
這篇文章主要為大家介紹了springboot實(shí)戰(zhàn)權(quán)限管理功能圖文步驟及示例源碼,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-06-06詳解Elasticsearch如何把一個(gè)索引變?yōu)橹蛔x
這篇文章主要為大家介紹了詳解Elasticsearch如何把一個(gè)索引變?yōu)橹蛔x示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-02-02Json字符串轉(zhuǎn)Java對(duì)象和List代碼實(shí)例
這篇文章主要介紹了Json字符串轉(zhuǎn)Java對(duì)象和List代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-06-06Java每隔兩個(gè)數(shù)刪掉一個(gè)數(shù)問題詳解
這篇文章主要介紹了Java每隔兩個(gè)數(shù)刪掉一個(gè)數(shù)問題詳解,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-04-04詳解Spring框架注解掃描開啟之配置細(xì)節(jié)
本篇文章主要介紹了詳解Spring框架注解掃描開啟之配置細(xì)節(jié),小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧2017-08-08利用Spring Cloud Config結(jié)合Bus實(shí)現(xiàn)分布式配置中心的步驟
這篇文章主要介紹了利用Spring Cloud Config結(jié)合Bus實(shí)現(xiàn)分布式配置中心的相關(guān)資料,文中通過示例代碼將實(shí)現(xiàn)的步驟一步步介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友下面來一起看看吧2018-05-05