SpringBoot+RabbitMQ?實(shí)現(xiàn)死信隊(duì)列的示例
前言
死信:無(wú)法被消費(fèi)的消息,稱為死信。
如果死信一直留在隊(duì)列中,會(huì)導(dǎo)致一直被消費(fèi),卻從不消費(fèi)成功。
所以我們專門開辟了一個(gè)來(lái)存放死信的隊(duì)列,叫死信隊(duì)列(DLX,dead-letter-exchange)。
死信的幾種來(lái)源:
- 消息 TTL 過(guò)期(time to live,存活時(shí)間,可以用在限時(shí)支付消息)
- 隊(duì)列達(dá)到最大長(zhǎng)度(隊(duì)列滿了,無(wú)法路由到該隊(duì)列)
- 消息被拒絕(
basic.reject / basic.nack),并且requeue = false

環(huán)境準(zhǔn)備配置
準(zhǔn)備 MQ 的隊(duì)列和環(huán)境:
- 正常交換機(jī)
- 正常隊(duì)列(最長(zhǎng)隊(duì)列 5) ---- 正常消費(fèi)者,拒絕消息
- ttl 隊(duì)列(過(guò)期時(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è)置最大長(zhǎng)度
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è)置消息過(guò)期時(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ì)列的消息在丟棄之前可以存活多長(zhǎng)時(shí)間(毫秒) |
| x-max-length | 限制隊(duì)列最大長(zhǎng)度(新增后擠出最早的),單位個(gè)數(shù) |
| x-expires | 隊(duì)列沒有訪問(wèn)超時(shí)時(shí),自動(dòng)刪除(包含沒有消費(fèi)的消息),單位毫秒 |
| x-max-length-bytes | 限制隊(duì)列最大容量 |
| x-dead-letter-exchange | 死信交換機(jī),將刪除/過(guò)期的數(shù)據(jù),放入指定交換機(jī) |
| x-dead-letter-routing-key | 死信路由,將刪除/過(guò)期的數(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ì)列最大長(zhǎng)度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ì)列消息,最大長(zhǎng)度 5");
rabbitTemplate.convertAndSend("normalExchange", "normalRouting", map, new CorrelationData());
return JSONObject.toJSONString(map);
}
}MQ控制臺(tái):

消息 TTL 過(guò)期隊(duì)列自動(dòng)消費(fèi)
消息的TTL 指的是消息的存活時(shí)間,我們可以通過(guò)設(shè)置消息的TTL或者隊(duì)列的TTL來(lái)實(shí)現(xiàn)。
消息的TTL :對(duì)于設(shè)置了過(guò)期時(shí)間屬性(expiration)的消息,消息如果在過(guò)期時(shí)間內(nèi)沒被消費(fèi),會(huì)過(guò)期 隊(duì)列的TTL :對(duì)于設(shè)置了過(guò)期時(shí)間屬性(x-message-ttl)的隊(duì)列,所有路由到這個(gè)隊(duì)列的消息,都會(huì)設(shè)置上這個(gè)過(guò)期時(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ā)送后:

過(guò)期后:

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層分頁(yè)
這篇文章主要為大家詳細(xì)介紹了使用PageHelper插件實(shí)現(xiàn)Service層分頁(yè),文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2022-04-04
java利用StringTokenizer分割字符串的實(shí)現(xiàn)
利用java.util.StringTokenizer的方法,可以將一個(gè)字符串拆分為一系列的標(biāo)記,本文就來(lái)介紹一下java利用StringTokenizer分割字符串的實(shí)現(xiàn),感興趣的可以了解一下2023-10-10
Mybatis-plus中IService接口的基本使用步驟
Mybatis-plus是一個(gè)Mybatis的增強(qiáng)工具,它提供了很多便捷的方法來(lái)簡(jiǎn)化開發(fā),IService是Mybatis-plus提供的通用service接口,封裝了常用的數(shù)據(jù)庫(kù)操作方法,包括增刪改查等,下面這篇文章主要給大家介紹了關(guān)于Mybatis-plus中IService接口的基本使用步驟,需要的朋友可以參考下2023-06-06
springboot實(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-02
Json字符串轉(zhuǎn)Java對(duì)象和List代碼實(shí)例
這篇文章主要介紹了Json字符串轉(zhuǎn)Java對(duì)象和List代碼實(shí)例,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-06-06
Java每隔兩個(gè)數(shù)刪掉一個(gè)數(shù)問(wèn)題詳解
這篇文章主要介紹了Java每隔兩個(gè)數(shù)刪掉一個(gè)數(shù)問(wèn)題詳解,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-04-04
詳解Spring框架注解掃描開啟之配置細(xì)節(jié)
本篇文章主要介紹了詳解Spring框架注解掃描開啟之配置細(xì)節(jié),小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-08-08
利用Spring Cloud Config結(jié)合Bus實(shí)現(xiàn)分布式配置中心的步驟
這篇文章主要介紹了利用Spring Cloud Config結(jié)合Bus實(shí)現(xiàn)分布式配置中心的相關(guān)資料,文中通過(guò)示例代碼將實(shí)現(xiàn)的步驟一步步介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友下面來(lái)一起看看吧2018-05-05

