前言：實際專案中經常遇到訊息消費失敗了，要進行訊息的重發，比如支付訊息消費失敗后，要分不同時間段進行N次的訊息重發提醒，
本文模擬場景
- 當金額少于100時，訊息消費成功
- 當金額大于100，小于200時，會進行3次重發，第一次1秒；第二次2秒；第三次3秒，
- 當金額大于200時，訊息消費失敗，會進行5次重發，第一次1秒；第二次2秒；第三次3秒；第四次4秒；第五次5秒，重試五次后，訊息自動進入死信佇列，在死信佇列存活60秒后消失，
代碼實體
特別注意代碼與組態檔中的注釋，各個使用說明都已經詳細寫在組態檔中
pom包引入
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.12.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.cloudstream</groupId>
<artifactId>demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
<spring-cloud.version>Greenwich.SR5</spring-cloud.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- ①關鍵配置：引入stream-rabbit 依賴-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<!-- ②關鍵配置：由于stream是基于spring-cloud的，所以這里要引入 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
</repository>
</repositories>
</project>
配置application.yml檔案
注意各個配置的縮進格式，別搞錯了
server:
port: 8081
spring:
application:
name: stream-demo
#rabbitmq連接配置
rabbitmq:
host: 127.0.0.1
port: 5672
username: admin
password: 123456
cloud:
stream:
bindings:
#訊息生產者，與DelayDemoTopic介面中的DELAY_DEMO_PRODUCER變數值一致
delay-demo-producer:
#①定義交換機名
destination: demo-delay-queue
#訊息消費者，與DelayDemoTopic介面中的DELAY_DEMO_CONSUMER變數值一致
delay-demo-consumer:
#定義交換機名，與①一致，就可以使發送和消費都指向一個佇列
destination: demo-delay-queue
#分組，這個配置可以開啟訊息持久化、可以解決在集群環境下重復消費的問題，
#比如A、B兩臺服務器集群，如果沒有這個配置，則A、B都能收到同樣的訊息，如果有該配置則只有其中一臺會收到訊息
group: delay-consumer-group
consumer:
#最大重試次數，默認為3，不使用默認的，這里定義為1，由我們程式控制發送時間和次數
maxAttempts: 1
rabbit:
bindings:
#訊息生產者，與DelayDemoTopic介面中的DELAY_DEMO_PRODUCER變數值一致
delay-demo-producer:
producer:
#②申明為延遲佇列
delayedExchange: true
#訊息消費者，與DelayDemoTopic介面中的DELAY_DEMO_CONSUMER變數值一致
delay-demo-consumer:
consumer:
#申明為延遲佇列，與②的配置的成對出現的
delayedExchange: true
#開啟死信佇列
autoBindDlq: true
#死信佇列中訊息的存活時間
dlqTtl: 60000
定義佇列通道
- 定義通道
/**
* 定義延遲訊息通道
*/
public interface DelayDemoTopic {
/**
* 生產者，與yml檔案配置對應
*/
String DELAY_DEMO_PRODUCER = "delay-demo-producer";
/**
* 消費者，與yml檔案配置對應
*/
String DELAY_DEMO_CONSUMER = "delay-demo-consumer";
/**
* 定義訊息消費者，在@StreamListener監聽訊息的時候用到
* @return
*/
@Input(DELAY_DEMO_CONSUMER)
SubscribableChannel delayDemoConsumer();
/**
* 定義訊息發送者，在發送訊息的時候用到
* @return
*/
@Output(DELAY_DEMO_PRODUCER)
MessageChannel delayDemoProducer();
}
- 系結通道
/**
* 配置訊息的binding
*
*/
@EnableBinding(value = https://www.cnblogs.com/mingsay/p/{DelayDemoTopic.class})
@Component
public class MessageConfig {
}
訊息發送模擬
/**
* 發送訊息
*/
@RestController
public class SendMessageController {
@Autowired
DelayDemoTopic delayDemoTopic;
@GetMapping("send")
public Boolean sendMessage(BigDecimal money) throws JsonProcessingException {
Message<BigDecimal> message = MessageBuilder.withPayload(money)
//設定訊息的延遲時間，首次發送，不設定延遲時間，直接發送
.setHeader(DelayConstant.X_DELAY_HEADER,0)
//設定訊息已經重試的次數，首次發送，設定為0
.setHeader(DelayConstant.X_RETRIES_HEADER,0)
.build();
return delayDemoTopic.delayDemoProducer().send(message);
}
}
訊息監聽處理
@Component
@Slf4j
public class DelayDemoTopicListener {
@Autowired
DelayDemoTopic delayDemoTopic;
/**
* 監聽延遲訊息通道中的訊息
* @param message
*/
@StreamListener(value = https://www.cnblogs.com/mingsay/p/DelayDemoTopic.DELAY_DEMO_CONSUMER)
public void listener(Message message) {
//獲取重試次數
int retries = (int)message.getHeaders().get(DelayConstant.X_RETRIES_HEADER);
//獲取訊息內容
BigDecimal money = message.getPayload();
try {
String now = DateUtils.formatDate(new Date(),"yyyy-MM-dd HH:mm:ss");
//模擬：如果金額大于200，則訊息無法消費成功；金額如果大于100，則重試3次；如果金額小于100，直接消費成功
if (money.compareTo(new BigDecimal(200)) == 1){
throw new RuntimeException(now+":金額超出200，無法交易，");
}else if (money.compareTo(new BigDecimal(100)) == 1 && retries <= 3) {
if (retries == 0) {
throw new RuntimeException(now+":金額超出100，消費失敗，將進入重試，");
}else {
throw new RuntimeException(now+":金額超出100，當前第" + retries + "次重試，");
}
}else {
log.info("訊息消費成功！");
}
}catch (Exception e) {
log.error(e.getMessage());
if (retries < DelayConstant.X_RETRIES_TOTAL){
//將訊息重新塞入佇列
MessageBuilder<BigDecimal> messageBuilder = MessageBuilder.fromMessage(message)
//設定訊息的延遲時間
.setHeader(DelayConstant.X_DELAY_HEADER,DelayConstant.ruleMap.get(retries + 1))
//設定訊息已經重試的次數
.setHeader(DelayConstant.X_RETRIES_HEADER,retries + 1);
Message<BigDecimal> reMessage = messageBuilder.build();
//將訊息重新發送到延遲佇列中
delayDemoTopic.delayDemoProducer().send(reMessage);
}else {
//超過重試次數，做相關處理（比如保存資料庫等操作），如果拋出例外，則會自動進入死信佇列
throw new RuntimeException("超過最大重試次數：" + DelayConstant.X_RETRIES_TOTAL);
}
}
}
}
規則定義
目前寫在一個常量類里，實際專案中，通常會配置在組態檔中
public class DelayConstant {
/**
* 定義當前重試次數
*/
public static final String X_RETRIES_HEADER = "x-retries";
/**
* 定義延遲訊息，固定值，該配置放到訊息的header中，會開啟延遲佇列
*/
public static final String X_DELAY_HEADER = "x-delay";
/**
* 定義最多重試次數
*/
public static final Integer X_RETRIES_TOTAL = 5;
/**
* 定義重試規則，毫秒為單位
*/
public static final Map<Integer,Integer> ruleMap = new HashMap(){{
put(1,1000);
put(2,2000);
put(3,3000);
put(4,4000);
put(5,5000);
}};
}
測驗
經過以上配置和實作就可完成模擬的重發場景，
- 瀏覽器中輸入
http://127.0.0.1:8081/send?money=10，可以看到控制臺中輸出：
訊息消費成功！
- 瀏覽器中輸入
http://127.0.0.1:8081/send?money=110，可以看到控制臺中輸出：
2020-06-20 10:59:42:金額超出100，消費失敗，將進入重試，
2020-06-20 10:59:43:金額超出100，當前第1次重試，
2020-06-20 10:59:45:金額超出100，當前第2次重試，
2020-06-20 10:59:48:金額超出100，當前第3次重試，
訊息消費成功！
- 瀏覽器中輸入
http://127.0.0.1:8081/send?money=110，可以看到控制臺中輸出：
注意事項
由于本文用到了延遲佇列，需要在rabbitMQ中安裝延遲插件，具體安裝方式，可以查看：延遲佇列安裝參考
原始碼獲取
以上示例都可以通過我的GitHub獲取完整的代碼.
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/159457.html
標籤：Java