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

環(huán)境準(zhǔn)備配置
準(zhǔn)備 MQ 的隊列和環(huán)境:
- 正常交換機
- 正常隊列(最長隊列 5) ---- 正常消費者,拒絕消息
- ttl 隊列(過期時間 60 秒) ---- 沒有消費者
- 死信交換機
- 死信隊列
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 #手動模式 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 {
/* 正常配置 **********************************************************************************************************/
/**
* 正常交換機,開啟持久化
*/
@Bean
DirectExchange normalExchange() {
return new DirectExchange("normalExchange", true, false);
}
@Bean
public Queue normalQueue() {
// durable: 是否持久化,默認(rèn)是false,持久化隊列:會被存儲在磁盤上,當(dāng)消息代理重啟時仍然存在,暫存隊列:當(dāng)前連接有效
// exclusive: 默認(rèn)也是false,只能被當(dāng)前創(chuàng)建的連接使用,而且當(dāng)連接關(guān)閉后隊列即被刪除。此參考優(yōu)先級高于durable
// autoDelete: 是否自動刪除,當(dāng)沒有生產(chǎn)者或者消費者使用此隊列,該隊列會自動刪除。
Map<String, Object> args = deadQueueArgs();
// 隊列設(shè)置最大長度
args.put("x-max-length", 5);
return new Queue("normalQueue", true, false, false, args);
}
@Bean
public Queue ttlQueue() {
Map<String, Object> args = deadQueueArgs();
// 隊列設(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");
}
/* 死信配置 **********************************************************************************************************/
/**
* 死信交換機
*/
@Bean
DirectExchange deadExchange() {
return new DirectExchange("deadExchange", true, false);
}
/**
* 死信隊列
*/
@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ā)到 死信隊列,配置參數(shù)
*/
private Map<String, Object> deadQueueArgs() {
Map<String, Object> map = new HashMap<>();
// 綁定該隊列到私信交換機
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ā)送到隊列的消息在丟棄之前可以存活多長時間(毫秒) |
| x-max-length | 限制隊列最大長度(新增后擠出最早的),單位個數(shù) |
| x-expires | 隊列沒有訪問超時時,自動刪除(包含沒有消費的消息),單位毫秒 |
| x-max-length-bytes | 限制隊列最大容量 |
| x-dead-letter-exchange | 死信交換機,將刪除/過期的數(shù)據(jù),放入指定交換機 |
| x-dead-letter-routing-key | 死信路由,將刪除/過期的數(shù)據(jù),放入指定routingKey |
| x-max-priority | 隊列優(yōu)先級 |
| x-queue-mode | 對列模式,默認(rèn)lazy(將數(shù)據(jù)放入磁盤,消費時放入內(nèi)存) |
| x-queue-master-locator | 鏡像隊列 |
正常隊列消費
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;
/**
* 正常消息隊列,隊列最大長度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() + ", 正常隊列消息,最大長度 5");
rabbitTemplate.convertAndSend("normalExchange", "normalRouting", map, new CorrelationData());
return JSONObject.toJSONString(map);
}
}MQ控制臺:

消息 TTL 過期隊列自動消費
消息的TTL 指的是消息的存活時間,我們可以通過設(shè)置消息的TTL或者隊列的TTL來實現(xiàn)。
消息的TTL :對于設(shè)置了過期時間屬性(expiration)的消息,消息如果在過期時間內(nèi)沒被消費,會過期 隊列的TTL :對于設(shè)置了過期時間屬性(x-message-ttl)的隊列,所有路由到這個隊列的消息,都會設(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隊列消息");
rabbitTemplate.convertAndSend("normalExchange", "ttlRouting", map, new CorrelationData());
return JSONObject.toJSONString(map);
}發(fā)送后:

過期后:

Demo 中只是為了方便,代碼中盡量使用 消息TTL,不要用 隊列TTL。
死信隊列消費
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("死信隊列收到消息 : " + message.toString());
channel.basicAck(mqMsg.getMessageProperties().getDeliveryTag(), false);
}
}測試結(jié)果:
死信隊列收到消息 : {data=1655109039693, deadQueue隊列消息, messageId=9712be7c-ee04-4f8f-af00-b73b5cdcd086}
死信隊列收到消息 : {data=1655109040100, deadQueue隊列消息, messageId=76dc4ebb-5f61-4c2e-8dff-c6607f22b112}
死信隊列收到消息 : {data=1655109040358, deadQueue隊列消息, messageId=ca45aa49-6e3d-4547-a622-c53110e208a4}
死信隊列收到消息 : {data=1655109040611, deadQueue隊列消息, messageId=fc3d2065-4acd-48bb-bf85-6e85b2e3ae72}
死信隊列收到消息 : {data=1655109040877, deadQueue隊列消息, messageId=a1874299-8773-4069-8b07-c9a644df0787}
死信隊列收到消息 : {data=1655109041133, deadQueue隊列消息, messageId=dd6bc562-8454-424d-8a6a-84c2379c0970}
到此這篇關(guān)于SpringBoot+RabbitMQ 實現(xiàn)死信隊列的示例的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMQ 死信隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- SpringBoot整合RabbitMQ處理死信隊列和延遲隊列
- 如何利用rabbitMq的死信隊列實現(xiàn)延時消息
- 深入分析RabbitMQ中死信隊列與死信交換機
- 關(guān)于SpringBoot整合RabbitMQ實現(xiàn)死信隊列
- 關(guān)于Rabbitmq死信隊列及延時隊列的實現(xiàn)
- Springboot結(jié)合rabbitmq實現(xiàn)的死信隊列
- RabbitMQ之死信隊列深入解析
- springboot中RabbitMQ死信隊列的實現(xiàn)示例
- SpringBoot整合RabbitMQ實現(xiàn)延遲隊列和死信隊列
- springboot整合RabbitMQ中死信隊列的實現(xiàn)
相關(guān)文章
使用PageHelper插件實現(xiàn)Service層分頁
這篇文章主要為大家詳細(xì)介紹了使用PageHelper插件實現(xiàn)Service層分頁,文中示例代碼介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們可以參考一下2022-04-04
java利用StringTokenizer分割字符串的實現(xiàn)
利用java.util.StringTokenizer的方法,可以將一個字符串拆分為一系列的標(biāo)記,本文就來介紹一下java利用StringTokenizer分割字符串的實現(xiàn),感興趣的可以了解一下2023-10-10
Mybatis-plus中IService接口的基本使用步驟
Mybatis-plus是一個Mybatis的增強工具,它提供了很多便捷的方法來簡化開發(fā),IService是Mybatis-plus提供的通用service接口,封裝了常用的數(shù)據(jù)庫操作方法,包括增刪改查等,下面這篇文章主要給大家介紹了關(guān)于Mybatis-plus中IService接口的基本使用步驟,需要的朋友可以參考下2023-06-06
springboot實戰(zhàn)權(quán)限管理功能圖文步驟附含源碼
這篇文章主要為大家介紹了springboot實戰(zhàn)權(quán)限管理功能圖文步驟及示例源碼,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-06-06
詳解Elasticsearch如何把一個索引變?yōu)橹蛔x
這篇文章主要為大家介紹了詳解Elasticsearch如何把一個索引變?yōu)橹蛔x示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-02-02
Json字符串轉(zhuǎn)Java對象和List代碼實例
這篇文章主要介紹了Json字符串轉(zhuǎn)Java對象和List代碼實例,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-06-06
詳解Spring框架注解掃描開啟之配置細(xì)節(jié)
本篇文章主要介紹了詳解Spring框架注解掃描開啟之配置細(xì)節(jié),小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2017-08-08
利用Spring Cloud Config結(jié)合Bus實現(xiàn)分布式配置中心的步驟
這篇文章主要介紹了利用Spring Cloud Config結(jié)合Bus實現(xiàn)分布式配置中心的相關(guān)資料,文中通過示例代碼將實現(xiàn)的步驟一步步介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友下面來一起看看吧2018-05-05

