SpringBoot+RabbitMQ?實(shí)現(xiàn)死信隊(duì)列的示例
前言
死信:無(wú)法被消費(fèi)的消息,稱(chēng)為死信。
如果死信一直留在隊(duì)列中,會(huì)導(dǎo)致一直被消費(fèi),卻從不消費(fèi)成功。
所以我們專(zhuān)門(mén)開(kāi)辟了一個(gè)來(lái)存放死信的隊(duì)列,叫死信隊(duì)列(DLX,dead-letter-exchange)。
死信的幾種來(lái)源:
- 消息 TTL 過(guò)期(time to live,存活時(shí)間,可以用在限時(shí)支付消息)
- 隊(duì)列達(dá)到最大長(zhǎng)度(隊(duì)列滿(mǎn)了,無(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 秒) ---- 沒(méi)有消費(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ī),開(kāi)啟持久化
*/
@Bean
DirectExchange normalExchange() {
return new DirectExchange("normalExchange", true, false);
}
@Bean
public Queue normalQueue() {
// durable: 是否持久化,默認(rèn)是false,持久化隊(duì)列:會(huì)被存儲(chǔ)在磁盤(pán)上,當(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)沒(méi)有生產(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ì)列沒(méi)有訪(fǎng)問(wèn)超時(shí)時(shí),自動(dòng)刪除(包含沒(méi)有消費(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ù)放入磁盤(pán),消費(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)沒(mé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)文章
spring-boot-starter-web更換默認(rèn)Tomcat容器的方法
Spring Boot支持容器的自動(dòng)配置,默認(rèn)是Tomcat,當(dāng)然我們也是可以進(jìn)行修改的。下面小編給大家?guī)?lái)了spring-boot-starter-web更換默認(rèn)Tomcat容器的方法,感興趣的朋友跟隨小編一起看看吧2019-04-04
java使用spring實(shí)現(xiàn)發(fā)送mail的方法
這篇文章主要介紹了java使用spring實(shí)現(xiàn)發(fā)送mail的方法,涉及java基于spring框架發(fā)送郵件的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-10-10
基于logback實(shí)現(xiàn)純java版本的SDK組件
這篇文章主要介紹了基于logback實(shí)現(xiàn)純java版本的SDK組件,在項(xiàng)目開(kāi)發(fā)過(guò)程中通常會(huì)使用logback作為日志記錄的依賴(lài)工具,使用方式是引入logback相關(guān)jar包,然后配置logback.xml配置文件的方式來(lái)實(shí)現(xiàn),需要的朋友可以參考下2023-11-11
Java重寫(xiě)(Override)與重載(Overload)區(qū)別原理解析
這篇文章主要介紹了Java重寫(xiě)(Override)與重載(Overload)區(qū)別原理解析,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-02-02
Java?Servlet實(shí)現(xiàn)表白墻的代碼實(shí)例
最近用Servlet做了個(gè)小項(xiàng)目,分享給大家,下面這篇文章主要給大家介紹了關(guān)于Java?Servlet實(shí)現(xiàn)表白墻的相關(guān)資料,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-02-02
Java中BufferedReader與BufferedWriter類(lèi)的使用示例
BufferedReader與BufferedWriter分別繼承于Reader和Writer類(lèi),分別為字符的讀取和寫(xiě)入添加緩沖功能,這里我們就來(lái)看一下Java中BufferedReader與BufferedWriter類(lèi)的使用示例:2016-06-06

