Spring Boot集成Redis Stream消息隊(duì)列從入門到實(shí)戰(zhàn)指南
更新時(shí)間：2026年03月19日 10:59:31 作者：海洋的漁夫
Redis Stream作為Redis 5.0引入的新數(shù)據(jù)類型,提供了完整的消息隊(duì)列功能,成為輕量級(jí)消息中間件的優(yōu)秀選擇,本文給大家介紹Spring Boot集成Redis Stream消息隊(duì)列從入門到實(shí)戰(zhàn)指南,感興趣的朋友跟隨小編一起看看吧
在現(xiàn)代分布式系統(tǒng)中，消息隊(duì)列是實(shí)現(xiàn)系統(tǒng)解耦、異步處理和流量削峰的重要組件。Redis Stream作為Redis 5.0引入的新數(shù)據(jù)類型，提供了完整的消息隊(duì)列功能，成為輕量級(jí)消息中間件的優(yōu)秀選擇。
前言
早期我使用 redis pubsub 的方式實(shí)現(xiàn)了消息訂閱，但是在使用過(guò)程，發(fā)現(xiàn)如果部署多實(shí)例將會(huì)重復(fù)處理消息事件，導(dǎo)致業(yè)務(wù)重復(fù)處理的情況。
根本原因是 redis pubsub 都是直接廣播，無(wú)法控制多實(shí)例重復(fù)訂閱消息的情況。
為了解決這個(gè)問(wèn)題，則需要改動(dòng) Redis Stream 的方式來(lái)構(gòu)建消息隊(duì)列。
以下我分別三個(gè)案例逐步實(shí)踐，達(dá)到一個(gè)可以適應(yīng)生產(chǎn)環(huán)境要求的成熟方案：
- 簡(jiǎn)單使用 StringRedisTemplate 構(gòu)建一個(gè)消費(fèi)隊(duì)列
- 使用 StringRedisTemplate 構(gòu)建多個(gè)消費(fèi)隊(duì)列
- 使用 RedissonClient 構(gòu)建多個(gè)消費(fèi)隊(duì)列，包含消費(fèi)隊(duì)列的運(yùn)維監(jiān)控、消費(fèi)隊(duì)列查詢/重置 等功能
一、Redis Stream簡(jiǎn)介與核心特性
Redis Stream是Redis 5.0版本專門為消息隊(duì)列場(chǎng)景設(shè)計(jì)的數(shù)據(jù)結(jié)構(gòu)，它借鑒了Kafka的設(shè)計(jì)理念，提供了消息持久化、消費(fèi)者組和消息確認(rèn)機(jī)制等核心功能。
Redis Stream核心優(yōu)勢(shì)
- 高性能：基于內(nèi)存操作，吞吐量高
- 持久化：消息可持久化保存，支持RDB和AOF
- 消費(fèi)者組：支持多消費(fèi)者負(fù)載均衡，確保消息不被重復(fù)消費(fèi)
- 阻塞操作：支持類似Kafka的長(zhǎng)輪詢機(jī)制
Redis Stream基礎(chǔ)操作示例
1. 添加消息到Stream
# 自動(dòng)生成消息ID XADD mystream * name "訂單創(chuàng)建" order_id "1001" amount "299.99" # 限制Stream最大長(zhǎng)度（保留最新1000條消息） XADD mystream MAXLEN 1000 * name "訂單支付" order_id "1001"
2. 查詢消息
# 查詢所有消息 XRANGE mystream - + # 分頁(yè)查詢，每次返回10條 XRANGE mystream - + COUNT 10 # 反向查詢（從新到舊） XREVRANGE mystream + - COUNT 5
3. 監(jiān)控新消息
# 阻塞監(jiān)聽新消息（0表示不超時(shí)） XREAD BLOCK 0 STREAMS mystream $
二、Spring Boot 簡(jiǎn)單集成案例
1. 添加依賴
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.18</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
</dependencies>2. 配置文件
# redis 配置
spring:
redis:
host: localhost
port: 6379
database: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
# redis stream 消費(fèi)組配置
app:
stream:
key: "order_stream"
group: "order_group"3. 消息生產(chǎn)者服務(wù)
package com.lijw.mp.event;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.UUID;
@Service
@Slf4j
public class MessageProducerService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Value("${app.stream.key}")
private String streamKey;
/**
* 發(fā)送消息到Redis Stream
*/
public String sendMessage(String messageType, Map<String, String> data) {
try {
// 添加消息類型和時(shí)間戳
data.put("messageType", messageType);
data.put("timestamp", String.valueOf(System.currentTimeMillis()));
data.put("messageId", UUID.randomUUID().toString());
RecordId messageId = stringRedisTemplate.opsForStream()
.add(streamKey, data);
log.info("消息發(fā)送成功: {}", messageId);
return messageId.toString();
} catch (Exception e) {
log.error("消息發(fā)送失敗: {}", e.getMessage());
throw new RuntimeException("消息發(fā)送失敗", e);
}
}
}4. 消息消費(fèi)者服務(wù)（支持冪等性）
package com.lijw.mp.event;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Map;
@Component
@Slf4j
public class MessageConsumerService implements StreamListener<String, MapRecord<String, String, String>> {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Value("${app.stream.group}")
private String groupName;
@Value("${app.stream.key}")
private String streamKey;
@Override
public void onMessage(MapRecord<String, String, String> message) {
String messageId = message.getId().toString();
Map<String, String> messageBody = message.getValue();
// 冪等性檢查：防止重復(fù)處理[1](@ref)
if (isMessageProcessed(messageId)) {
log.info("消息已處理，跳過(guò): {}", messageId);
acknowledgeMessage(message);
return;
}
try {
log.info("消費(fèi)者收到消息 - ID: {}, 內(nèi)容: {}", messageId, messageBody);
// 處理業(yè)務(wù)邏輯
boolean processSuccess = processBusiness(messageBody);
if (processSuccess) {
// 標(biāo)記消息已處理
markMessageProcessed(messageId);
// 清除重試次數(shù)
clearRetryCount(messageId);
// 手動(dòng)確認(rèn)消息
acknowledgeMessage(message);
log.info("消息處理完成: {}", messageId);
} else {
log.error("業(yè)務(wù)處理失敗，消息將重試: {}", messageId);
handleRetry(messageId, messageBody, message, "業(yè)務(wù)處理失敗");
}
} catch (Exception e) {
log.error("消息處理異常: {}", messageId, e);
handleRetry(messageId, messageBody, message, "消息處理異常");
}
}
/**
* 冪等性檢查
*/
private boolean isMessageProcessed(String messageId) {
// 使用Redis存儲(chǔ)，判斷redis是否已存在處理key，如果存在則說(shuō)明消息已處理，確保多實(shí)例間冪等性
return stringRedisTemplate.opsForValue().get("processed:" + messageId) != null;
}
/**
* 標(biāo)記消息已處理
*/
private void markMessageProcessed(String messageId) {
// 使用Redis存儲(chǔ)事件處理ID，設(shè)置過(guò)期時(shí)間
stringRedisTemplate.opsForValue().set("processed:" + messageId, "1",
Duration.ofHours(24));
}
/**
* 業(yè)務(wù)處理邏輯
*/
private boolean processBusiness(Map<String, String> messageBody) {
try {
String messageType = messageBody.get("messageType");
String orderId = messageBody.get("orderId");
log.info("處理{}消息，訂單ID: {}", messageType, orderId);
// 模擬業(yè)務(wù)處理
Thread.sleep(5000);
return true;
} catch (Exception e) {
log.error("業(yè)務(wù)處理異常", e);
return false;
}
}
/**
* 手動(dòng)確認(rèn)消息
*/
private void acknowledgeMessage(MapRecord<String, String, String> message) {
try {
stringRedisTemplate.opsForStream()
.acknowledge(groupName, message);
} catch (Exception e) {
log.error("消息確認(rèn)失敗: {}", message.getId(), e);
}
}
/**
* 增加重試次數(shù)
*/
private int incrementRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
String countStr = stringRedisTemplate.opsForValue().get(retryKey);
int retryCount = countStr == null ? 0 : Integer.parseInt(countStr);
retryCount++;
// 設(shè)置重試次數(shù)，過(guò)期時(shí)間為24小時(shí)
stringRedisTemplate.opsForValue().set(retryKey, String.valueOf(retryCount), Duration.ofHours(24));
return retryCount;
}
/**
* 清除重試次數(shù)
*/
private void clearRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
stringRedisTemplate.delete(retryKey);
}
/**
* 處理消息重試邏輯
*/
private void handleRetry(String messageId, Map<String, String> messageBody,
MapRecord<String, String, String> message, String errorDescription) {
// 記錄重試次數(shù)
int retryCount = incrementRetryCount(messageId);
// 檢查是否超過(guò)最大重試次數(shù)
int maxRetryCount = 3; // 最大重試次數(shù)，可根據(jù)業(yè)務(wù)需求調(diào)整
if (retryCount >= maxRetryCount) {
log.error("消息{}重試次數(shù)已達(dá)上限({})，將停止重試并記錄: {}", errorDescription, maxRetryCount, messageId);
// 記錄失敗消息到死信隊(duì)列或告警（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)）
handleMaxRetryExceeded(messageId, messageBody, retryCount);
// 確認(rèn)消息，避免無(wú)限重試
acknowledgeMessage(message);
} else {
log.warn("消息{}，當(dāng)前重試次數(shù): {}/{}, 消息ID: {}", errorDescription, retryCount, maxRetryCount, messageId);
// 不確認(rèn)消息，等待重試
}
}
/**
* 處理超過(guò)最大重試次數(shù)的消息
*/
private void handleMaxRetryExceeded(String messageId, Map<String, String> messageBody, int retryCount) {
try {
// 記錄失敗消息詳情（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)，如存儲(chǔ)到數(shù)據(jù)庫(kù)、發(fā)送告警等）
String failedKey = "failed:message:" + messageId;
String failedInfo = String.format("消息ID: %s, 重試次數(shù): %d, 消息內(nèi)容: %s, 失敗時(shí)間: %s",
messageId, retryCount, messageBody, System.currentTimeMillis());
stringRedisTemplate.opsForValue().set(failedKey, failedInfo, Duration.ofDays(7));
log.error("失敗消息已記錄: {}", failedInfo);
// TODO: 可根據(jù)業(yè)務(wù)需求添加其他處理邏輯，如：
// 1. 發(fā)送告警通知
// 2. 存儲(chǔ)到數(shù)據(jù)庫(kù)死信表
// 3. 發(fā)送到死信隊(duì)列
} catch (Exception e) {
log.error("處理超過(guò)最大重試次數(shù)消息異常: {}", messageId, e);
}
}
}5. 消費(fèi)者容器配置（支持多實(shí)例負(fù)載均衡）
package com.lijw.mp.config.redisstream;
import com.lijw.mp.event.MessageConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.net.InetAddress;
import java.time.Duration;
import java.lang.management.ManagementFactory;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
@Configuration
@Slf4j
public class RedisStreamConfig {
@Autowired
private RedisConnectionFactory redisConnectionFactory;
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Value("${app.stream.key}")
private String streamKey;
@Value("${app.stream.group}")
private String groupName;
/**
* 創(chuàng)建消費(fèi)者組（如果不存在）
*/
@EventListener(ApplicationReadyEvent.class)
public void createConsumerGroup() {
try {
stringRedisTemplate.opsForStream()
.createGroup(streamKey, groupName);
log.info("創(chuàng)建消費(fèi)者組成功: {}", groupName);
} catch (Exception e) {
log.info("消費(fèi)者組已存在: {}", e.getMessage());
}
}
/**
* 配置Stream消息監(jiān)聽容器
*/
@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
streamMessageListenerContainer(MessageConsumerService messageConsumerService) {
// 容器配置
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
StreamMessageListenerContainer.create(redisConnectionFactory,
StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
.pollTimeout(Duration.ofSeconds(2))
.batchSize(10) // 批量處理提高性能
.executor(createThreadPool()) // 自定義線程池
.build());
// 為每個(gè)實(shí)例生成唯一消費(fèi)者名稱
String consumerName = generateUniqueConsumerName();
// 配置消費(fèi)偏移量
StreamOffset<String> offset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());
// 創(chuàng)建消費(fèi)者
Consumer consumer = Consumer.from(groupName, consumerName);
// 構(gòu)建讀取請(qǐng)求（手動(dòng)確認(rèn)模式）
StreamMessageListenerContainer.StreamReadRequest<String> request =
StreamMessageListenerContainer.StreamReadRequest.builder(offset)
.consumer(consumer)
.autoAcknowledge(false) // 手動(dòng)確認(rèn)確?？煽啃訹2](@ref)
.cancelOnError(e -> false) // 錯(cuò)誤時(shí)不停止消費(fèi)
.build();
container.register(request, messageConsumerService);
container.start();
log.info("Redis Stream消費(fèi)者啟動(dòng)成功 - 消費(fèi)者名稱: {}", consumerName);
return container;
}
/**
* 生成唯一消費(fèi)者名稱（支持多實(shí)例部署的關(guān)鍵）
* 使用IP+進(jìn)程ID確保集群環(huán)境下唯一性
*/
private String generateUniqueConsumerName() {
try {
String hostAddress = InetAddress.getLocalHost().getHostAddress();
String processId = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
return hostAddress + "_" + processId + "_" + System.currentTimeMillis();
} catch (Exception e) {
// fallback：使用UUID
return "consumer_" + UUID.randomUUID().toString().substring(0, 8);
}
}
/**
* 創(chuàng)建專用線程池
*/
private ExecutorService createThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2);
executor.setMaxPoolSize(5);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("redis-stream-");
executor.setDaemon(true);
executor.initialize();
return executor.getThreadPoolExecutor();
}
}6. 待處理消息重試機(jī)制
package com.lijw.mp.config.redisstream;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.Duration;
@Component
@Slf4j
public class PendingMessageRetryService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Value("${app.stream.key}")
private String streamKey;
@Value("${app.stream.group}")
private String groupName;
/**
* 定時(shí)處理未確認(rèn)的消息
* 執(zhí)行時(shí)機(jī)：每30秒執(zhí)行一次（通過(guò)@Scheduled注解配置）
*/
@Scheduled(fixedDelay = 30000) // 每30秒執(zhí)行一次
public void retryPendingMessages() {
try {
// 獲取待處理消息摘要
PendingMessagesSummary pendingSummary = stringRedisTemplate.opsForStream()
.pending(streamKey, groupName);
if (pendingSummary != null) {
log.info("檢查待處理消息，消費(fèi)者組: {}", groupName);
// TODO: 根據(jù)Spring Data Redis版本，使用正確的API獲取詳細(xì)的pending消息列表
// 例如：使用pendingRange方法或其他方法獲取PendingMessages
// 獲取到PendingMessages后，調(diào)用processPendingMessages方法進(jìn)行處理
//
// 示例調(diào)用（需要根據(jù)實(shí)際API調(diào)整）：
// PendingMessages pendingMessages = stringRedisTemplate.opsForStream()
// .pendingRange(streamKey, groupName, ...);
// if (pendingMessages != null && pendingMessages.size() > 0) {
// processPendingMessages(pendingMessages);
// }
}
} catch (Exception e) {
log.error("處理待處理消息異常", e);
}
}
/**
* 處理待處理消息列表
*
* 執(zhí)行時(shí)機(jī)：
* 1. 由 retryPendingMessages() 定時(shí)任務(wù)調(diào)用（每30秒執(zhí)行一次）
* 2. 當(dāng) retryPendingMessages() 獲取到 PendingMessages 列表后調(diào)用
* 3. 用于處理Redis Stream中未被確認(rèn)（ACK）的消息
*
* 處理邏輯：
* - 遍歷每條pending消息
* - 記錄消息ID和消費(fèi)者名稱
* - 檢查消息空閑時(shí)間，如果超過(guò)閾值則重新分配
*/
private void processPendingMessages(PendingMessages pendingMessages) {
pendingMessages.forEach(message -> {
String messageId = message.getId().toString();
String consumerName = message.getConsumerName();
// 注意：根據(jù)Spring Data Redis版本，PendingMessage的API可能不同
// 獲取空閑時(shí)間的方法名可能是getIdleTimeMs()、getElapsedTimeMs()等
// 這里提供基礎(chǔ)框架，需要根據(jù)實(shí)際API調(diào)整
log.info("處理待處理消息: {}, 消費(fèi)者: {}", messageId, consumerName);
// 如果消息空閑時(shí)間超過(guò)閾值（如5分鐘），重新分配
// 示例邏輯（需要根據(jù)實(shí)際API調(diào)整）：
// Duration idleTime = Duration.ofMillis(message.getIdleTimeMs());
// if (idleTime.toMinutes() > 5) {
// log.info("重新分配超時(shí)消息: {}, 原消費(fèi)者: {}, 空閑時(shí)間: {}分鐘",
// messageId, consumerName, idleTime.toMinutes());
// // 可以使用XCLAIM命令將消息重新分配給其他消費(fèi)者
// // stringRedisTemplate.opsForStream().claim(...);
// }
});
}
}7. REST控制器
package com.lijw.mp.controller;
import com.lijw.mp.event.MessageProducerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.HashMap;
import java.util.Map;
@RestController
@RequestMapping("/api/message")
@Slf4j
public class MessageController {
@Autowired
private MessageProducerService messageProducerService;
@PostMapping("/send-order")
public ResponseEntity<Map<String, Object>> sendOrderMessage(
@RequestParam String orderId,
@RequestParam String amount) {
Map<String, String> message = new HashMap<>();
message.put("orderId", orderId);
message.put("amount", amount);
message.put("messageType", "ORDER_CREATED");
try {
String messageId = messageProducerService.sendMessage("ORDER", message);
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
@PostMapping("/send-custom")
public ResponseEntity<Map<String, Object>> sendCustomMessage(
@RequestBody Map<String, String> messageData) {
try {
String messageId = messageProducerService.sendMessage("CUSTOM", messageData);
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送自定義消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
}7.1 測(cè)試發(fā)送訂單消息
POST http://localhost:8083/api/message/send-order orderId=order123456 amount=500
7.2 測(cè)試發(fā)送自定義消息
POST http://localhost:8083/api/message/send-custom
{
"key1": "value1",
"key2": "valuie2"
}8.啟動(dòng)多個(gè)SpringBoot實(shí)例，測(cè)試實(shí)例是否會(huì)重復(fù)處理事件
8.1 通過(guò)修改端口號(hào)，啟動(dòng)多個(gè)實(shí)例
實(shí)例1
server: port: 8083
實(shí)例2
server: port: 8084
8.2 發(fā)送自定義事件消息
發(fā)送多條事件消息
8.3 查看實(shí)例日志，確認(rèn)事件未被重復(fù)消費(fèi)
實(shí)例1
實(shí)例2
可以看到 1764271408044-0 只在實(shí)例2處理，并沒有在實(shí)例1處理。說(shuō)明多個(gè)實(shí)例并不會(huì)重復(fù)消費(fèi)同一事件。
三、SpringBoot 使用 StringRedisTemplate 集成 Redis Stream 進(jìn)階：配置多個(gè)消費(fèi)組
1.配置文件
配置多個(gè)消費(fèi)組key
# redis 配置
spring:
redis:
host: localhost
port: 6379
database: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
# redis stream 配置多個(gè)消費(fèi)組key
app:
stream:
groups:
- key: "order_stream"
group: "order_group"
consumer-prefix: "consumer_"
- key: "payment_stream"
group: "payment_group"
consumer-prefix: "consumer_"
- key: "notification_stream"
group: "notification_group"
consumer-prefix: "consumer_"2.Redis Stream消費(fèi)組配置屬性
package com.lijw.mp.config.redisstream;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* Redis Stream消費(fèi)組配置屬性
* - 讀取 app.stream 配置前綴下的所有配置項(xiàng)
*
* @author Aron.li
* @date 2025/11/30 11:21
*/
@Data
@Component
@ConfigurationProperties(prefix = "app.stream")
public class StreamGroupProperties {
/**
* 消費(fèi)組列表
*/
private List<StreamGroupConfig> groups;
/**
* 單個(gè)消費(fèi)組配置
*/
@Data
public static class StreamGroupConfig {
/**
* Stream鍵名
*/
private String key;
/**
* 消費(fèi)組名稱
*/
private String group;
/**
* 消費(fèi)者名稱前綴
*/
private String consumerPrefix;
}
}3. 消息生產(chǎn)者服務(wù)
package com.lijw.mp.event;
import com.lijw.mp.config.redisstream.StreamGroupProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.UUID;
/**
* Redis Stream消息生產(chǎn)者服務(wù)
* 負(fù)責(zé)向Redis Stream發(fā)送消息
*/
@Service
@Slf4j
public class MessageProducerService {
/**
* Redis字符串模板
*/
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 發(fā)送消息到Redis Stream（根據(jù)消息類型自動(dòng)選擇stream）
*
* @param messageType 消息類型
* @param data 消息數(shù)據(jù)
* @return 消息ID
*/
public String sendMessage(String messageType, Map<String, String> data) {
// 根據(jù)消息類型自動(dòng)選擇stream key
String streamKey = getStreamKeyByMessageType(messageType);
return sendMessage(streamKey, messageType, data);
}
/**
* 發(fā)送消息到指定的Redis Stream
*
* @param streamKey Stream鍵名
* @param messageType 消息類型
* @param data 消息數(shù)據(jù)
* @return 消息ID
*/
public String sendMessage(String streamKey, String messageType, Map<String, String> data) {
try {
// 添加消息類型和時(shí)間戳
data.put("messageType", messageType);
data.put("timestamp", String.valueOf(System.currentTimeMillis()));
data.put("messageId", UUID.randomUUID().toString());
RecordId messageId = stringRedisTemplate.opsForStream()
.add(streamKey, data);
log.info("消息發(fā)送成功 - Stream: {}, MessageId: {}", streamKey, messageId);
return messageId.toString();
} catch (Exception e) {
log.error("消息發(fā)送失敗 - Stream: {}", streamKey, e);
throw new RuntimeException("消息發(fā)送失敗", e);
}
}
/**
* 根據(jù)消息類型獲取對(duì)應(yīng)的stream key
*
* @param messageType 消息類型
* @return Stream鍵名
*/
private String getStreamKeyByMessageType(String messageType) {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
throw new RuntimeException("未配置任何消費(fèi)組");
}
// 根據(jù)消息類型前綴匹配stream key
String upperMessageType = messageType.toUpperCase();
if (upperMessageType.contains("ORDER")) {
return findStreamKey("order");
} else if (upperMessageType.contains("PAYMENT") || upperMessageType.contains("PAY")) {
return findStreamKey("payment");
} else if (upperMessageType.contains("NOTIFICATION") || upperMessageType.contains("NOTIFY")) {
return findStreamKey("notification");
} else {
// 默認(rèn)使用第一個(gè)stream
log.warn("未匹配到消息類型對(duì)應(yīng)的stream，使用默認(rèn)stream: {}", messageType);
return streamGroupProperties.getGroups().get(0).getKey();
}
}
/**
* 根據(jù)關(guān)鍵字查找stream key
*
* @param keyword 關(guān)鍵字
* @return Stream鍵名
*/
private String findStreamKey(String keyword) {
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
if (groupConfig.getKey().toLowerCase().contains(keyword.toLowerCase())) {
return groupConfig.getKey();
}
}
}
// 如果找不到，返回第一個(gè)stream
if (streamGroupProperties.getGroups() != null && !streamGroupProperties.getGroups().isEmpty()) {
return streamGroupProperties.getGroups().get(0).getKey();
}
throw new RuntimeException("未找到匹配的stream key: " + keyword);
}
}4. 消息消費(fèi)者服務(wù)（支持冪等性）
package com.lijw.mp.event;
import com.lijw.mp.config.redisstream.StreamGroupProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Redis Stream消息消費(fèi)者服務(wù)
* 負(fù)責(zé)消費(fèi)Redis Stream中的消息并進(jìn)行業(yè)務(wù)處理
*/
@Component
@Slf4j
public class MessageConsumerService implements StreamListener<String, MapRecord<String, String, String>> {
/**
* Redis字符串模板
*/
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 緩存stream key到group name的映射關(guān)系
*/
private final Map<String, String> streamToGroupMap = new ConcurrentHashMap<>();
/**
* 消息消費(fèi)處理方法
* 當(dāng)Redis Stream中有新消息時(shí)，會(huì)調(diào)用此方法
*
* @param message Stream消息記錄
*/
@Override
public void onMessage(MapRecord<String, String, String> message) {
String messageId = message.getId().toString();
String streamKey = message.getStream();
Map<String, String> messageBody = message.getValue();
// 獲取對(duì)應(yīng)的消費(fèi)組名稱
String groupName = getGroupNameByStreamKey(streamKey);
if (groupName == null) {
log.error("未找到stream key對(duì)應(yīng)的消費(fèi)組: {}, 消息ID: {}", streamKey, messageId);
return;
}
// 冪等性檢查：防止重復(fù)處理
if (isMessageProcessed(messageId)) {
log.info("消息已處理，跳過(guò): {}", messageId);
acknowledgeMessage(message, groupName);
return;
}
try {
log.info("消費(fèi)者收到消息 - Stream: {}, Group: {}, ID: {}, 內(nèi)容: {}", streamKey, groupName, messageId, messageBody);
// 處理業(yè)務(wù)邏輯（根據(jù)stream key路由到不同的業(yè)務(wù)處理）
boolean processSuccess = processBusiness(streamKey, messageBody);
if (processSuccess) {
// 標(biāo)記消息已處理
markMessageProcessed(messageId);
// 清除重試次數(shù)
clearRetryCount(messageId);
// 手動(dòng)確認(rèn)消息
acknowledgeMessage(message, groupName);
log.info("消息處理完成: {}", messageId);
} else {
log.error("業(yè)務(wù)處理失敗，消息將重試: {}", messageId);
handleRetry(messageId, messageBody, message, groupName, "業(yè)務(wù)處理失敗");
}
} catch (Exception e) {
log.error("消息處理異常: {}", messageId, e);
handleRetry(messageId, messageBody, message, groupName, "消息處理異常");
}
}
/**
* 根據(jù)stream key獲取對(duì)應(yīng)的消費(fèi)組名稱
*
* @param streamKey Stream鍵名
* @return 消費(fèi)組名稱，如果未找到返回null
*/
private String getGroupNameByStreamKey(String streamKey) {
// 先從緩存中獲取
if (streamToGroupMap.containsKey(streamKey)) {
return streamToGroupMap.get(streamKey);
}
// 從配置中查找
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
if (groupConfig.getKey().equals(streamKey)) {
streamToGroupMap.put(streamKey, groupConfig.getGroup());
return groupConfig.getGroup();
}
}
}
return null;
}
/**
* 冪等性檢查
* 使用Redis存儲(chǔ)，判斷redis是否已存在處理key，如果存在則說(shuō)明消息已處理，確保多實(shí)例間冪等性
*
* @param messageId 消息ID
* @return 如果消息已處理返回true，否則返回false
*/
private boolean isMessageProcessed(String messageId) {
// 使用Redis存儲(chǔ)，判斷redis是否已存在處理key，如果存在則說(shuō)明消息已處理，確保多實(shí)例間冪等性
return stringRedisTemplate.opsForValue().get("processed:" + messageId) != null;
}
/**
* 標(biāo)記消息已處理
* 使用Redis存儲(chǔ)事件處理ID，設(shè)置過(guò)期時(shí)間
*
* @param messageId 消息ID
*/
private void markMessageProcessed(String messageId) {
// 使用Redis存儲(chǔ)事件處理ID，設(shè)置過(guò)期時(shí)間
stringRedisTemplate.opsForValue().set("processed:" + messageId, "1",
Duration.ofHours(24));
}
/**
* 業(yè)務(wù)處理邏輯（根據(jù)stream key路由到不同的業(yè)務(wù)處理）
*/
private boolean processBusiness(String streamKey, Map<String, String> messageBody) {
try {
// 根據(jù)stream key路由到不同的業(yè)務(wù)處理
if (streamKey.contains("order")) {
return processOrderBusiness(messageBody);
} else if (streamKey.contains("payment")) {
return processPaymentBusiness(messageBody);
} else if (streamKey.contains("notification")) {
return processNotificationBusiness(messageBody);
} else {
log.warn("未識(shí)別的stream key: {}, 使用默認(rèn)處理邏輯", streamKey);
return processDefaultBusiness(messageBody);
}
} catch (Exception e) {
log.error("業(yè)務(wù)處理異常", e);
return false;
}
}
/**
* 處理訂單業(yè)務(wù)
*/
private boolean processOrderBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String orderId = messageBody.get("orderId");
log.info("處理訂單{}消息，訂單ID: {}", messageType, orderId);
// TODO: 實(shí)現(xiàn)訂單業(yè)務(wù)處理邏輯
try {
// 模擬10秒處理業(yè)務(wù)
Thread.sleep(10000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return true;
}
/**
* 處理支付業(yè)務(wù)
*/
private boolean processPaymentBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String paymentId = messageBody.get("paymentId");
log.info("處理支付{}消息，支付ID: {}", messageType, paymentId);
// TODO: 實(shí)現(xiàn)支付業(yè)務(wù)處理邏輯
return true;
}
/**
* 處理通知業(yè)務(wù)
*/
private boolean processNotificationBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String userId = messageBody.get("userId");
log.info("處理通知{}消息，用戶ID: {}", messageType, userId);
// TODO: 實(shí)現(xiàn)通知業(yè)務(wù)處理邏輯
return true;
}
/**
* 默認(rèn)業(yè)務(wù)處理
*/
private boolean processDefaultBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
log.info("處理默認(rèn)業(yè)務(wù)消息: {}", messageType);
// TODO: 實(shí)現(xiàn)默認(rèn)業(yè)務(wù)處理邏輯
return true;
}
/**
* 手動(dòng)確認(rèn)消息
*
* @param message 消息記錄
* @param groupName 消費(fèi)組名稱
*/
private void acknowledgeMessage(MapRecord<String, String, String> message, String groupName) {
try {
stringRedisTemplate.opsForStream()
.acknowledge(groupName, message);
} catch (Exception e) {
log.error("消息確認(rèn)失敗: {}", message.getId(), e);
}
}
/**
* 增加重試次數(shù)
*
* @param messageId 消息ID
* @return 當(dāng)前重試次數(shù)
*/
private int incrementRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
String countStr = stringRedisTemplate.opsForValue().get(retryKey);
int retryCount = countStr == null ? 0 : Integer.parseInt(countStr);
retryCount++;
// 設(shè)置重試次數(shù)，過(guò)期時(shí)間為24小時(shí)
stringRedisTemplate.opsForValue().set(retryKey, String.valueOf(retryCount), Duration.ofHours(24));
return retryCount;
}
/**
* 清除重試次數(shù)
*
* @param messageId 消息ID
*/
private void clearRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
stringRedisTemplate.delete(retryKey);
}
/**
* 處理消息重試邏輯
*
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param message 消息記錄
* @param groupName 消費(fèi)組名稱
* @param errorDescription 錯(cuò)誤描述
*/
private void handleRetry(String messageId, Map<String, String> messageBody,
MapRecord<String, String, String> message, String groupName, String errorDescription) {
// 記錄重試次數(shù)
int retryCount = incrementRetryCount(messageId);
// 檢查是否超過(guò)最大重試次數(shù)
int maxRetryCount = 3; // 最大重試次數(shù)，可根據(jù)業(yè)務(wù)需求調(diào)整
if (retryCount >= maxRetryCount) {
log.error("消息{}重試次數(shù)已達(dá)上限({})，將停止重試并記錄: {}", errorDescription, maxRetryCount, messageId);
// 記錄失敗消息到死信隊(duì)列或告警（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)）
handleMaxRetryExceeded(messageId, messageBody, retryCount);
// 確認(rèn)消息，避免無(wú)限重試
acknowledgeMessage(message, groupName);
} else {
log.warn("消息{}，當(dāng)前重試次數(shù): {}/{}, 消息ID: {}", errorDescription, retryCount, maxRetryCount, messageId);
// 不確認(rèn)消息，等待重試
}
}
/**
* 處理超過(guò)最大重試次數(shù)的消息
* 記錄失敗消息詳情，可根據(jù)業(yè)務(wù)需求擴(kuò)展（如存儲(chǔ)到數(shù)據(jù)庫(kù)、發(fā)送告警等）
*
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param retryCount 重試次數(shù)
*/
private void handleMaxRetryExceeded(String messageId, Map<String, String> messageBody, int retryCount) {
try {
// 記錄失敗消息詳情（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)，如存儲(chǔ)到數(shù)據(jù)庫(kù)、發(fā)送告警等）
String failedKey = "failed:message:" + messageId;
String failedInfo = String.format("消息ID: %s, 重試次數(shù): %d, 消息內(nèi)容: %s, 失敗時(shí)間: %s",
messageId, retryCount, messageBody, System.currentTimeMillis());
stringRedisTemplate.opsForValue().set(failedKey, failedInfo, Duration.ofDays(7));
log.error("失敗消息已記錄: {}", failedInfo);
// TODO: 可根據(jù)業(yè)務(wù)需求添加其他處理邏輯，如：
// 1. 發(fā)送告警通知
// 2. 存儲(chǔ)到數(shù)據(jù)庫(kù)死信表
// 3. 發(fā)送到死信隊(duì)列
} catch (Exception e) {
log.error("處理超過(guò)最大重試次數(shù)消息異常: {}", messageId, e);
}
}
}5.消費(fèi)者容器配置（支持多實(shí)例負(fù)載均衡）
package com.lijw.mp.config.redisstream;
import com.lijw.mp.event.MessageConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.net.InetAddress;
import java.time.Duration;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
/**
* Redis Stream配置類
* 負(fù)責(zé)創(chuàng)建消費(fèi)組、配置消息監(jiān)聽容器等
*/
@Configuration
@Slf4j
public class RedisStreamConfig {
/**
* Redis連接工廠
*/
@Autowired
private RedisConnectionFactory redisConnectionFactory;
/**
* Redis字符串模板
*/
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 應(yīng)用啟動(dòng)完成后，驗(yàn)證所有消費(fèi)組是否已創(chuàng)建
* 主要用于日志記錄和驗(yàn)證
*/
@EventListener(ApplicationReadyEvent.class)
public void verifyConsumerGroups() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組");
return;
}
log.info("驗(yàn)證所有消費(fèi)組狀態(tài)...");
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
// 驗(yàn)證消費(fèi)組是否存在
stringRedisTemplate.opsForStream().pending(groupConfig.getKey(), groupConfig.getGroup());
log.info("消費(fèi)組驗(yàn)證成功: stream={}, group={}", groupConfig.getKey(), groupConfig.getGroup());
} catch (Exception e) {
log.warn("消費(fèi)組驗(yàn)證失敗，嘗試重新創(chuàng)建: stream={}, group={}",
groupConfig.getKey(), groupConfig.getGroup());
// 嘗試重新創(chuàng)建
try {
ensureStreamExists(groupConfig.getKey());
stringRedisTemplate.opsForStream()
.createGroup(groupConfig.getKey(), groupConfig.getGroup());
log.info("重新創(chuàng)建消費(fèi)者組成功: stream={}, group={}",
groupConfig.getKey(), groupConfig.getGroup());
} catch (Exception ex) {
log.error("重新創(chuàng)建消費(fèi)者組失敗: stream={}, group={}, error={}",
groupConfig.getKey(), groupConfig.getGroup(), ex.getMessage());
}
}
}
}
/**
* 確保stream存在，如果不存在則創(chuàng)建
*
* @param streamKey Stream鍵名
*/
private void ensureStreamExists(String streamKey) {
try {
// 檢查stream是否存在
Long size = stringRedisTemplate.opsForStream().size(streamKey);
if (size == null || size == 0) {
// Stream不存在，創(chuàng)建一個(gè)空消息然后立即刪除，以創(chuàng)建stream
// 或者直接使用XGROUP CREATE的MKSTREAM選項(xiàng)
// 由于Spring Data Redis可能不支持MKSTREAM，我們通過(guò)添加一條臨時(shí)消息來(lái)創(chuàng)建stream
Map<String, String> tempData = new HashMap<>();
tempData.put("_init", "true");
tempData.put("_timestamp", String.valueOf(System.currentTimeMillis()));
try {
stringRedisTemplate.opsForStream().add(streamKey, tempData);
log.debug("創(chuàng)建stream成功: {}", streamKey);
} catch (Exception ex) {
log.debug("Stream可能已存在或創(chuàng)建失敗: {}, {}", streamKey, ex.getMessage());
}
}
} catch (Exception e) {
// Stream不存在，創(chuàng)建它
try {
Map<String, String> tempData = new HashMap<>();
tempData.put("_init", "true");
tempData.put("_timestamp", String.valueOf(System.currentTimeMillis()));
stringRedisTemplate.opsForStream().add(streamKey, tempData);
log.info("創(chuàng)建stream成功: {}", streamKey);
} catch (Exception ex) {
log.warn("創(chuàng)建stream失敗: {}, {}", streamKey, ex.getMessage());
}
}
}
/**
* 為每個(gè)消費(fèi)組配置Stream消息監(jiān)聽容器
* 在創(chuàng)建監(jiān)聽容器之前，先確保所有消費(fèi)組都已創(chuàng)建
*/
@Bean
public List<StreamMessageListenerContainer<String, MapRecord<String, String, String>>>
streamMessageListenerContainers(MessageConsumerService messageConsumerService) {
List<StreamMessageListenerContainer<String, MapRecord<String, String, String>>> containers = new ArrayList<>();
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組，無(wú)法創(chuàng)建監(jiān)聽容器");
return containers;
}
// 在創(chuàng)建監(jiān)聽容器之前，先確保所有消費(fèi)組都已創(chuàng)建
createAllConsumerGroups();
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
createListenerContainer(groupConfig, messageConsumerService);
containers.add(container);
}
return containers;
}
/**
* 創(chuàng)建所有消費(fèi)者組（如果不存在）
* 如果stream不存在，會(huì)先創(chuàng)建stream
*/
private void createAllConsumerGroups() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組");
return;
}
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
// 先檢查stream是否存在，如果不存在則創(chuàng)建
ensureStreamExists(groupConfig.getKey());
// 創(chuàng)建消費(fèi)組
stringRedisTemplate.opsForStream()
.createGroup(groupConfig.getKey(), groupConfig.getGroup());
log.info("創(chuàng)建消費(fèi)者組成功: stream={}, group={}", groupConfig.getKey(), groupConfig.getGroup());
} catch (Exception e) {
// 如果消費(fèi)組已存在，這是正常情況
String errorMsg = e.getMessage() != null ? e.getMessage() : "";
if (errorMsg.contains("BUSYGROUP") || errorMsg.contains("already exists")) {
log.info("消費(fèi)者組已存在: stream={}, group={}", groupConfig.getKey(), groupConfig.getGroup());
} else {
log.warn("創(chuàng)建消費(fèi)者組失敗: stream={}, group={}, error={}",
groupConfig.getKey(), groupConfig.getGroup(), errorMsg);
// 即使創(chuàng)建失敗，也繼續(xù)處理其他消費(fèi)組
}
}
}
}
/**
* 為單個(gè)消費(fèi)組創(chuàng)建監(jiān)聽容器
*
* @param groupConfig 消費(fèi)組配置
* @param messageConsumerService 消息消費(fèi)者服務(wù)
* @return Stream消息監(jiān)聽容器
*/
private StreamMessageListenerContainer<String, MapRecord<String, String, String>>
createListenerContainer(StreamGroupProperties.StreamGroupConfig groupConfig,
MessageConsumerService messageConsumerService) {
// 容器配置
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
StreamMessageListenerContainer.create(redisConnectionFactory,
StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
.pollTimeout(Duration.ofSeconds(2)) // 輪詢超時(shí)時(shí)間
.batchSize(10) // 批量處理提高性能
.executor(createThreadPool()) // 自定義線程池
.build());
// 為每個(gè)實(shí)例生成唯一消費(fèi)者名稱
String consumerName = generateUniqueConsumerName(groupConfig.getConsumerPrefix());
// 配置消費(fèi)偏移量（從最后消費(fèi)的位置開始）
StreamOffset<String> offset = StreamOffset.create(groupConfig.getKey(), ReadOffset.lastConsumed());
// 創(chuàng)建消費(fèi)者
Consumer consumer = Consumer.from(groupConfig.getGroup(), consumerName);
// 構(gòu)建讀取請(qǐng)求（手動(dòng)確認(rèn)模式）
StreamMessageListenerContainer.StreamReadRequest<String> request =
StreamMessageListenerContainer.StreamReadRequest.builder(offset)
.consumer(consumer)
.autoAcknowledge(false) // 手動(dòng)確認(rèn)確保可靠性
.cancelOnError(e -> false) // 錯(cuò)誤時(shí)不停止消費(fèi)
.build();
// 注冊(cè)監(jiān)聽器并啟動(dòng)容器
container.register(request, messageConsumerService);
container.start();
log.info("Redis Stream消費(fèi)者啟動(dòng)成功 - Stream: {}, Group: {}, Consumer: {}",
groupConfig.getKey(), groupConfig.getGroup(), consumerName);
return container;
}
/**
* 生成唯一消費(fèi)者名稱（支持多實(shí)例部署的關(guān)鍵）
* 使用IP+進(jìn)程ID確保集群環(huán)境下唯一性
*
* @param prefix 消費(fèi)者名稱前綴
* @return 唯一的消費(fèi)者名稱
*/
private String generateUniqueConsumerName(String prefix) {
try {
String hostAddress = InetAddress.getLocalHost().getHostAddress();
String processId = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
return (prefix != null ? prefix : "consumer_") + hostAddress + "_" + processId + "_" + System.currentTimeMillis();
} catch (Exception e) {
// fallback：使用UUID
return (prefix != null ? prefix : "consumer_") + UUID.randomUUID().toString().substring(0, 8);
}
}
/**
* 創(chuàng)建專用線程池
* 用于處理Redis Stream消息消費(fèi)
*
* @return 線程池執(zhí)行器
*/
private ExecutorService createThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2); // 核心線程數(shù)
executor.setMaxPoolSize(5); // 最大線程數(shù)
executor.setQueueCapacity(100); // 隊(duì)列容量
executor.setThreadNamePrefix("redis-stream-"); // 線程名前綴
executor.setDaemon(true); // 守護(hù)線程
executor.initialize();
return executor.getThreadPoolExecutor();
}
}6.處理消息告警機(jī)制
當(dāng)消息堵塞達(dá)到一定，則出發(fā)告警，提醒需要擴(kuò)容實(shí)例服務(wù)
package com.lijw.mp.config.redisstream;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* 待處理消息告警服務(wù)
* 定時(shí)檢查Redis Stream中未被確認(rèn)（ACK）的消息，僅用于監(jiān)控和告警
* 注意：不進(jìn)行實(shí)際的消息處理，通過(guò)增加消費(fèi)實(shí)例來(lái)提升處理能力
*/
@Component
@Slf4j
public class PendingMessageAlertService {
/**
* Redis字符串模板
*/
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 定時(shí)監(jiān)控未確認(rèn)的消息
* 執(zhí)行時(shí)機(jī)：每30秒執(zhí)行一次（通過(guò)@Scheduled注解配置）
* 僅用于監(jiān)控和告警，不進(jìn)行實(shí)際的消息處理
* 建議通過(guò)增加消費(fèi)實(shí)例來(lái)提升處理能力
*/
@Scheduled(fixedDelay = 30000) // 每30秒執(zhí)行一次
public void monitorPendingMessages() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
return;
}
// 遍歷所有配置的消費(fèi)組
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
// 獲取待處理消息摘要
PendingMessagesSummary pendingSummary = stringRedisTemplate.opsForStream()
.pending(groupConfig.getKey(), groupConfig.getGroup());
if (pendingSummary != null && pendingSummary.getTotalPendingMessages() > 0) {
// 記錄監(jiān)控信息
logPendingMessageInfo(pendingSummary, groupConfig);
}
} catch (Exception e) {
log.error("監(jiān)控待處理消息異常 - Stream: {}, Group: {}",
groupConfig.getKey(), groupConfig.getGroup(), e);
}
}
}
/**
* 記錄pending消息監(jiān)控信息
* 根據(jù)pending消息數(shù)量進(jìn)行不同級(jí)別的告警
*
* @param pendingSummary 待處理消息摘要
* @param groupConfig 消費(fèi)組配置
*/
private void logPendingMessageInfo(PendingMessagesSummary pendingSummary,
StreamGroupProperties.StreamGroupConfig groupConfig) {
long totalPending = pendingSummary.getTotalPendingMessages();
// 根據(jù)pending消息數(shù)量進(jìn)行分級(jí)告警
if (totalPending > 2) {
// 嚴(yán)重告警：pending消息數(shù)量超過(guò)1000
log.error("【嚴(yán)重告警】待處理消息數(shù)量過(guò)多 - Stream: {}, Group: {}, 數(shù)量: {}, " +
"建議：1.檢查消費(fèi)者是否正常運(yùn)行 2.增加消費(fèi)實(shí)例 3.檢查消息處理邏輯",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
} else if (totalPending > 1) {
// 警告：pending消息數(shù)量超過(guò)100
log.warn("【警告】待處理消息數(shù)量較多 - Stream: {}, Group: {}, 數(shù)量: {}, " +
"建議：考慮增加消費(fèi)實(shí)例或檢查消息處理性能",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
} else {
// 信息：pending消息數(shù)量正常
log.info("監(jiān)控待處理消息 - Stream: {}, Group: {}, 數(shù)量: {}",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
}
}
}7.REST控制器
package com.lijw.mp.controller;
import com.lijw.mp.event.MessageProducerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.HashMap;
import java.util.Map;
/**
* 測(cè)試發(fā)送redis stream事件消息
*/
@RestController
@RequestMapping("/api/message")
@Slf4j
public class MessageController {
@Autowired
private MessageProducerService messageProducerService;
/**
* 發(fā)送訂單消息
*/
@PostMapping("/send-order")
public ResponseEntity<Map<String, Object>> sendOrderMessage(
@RequestParam String orderId,
@RequestParam String amount,
@RequestParam(required = false) String streamKey) {
Map<String, String> message = new HashMap<>();
message.put("orderId", orderId);
message.put("amount", amount);
message.put("messageType", "ORDER_CREATED");
try {
String messageId;
if (streamKey != null && !streamKey.isEmpty()) {
// 指定stream key發(fā)送
messageId = messageProducerService.sendMessage(streamKey, "ORDER", message);
} else {
// 自動(dòng)根據(jù)消息類型路由
messageId = messageProducerService.sendMessage("ORDER", message);
}
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
result.put("streamKey", streamKey != null ? streamKey : "auto-routed");
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送訂單消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
/**
* 發(fā)送支付消息
*/
@PostMapping("/send-payment")
public ResponseEntity<Map<String, Object>> sendPaymentMessage(
@RequestParam String paymentId,
@RequestParam String amount,
@RequestParam String status,
@RequestParam(required = false) String streamKey) {
Map<String, String> message = new HashMap<>();
message.put("paymentId", paymentId);
message.put("amount", amount);
message.put("status", status);
message.put("messageType", "PAYMENT_" + status.toUpperCase());
try {
String messageId;
if (streamKey != null && !streamKey.isEmpty()) {
messageId = messageProducerService.sendMessage(streamKey, "PAYMENT", message);
} else {
messageId = messageProducerService.sendMessage("PAYMENT", message);
}
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
result.put("streamKey", streamKey != null ? streamKey : "auto-routed");
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送支付消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
/**
* 發(fā)送通知消息
*/
@PostMapping("/send-notification")
public ResponseEntity<Map<String, Object>> sendNotificationMessage(
@RequestParam String userId,
@RequestParam String title,
@RequestParam String content,
@RequestParam(required = false) String streamKey) {
Map<String, String> message = new HashMap<>();
message.put("userId", userId);
message.put("title", title);
message.put("content", content);
message.put("messageType", "NOTIFICATION");
try {
String messageId;
if (streamKey != null && !streamKey.isEmpty()) {
messageId = messageProducerService.sendMessage(streamKey, "NOTIFICATION", message);
} else {
messageId = messageProducerService.sendMessage("NOTIFICATION", message);
}
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
result.put("streamKey", streamKey != null ? streamKey : "auto-routed");
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送通知消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
/**
* 發(fā)送自定義消息（支持指定stream key和消息類型）
*/
@PostMapping("/send-custom")
public ResponseEntity<Map<String, Object>> sendCustomMessage(
@RequestBody Map<String, String> messageData,
@RequestParam(required = false) String streamKey,
@RequestParam(required = false) String messageType) {
try {
// 如果未指定消息類型，從消息數(shù)據(jù)中獲取或使用默認(rèn)值
String msgType = messageType != null ? messageType :
messageData.getOrDefault("messageType", "CUSTOM");
String messageId;
if (streamKey != null && !streamKey.isEmpty()) {
// 指定stream key發(fā)送
messageId = messageProducerService.sendMessage(streamKey, msgType, messageData);
} else {
// 自動(dòng)根據(jù)消息類型路由
messageId = messageProducerService.sendMessage(msgType, messageData);
}
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId);
result.put("streamKey", streamKey != null ? streamKey : "auto-routed");
result.put("messageType", msgType);
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送自定義消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(500).body(result);
}
}
}7.1 測(cè)試發(fā)送訂單消息
POST http://localhost:8083/api/message/send-order # 請(qǐng)求參數(shù) streamKey:order_stream orderId:order123456 amount:500 status:ok
7.2 測(cè)試發(fā)送支付消息
POST http://localhost:8083/api/message/send-payment # 請(qǐng)求參數(shù) streamKey:payment_stream paymentId:payId123456 amount:500 status:ok
7.3 測(cè)試發(fā)送通知消息
POST http://localhost:8083/api/message/send-notification # 請(qǐng)求參數(shù) streamKey:notification_stream userId:userId123456 title:通知消息 content:通知內(nèi)容
7.4 測(cè)試發(fā)送通知消息
POST http://localhost:8083/api/message/send-custom
# 請(qǐng)求參數(shù)
streamKey:custom_stream
# 請(qǐng)求體
{
"key1": "value1",
"key2": "valuie2"
}8.啟動(dòng)多個(gè)SpringBoot實(shí)例，測(cè)試實(shí)例是否會(huì)重復(fù)處理事件
跟案例1的操作一致，以發(fā)送訂單消息的事件作為驗(yàn)證示例
8.1 通過(guò)修改端口號(hào)，啟動(dòng)多個(gè)實(shí)例
實(shí)例1
server: port: 8083
實(shí)例2
server: port: 8084
8.2 發(fā)送訂單事件消息，查看實(shí)例日志，確認(rèn)事件未被重復(fù)消費(fèi)
實(shí)例1
實(shí)例2
可以看到只有一個(gè)實(shí)例處理 1764865303698-0 事件，并沒有被重復(fù)消費(fèi)
9.觸發(fā)多個(gè)消息，造成堵塞，檢查告警日志
模擬大量出現(xiàn)訂閱消息，導(dǎo)致事件堵塞的情況。
10. Redis 查看隊(duì)列消息
10.1 查看當(dāng)前redis的所有keys
# 查看當(dāng)前redis的所有keys 172.17.0.6:6379> KEYS * 1) "payment_stream" 2) "mystream" 3) "notification_stream" 4) "custom_stream" 5) "order_stream" 172.17.0.6:6379>
10.2 查詢所有order隊(duì)列消息
XRANGE order_stream - +
# 查詢消息如下：
140) 1) "1764898453817-0"
2) 1) "amount"
2) "500"
3) "messageType"
4) "ORDER"
5) "orderId"
6) "order123456"
7) "messageId"
8) "98e48bde-88df-46c1-a0a1-ac917ac14fa6"
9) "timestamp"
10) "1764898453753"
141) 1) "1764898468568-0"
2) 1) "amount"
2) "500"
3) "messageType"
4) "ORDER"
5) "orderId"
6) "order123456"
7) "messageId"
8) "93978d1c-d18e-4760-8657-8cf0042906a2"
9) "timestamp"
10) "1764898468507"
172.17.0.6:6379> 10.3 分頁(yè)查詢，每次返回3條
# 分頁(yè)查詢，每次返回3條
XRANGE order_stream - + COUNT 3
# 執(zhí)行如下：
172.17.0.6:6379> XRANGE order_stream - + COUNT 3
1) 1) "1764268213474-0"
2) 1) "source"
2) "TEST_CONTROLLER"
3) "type"
4) "TEST_MESSAGE"
5) "content"
6) "test msg"
7) "timestamp"
8) "1764268202906"
2) 1) "1764268564758-0"
2) 1) "amount"
2) "500"
3) "messageType"
4) "ORDER"
5) "orderId"
6) "order123456"
7) "messageId"
8) "eb1a1fc1-12a9-4a80-84f4-f219a05094b5"
9) "timestamp"
10) "1764268564695"
3) 1) "1764268699963-0"
2) 1) "amount"
2) "500"
3) "messageType"
4) "ORDER"
5) "orderId"
6) "order123456"
7) "messageId"
8) "419c1c9a-7bcb-4f4e-b08c-70e86c5e7387"
9) "timestamp"
10) "1764268693593"
172.17.0.6:6379> 10.4 反向查詢（從新到舊）
# 反向查詢（從新到舊） XREVRANGE order_stream + - COUNT 3
10.5 查詢隊(duì)列總數(shù)
172.17.0.6:6379> XLEN order_stream (integer) 141 172.17.0.6:6379>
11. 總結(jié)
上面的項(xiàng)目是通過(guò) StringRedisTemplate 做隊(duì)列消息的讀取，有些時(shí)候項(xiàng)目不能直接使用。
@Autowired private StringRedisTemplate stringRedisTemplate;
所以下一章節(jié)，我將會(huì)使用 RedissonClient 來(lái)做替代實(shí)現(xiàn)的方法，增加符合生產(chǎn)環(huán)境使用，并且還會(huì)增加消息隊(duì)列總數(shù)的監(jiān)控、查詢消息隊(duì)列總數(shù)接口、清除消息隊(duì)列的運(yùn)維接口。
四、SpringBoot 使用 RedissonClient 集成 Redis Stream 進(jìn)階：配置多個(gè)消費(fèi)組
1. 添加依賴
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.18</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<!-- Spring Boot 2.7.18 兼容的 redisson 版本 -->
<!-- 注意：Redisson 3.20.1需要Spring Boot 3.x，Spring Boot 2.7使用3.17.x版本 -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.17.7</version>
</dependency>2.配置文件
配置多個(gè)消費(fèi)組key，跟上一個(gè)案例的配置一致
# redis stream 配置多個(gè)消費(fèi)組key
app:
stream:
groups:
- key: "order_stream"
group: "order_group"
consumer-prefix: "consumer_"
- key: "payment_stream"
group: "payment_group"
consumer-prefix: "consumer_"
- key: "notification_stream"
group: "notification_group"
consumer-prefix: "consumer_"
spring:
# redisson配置
redisson:
config: |
singleServerConfig:
address: "redis://127.0.0.1:6379"
database: 0
# 連接超時(shí)時(shí)間（毫秒），默認(rèn)3000
connectionTimeout: 10000
# 命令執(zhí)行超時(shí)時(shí)間（毫秒），默認(rèn)3000，增加超時(shí)時(shí)間避免PING超時(shí)
timeout: 10000
# 空閑連接超時(shí)時(shí)間（毫秒），默認(rèn)10000
idleConnectionTimeout: 10000
# PING連接間隔（毫秒），默認(rèn)0（禁用），設(shè)置為0可以禁用PING避免超時(shí)
# 如果Redis服務(wù)器穩(wěn)定，可以設(shè)置為0禁用PING
pingConnectionInterval: 0
# 重試次數(shù)
retryAttempts: 3
# 重試間隔（毫秒）
retryInterval: 1500
# 保持連接活躍
keepAlive: true
# TCP無(wú)延遲
tcpNoDelay: true
# 連接池配置
connectionPoolSize: 64
connectionMinimumIdleSize: 24
# 如果Redis設(shè)置了密碼，取消下面的注釋并填寫密碼
# password: your_password3.Redis Stream消費(fèi)組配置屬性
package com.lijw.mp.config.redisstream;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* Redis Stream消費(fèi)組配置屬性
* - 讀取 app.stream 配置前綴下的所有配置項(xiàng)
*
* @author Aron.li
* @date 2025/11/30 11:21
*/
@Data
@Component
@ConfigurationProperties(prefix = "app.stream")
public class StreamGroupProperties {
/**
* 消費(fèi)組列表
*/
private List<StreamGroupConfig> groups;
/**
* 單個(gè)消費(fèi)組配置
*/
@Data
public static class StreamGroupConfig {
/**
* Stream鍵名
*/
private String key;
/**
* 消費(fèi)組名稱
*/
private String group;
/**
* 消費(fèi)者名稱前綴
*/
private String consumerPrefix;
}
}4. 消息生產(chǎn)者服務(wù)
package com.lijw.mp.event;
import com.lijw.mp.config.redisstream.StreamGroupProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.UUID;
/**
* Redis Stream消息生產(chǎn)者服務(wù)
* 負(fù)責(zé)向Redis Stream發(fā)送消息
*/
@Service
@Slf4j
public class MessageProducerService {
/**
* Redisson客戶端
*/
@Autowired
private RedissonClient redissonClient;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 發(fā)送消息到Redis Stream（根據(jù)消息類型自動(dòng)選擇stream）
*
* @param messageType 消息類型
* @param data 消息數(shù)據(jù)
* @return 消息ID
*/
public String sendMessage(String messageType, Map<String, String> data) {
// 根據(jù)消息類型自動(dòng)選擇stream key
String streamKey = getStreamKeyByMessageType(messageType);
return sendMessage(streamKey, messageType, data);
}
/**
* 發(fā)送消息到指定的Redis Stream
*
* @param streamKey Stream鍵名
* @param messageType 消息類型
* @param data 消息數(shù)據(jù)
* @return 消息ID
*/
/**
* 發(fā)送消息到指定的Redis Stream
*
* @param streamKey Stream鍵名
* @param messageType 消息類型
* @param data 消息數(shù)據(jù)
* @return 消息ID
*/
public String sendMessage(String streamKey, String messageType, Map<String, String> data) {
try {
// 添加消息類型和時(shí)間戳
data.put("messageType", messageType);
data.put("timestamp", String.valueOf(System.currentTimeMillis()));
data.put("messageId", UUID.randomUUID().toString());
// 使用Redisson發(fā)送消息
RStream<String, String> stream = redissonClient.getStream(streamKey);
// Redisson 3.17.7中，add方法直接接受Map參數(shù)
StreamMessageId messageId = stream.addAll(data);
log.info("消息發(fā)送成功 - Stream: {}, MessageId: {}", streamKey, messageId);
return messageId.toString();
} catch (Exception e) {
log.error("消息發(fā)送失敗 - Stream: {}", streamKey, e);
throw new RuntimeException("消息發(fā)送失敗", e);
}
}
/**
* 根據(jù)消息類型獲取對(duì)應(yīng)的stream key
*
* @param messageType 消息類型
* @return Stream鍵名
*/
private String getStreamKeyByMessageType(String messageType) {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
throw new RuntimeException("未配置任何消費(fèi)組");
}
// 根據(jù)消息類型前綴匹配stream key
String upperMessageType = messageType.toUpperCase();
if (upperMessageType.contains("ORDER")) {
return findStreamKey("order");
} else if (upperMessageType.contains("PAYMENT") || upperMessageType.contains("PAY")) {
return findStreamKey("payment");
} else if (upperMessageType.contains("NOTIFICATION") || upperMessageType.contains("NOTIFY")) {
return findStreamKey("notification");
} else {
// 默認(rèn)使用第一個(gè)stream
log.warn("未匹配到消息類型對(duì)應(yīng)的stream，使用默認(rèn)stream: {}", messageType);
return streamGroupProperties.getGroups().get(0).getKey();
}
}
/**
* 根據(jù)關(guān)鍵字查找stream key
*
* @param keyword 關(guān)鍵字
* @return Stream鍵名
*/
private String findStreamKey(String keyword) {
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
if (groupConfig.getKey().toLowerCase().contains(keyword.toLowerCase())) {
return groupConfig.getKey();
}
}
}
// 如果找不到，返回第一個(gè)stream
if (streamGroupProperties.getGroups() != null && !streamGroupProperties.getGroups().isEmpty()) {
return streamGroupProperties.getGroups().get(0).getKey();
}
throw new RuntimeException("未找到匹配的stream key: " + keyword);
}
}5. 消息消費(fèi)者服務(wù)（支持冪等性）
package com.lijw.mp.event;
import com.lijw.mp.config.redisstream.StreamGroupProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* Redis Stream消息消費(fèi)者服務(wù)
* 負(fù)責(zé)消費(fèi)Redis Stream中的消息并進(jìn)行業(yè)務(wù)處理
*/
@Component
@Slf4j
public class MessageConsumerService {
/**
* Redisson客戶端
*/
@Autowired
private RedissonClient redissonClient;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 緩存stream key到group name的映射關(guān)系
*/
private final Map<String, String> streamToGroupMap = new ConcurrentHashMap<>();
/**
* 消息消費(fèi)處理方法
* 當(dāng)Redis Stream中有新消息時(shí)，會(huì)調(diào)用此方法
*
* 多實(shí)例部署安全性：
* 1. Redis Stream Consumer Group 保證每條消息只會(huì)被一個(gè)消費(fèi)者讀取
* 2. 分布式鎖防止多實(shí)例同時(shí)處理同一條消息
* 3. 冪等性檢查作為雙重保障
*
* @param streamKey Stream鍵名
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param groupName 消費(fèi)組名稱
*/
public void onMessage(String streamKey, String messageId, Map<String, String> messageBody, String groupName) {
if (groupName == null) {
log.error("未找到stream key對(duì)應(yīng)的消費(fèi)組: {}, 消息ID: {}", streamKey, messageId);
return;
}
// 冪等性檢查：防止重復(fù)處理（第一道防線）
if (isMessageProcessed(messageId)) {
log.info("消息已處理，跳過(guò): {}", messageId);
acknowledgeMessage(streamKey, messageId, groupName);
return;
}
// 分布式鎖：防止多實(shí)例同時(shí)處理同一條消息（第二道防線）
String lockKey = "message:process:lock:" + messageId;
RBucket<String> lockBucket = redissonClient.getBucket(lockKey);
boolean lockAcquired = lockBucket.trySet("1", 5, TimeUnit.MINUTES);
if (!lockAcquired) {
log.info("消息正在被其他實(shí)例處理，跳過(guò): {}", messageId);
return;
}
try {
log.info("消費(fèi)者收到消息 - Stream: {}, Group: {}, ID: {}, 內(nèi)容: {}", streamKey, groupName, messageId, messageBody);
// 再次冪等性檢查（雙重檢查，防止在獲取鎖的過(guò)程中消息已被處理）
if (isMessageProcessed(messageId)) {
log.info("消息在處理過(guò)程中已被其他實(shí)例處理，跳過(guò): {}", messageId);
acknowledgeMessage(streamKey, messageId, groupName);
return;
}
// 處理業(yè)務(wù)邏輯（根據(jù)stream key路由到不同的業(yè)務(wù)處理）
boolean processSuccess = processBusiness(streamKey, messageBody);
if (processSuccess) {
// 標(biāo)記消息已處理
markMessageProcessed(messageId);
// 清除重試次數(shù)
clearRetryCount(messageId);
// 手動(dòng)確認(rèn)消息
acknowledgeMessage(streamKey, messageId, groupName);
log.info("消息處理完成: {}", messageId);
} else {
log.error("業(yè)務(wù)處理失敗，消息將重試: {}", messageId);
handleRetry(streamKey, messageId, messageBody, groupName, "業(yè)務(wù)處理失敗");
}
} catch (Exception e) {
log.error("消息處理異常: {}", messageId, e);
handleRetry(streamKey, messageId, messageBody, groupName, "消息處理異常");
} finally {
// 釋放分布式鎖
try {
lockBucket.delete();
} catch (Exception e) {
log.warn("釋放分布式鎖失敗: {}", messageId, e);
}
}
}
/**
* 根據(jù)stream key獲取對(duì)應(yīng)的消費(fèi)組名稱
*
* @param streamKey Stream鍵名
* @return 消費(fèi)組名稱，如果未找到返回null
*/
private String getGroupNameByStreamKey(String streamKey) {
// 先從緩存中獲取
if (streamToGroupMap.containsKey(streamKey)) {
return streamToGroupMap.get(streamKey);
}
// 從配置中查找
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
if (groupConfig.getKey().equals(streamKey)) {
streamToGroupMap.put(streamKey, groupConfig.getGroup());
return groupConfig.getGroup();
}
}
}
return null;
}
/**
* 冪等性檢查
* 使用Redis存儲(chǔ)，判斷redis是否已存在處理key，如果存在則說(shuō)明消息已處理，確保多實(shí)例間冪等性
*
* @param messageId 消息ID
* @return 如果消息已處理返回true，否則返回false
*/
private boolean isMessageProcessed(String messageId) {
RBucket<String> bucket = redissonClient.getBucket("processed:" + messageId);
return bucket.isExists();
}
/**
* 標(biāo)記消息已處理
* 使用Redis存儲(chǔ)事件處理ID，設(shè)置過(guò)期時(shí)間
*
* @param messageId 消息ID
*/
private void markMessageProcessed(String messageId) {
RBucket<String> bucket = redissonClient.getBucket("processed:" + messageId);
bucket.set("1", 24, TimeUnit.HOURS);
}
/**
* 業(yè)務(wù)處理邏輯（根據(jù)stream key路由到不同的業(yè)務(wù)處理）
*/
private boolean processBusiness(String streamKey, Map<String, String> messageBody) {
try {
// 根據(jù)stream key路由到不同的業(yè)務(wù)處理
if (streamKey.contains("order")) {
return processOrderBusiness(messageBody);
} else if (streamKey.contains("payment")) {
return processPaymentBusiness(messageBody);
} else if (streamKey.contains("notification")) {
return processNotificationBusiness(messageBody);
} else {
log.warn("未識(shí)別的stream key: {}, 使用默認(rèn)處理邏輯", streamKey);
return processDefaultBusiness(messageBody);
}
} catch (Exception e) {
log.error("業(yè)務(wù)處理異常", e);
return false;
}
}
/**
* 處理訂單業(yè)務(wù)
*/
private boolean processOrderBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String orderId = messageBody.get("orderId");
log.info("處理訂單{}消息，訂單ID: {}", messageType, orderId);
// TODO: 實(shí)現(xiàn)訂單業(yè)務(wù)處理邏輯
// try {
// // 模擬10秒處理業(yè)務(wù)
// Thread.sleep(100000);
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// }
// throw new RuntimeException("訂單業(yè)務(wù)處理異常");
return true;
}
/**
* 處理支付業(yè)務(wù)
*/
private boolean processPaymentBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String paymentId = messageBody.get("paymentId");
log.info("處理支付{}消息，支付ID: {}", messageType, paymentId);
// TODO: 實(shí)現(xiàn)支付業(yè)務(wù)處理邏輯
return true;
}
/**
* 處理通知業(yè)務(wù)
*/
private boolean processNotificationBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
String userId = messageBody.get("userId");
log.info("處理通知{}消息，用戶ID: {}", messageType, userId);
// TODO: 實(shí)現(xiàn)通知業(yè)務(wù)處理邏輯
return true;
}
/**
* 默認(rèn)業(yè)務(wù)處理
*/
private boolean processDefaultBusiness(Map<String, String> messageBody) {
String messageType = messageBody.get("messageType");
log.info("處理默認(rèn)業(yè)務(wù)消息: {}", messageType);
// TODO: 實(shí)現(xiàn)默認(rèn)業(yè)務(wù)處理邏輯
return true;
}
/**
* 手動(dòng)確認(rèn)消息
*
* @param streamKey Stream鍵名
* @param messageId 消息ID
* @param groupName 消費(fèi)組名稱
*/
private void acknowledgeMessage(String streamKey, String messageId, String groupName) {
try {
RStream<String, String> stream = redissonClient.getStream(streamKey);
// 解析消息ID字符串為StreamMessageId
StreamMessageId streamMessageId = parseStreamMessageId(messageId);
stream.ack(groupName, streamMessageId);
} catch (Exception e) {
log.error("消息確認(rèn)失敗: {}", messageId, e);
}
}
/**
* 解析消息ID字符串為StreamMessageId對(duì)象
*
* @param messageId 消息ID字符串，格式如 "1234567890-0"
* @return StreamMessageId對(duì)象
*/
private org.redisson.api.StreamMessageId parseStreamMessageId(String messageId) {
// StreamMessageId格式為 "timestamp-sequence"
String[] parts = messageId.split("-");
if (parts.length == 2) {
return new org.redisson.api.StreamMessageId(Long.parseLong(parts[0]), Long.parseLong(parts[1]));
}
// 如果格式不正確，嘗試直接解析
return new org.redisson.api.StreamMessageId(Long.parseLong(messageId), 0);
}
/**
* 增加重試次數(shù)
*
* @param messageId 消息ID
* @return 當(dāng)前重試次數(shù)
*/
private int incrementRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
RBucket<String> bucket = redissonClient.getBucket(retryKey);
String countStr = bucket.get();
int retryCount = countStr == null ? 0 : Integer.parseInt(countStr);
retryCount++;
// 設(shè)置重試次數(shù)，過(guò)期時(shí)間為24小時(shí)
bucket.set(String.valueOf(retryCount), 24, TimeUnit.HOURS);
return retryCount;
}
/**
* 清除重試次數(shù)
*
* @param messageId 消息ID
*/
private void clearRetryCount(String messageId) {
String retryKey = "retry:count:" + messageId;
RBucket<String> bucket = redissonClient.getBucket(retryKey);
bucket.delete();
}
/**
* 處理消息重試邏輯
*
* @param streamKey Stream鍵名
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param groupName 消費(fèi)組名稱
* @param errorDescription 錯(cuò)誤描述
*/
private void handleRetry(String streamKey, String messageId, Map<String, String> messageBody,
String groupName, String errorDescription) {
// 記錄重試次數(shù)
int retryCount = incrementRetryCount(messageId);
// 檢查是否超過(guò)最大重試次數(shù)
int maxRetryCount = 3; // 最大重試次數(shù)，可根據(jù)業(yè)務(wù)需求調(diào)整
if (retryCount >= maxRetryCount) {
log.error("消息{}重試次數(shù)已達(dá)上限({})，將停止重試并記錄: {}", errorDescription, maxRetryCount, messageId);
// 記錄失敗消息到死信隊(duì)列或告警（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)）
handleMaxRetryExceeded(messageId, messageBody, retryCount);
// 確認(rèn)消息，避免無(wú)限重試
acknowledgeMessage(streamKey, messageId, groupName);
return;
}
log.warn("消息{}，當(dāng)前重試次數(shù): {}/{}, 消息ID: {}", errorDescription, retryCount, maxRetryCount, messageId);
// 異步延遲重新處理消息
scheduleRetry(streamKey, messageId, messageBody, groupName, retryCount);
}
/**
* 調(diào)度消息重試
* 使用異步方式延遲重新處理消息，避免立即重試
*
* @param streamKey Stream鍵名
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param groupName 消費(fèi)組名稱
* @param retryCount 當(dāng)前重試次數(shù)
*/
private void scheduleRetry(String streamKey, String messageId, Map<String, String> messageBody,
String groupName, int retryCount) {
// 計(jì)算延遲時(shí)間：重試次數(shù)越多，延遲時(shí)間越長(zhǎng)（指數(shù)退避策略）
long delaySeconds = (long) Math.pow(2, retryCount - 1);
log.info("消息將在{}秒后重新處理，消息ID: {}", delaySeconds, messageId);
// 異步延遲重新處理消息
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(delaySeconds * 1000);
log.info("開始重新處理消息，消息ID: {}", messageId);
// 重新調(diào)用onMessage處理消息
onMessage(streamKey, messageId, messageBody, groupName);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("消息重試被中斷，消息ID: {}", messageId);
} catch (Exception e) {
log.error("消息重試處理異常，消息ID: {}", messageId, e);
}
});
}
/**
* 處理超過(guò)最大重試次數(shù)的消息
* 記錄失敗消息詳情，可根據(jù)業(yè)務(wù)需求擴(kuò)展（如存儲(chǔ)到數(shù)據(jù)庫(kù)、發(fā)送告警等）
*
* @param messageId 消息ID
* @param messageBody 消息內(nèi)容
* @param retryCount 重試次數(shù)
*/
private void handleMaxRetryExceeded(String messageId, Map<String, String> messageBody, int retryCount) {
try {
// 記錄失敗消息詳情（可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)，如存儲(chǔ)到數(shù)據(jù)庫(kù)、發(fā)送告警等）
String failedKey = "failed:message:" + messageId;
String failedInfo = String.format("消息ID: %s, 重試次數(shù): %d, 消息內(nèi)容: %s, 失敗時(shí)間: %s",
messageId, retryCount, messageBody, System.currentTimeMillis());
RBucket<String> bucket = redissonClient.getBucket(failedKey);
bucket.set(failedInfo, 7, TimeUnit.DAYS);
log.error("失敗消息已記錄: {}", failedInfo);
// TODO: 可根據(jù)業(yè)務(wù)需求添加其他處理邏輯，如：
// 1. 發(fā)送告警通知
// 2. 存儲(chǔ)到數(shù)據(jù)庫(kù)死信表
// 3. 發(fā)送到死信隊(duì)列
} catch (Exception e) {
log.error("處理超過(guò)最大重試次數(shù)消息異常: {}", messageId, e);
}
}
}6. 消費(fèi)者容器配置（支持多實(shí)例負(fù)載均衡）
package com.lijw.mp.config.redisstream;
import com.lijw.mp.event.MessageConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.net.InetAddress;
import java.lang.management.ManagementFactory;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* Redis Stream配置類
* 負(fù)責(zé)創(chuàng)建消費(fèi)組、配置消息監(jiān)聽容器等
* 使用Redisson實(shí)現(xiàn)消息消費(fèi)
*/
@Configuration
@Slf4j
public class RedisStreamConfig {
/**
* Redisson客戶端
*/
@Autowired
private RedissonClient redissonClient;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 消息消費(fèi)者服務(wù)
*/
@Autowired
private MessageConsumerService messageConsumerService;
/**
* 消費(fèi)者線程池
*/
private ExecutorService consumerExecutor;
/**
* 應(yīng)用啟動(dòng)完成后，初始化消費(fèi)組并啟動(dòng)消費(fèi)者
*/
@EventListener(ApplicationReadyEvent.class)
public void initStreamConsumers() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組");
return;
}
// 創(chuàng)建消費(fèi)者線程池
consumerExecutor = createConsumerThreadPool();
// 創(chuàng)建所有消費(fèi)組
createAllConsumerGroups();
// 啟動(dòng)所有消費(fèi)者
startAllConsumers();
}
/**
* 確保stream存在，如果不存在則創(chuàng)建
*
* @param streamKey Stream鍵名
*/
private void ensureStreamExists(String streamKey) {
try {
RStream<String, String> stream = redissonClient.getStream(streamKey);
// 檢查stream是否存在
Long size = stream.size();
if (size == null || size == 0) {
// Stream不存在，創(chuàng)建一條臨時(shí)消息來(lái)初始化stream
Map<String, String> tempData = new HashMap<>();
tempData.put("_init", "true");
tempData.put("_timestamp", String.valueOf(System.currentTimeMillis()));
stream.addAll(tempData);
log.debug("創(chuàng)建stream成功: {}", streamKey);
}
} catch (Exception e) {
// Stream不存在，創(chuàng)建它
try {
RStream<String, String> stream = redissonClient.getStream(streamKey);
Map<String, String> tempData = new HashMap<>();
tempData.put("_init", "true");
tempData.put("_timestamp", String.valueOf(System.currentTimeMillis()));
stream.addAll(tempData);
log.info("創(chuàng)建stream成功: {}", streamKey);
} catch (Exception ex) {
log.warn("創(chuàng)建stream失敗: {}, {}", streamKey, ex.getMessage());
}
}
}
/**
* 啟動(dòng)所有消費(fèi)者
*/
private void startAllConsumers() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組，無(wú)法啟動(dòng)消費(fèi)者");
return;
}
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
startConsumer(groupConfig);
}
}
/**
* 創(chuàng)建所有消費(fèi)者組（如果不存在）
* 如果stream不存在，會(huì)先創(chuàng)建stream
*/
private void createAllConsumerGroups() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
log.warn("未配置任何消費(fèi)組");
return;
}
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
// 先檢查stream是否存在，如果不存在則創(chuàng)建
ensureStreamExists(groupConfig.getKey());
// 創(chuàng)建消費(fèi)組
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
stream.createGroup(groupConfig.getGroup(), StreamMessageId.ALL);
log.info("創(chuàng)建消費(fèi)者組成功: stream={}, group={}", groupConfig.getKey(), groupConfig.getGroup());
} catch (Exception e) {
// 如果消費(fèi)組已存在，這是正常情況
String errorMsg = e.getMessage() != null ? e.getMessage() : "";
if (errorMsg.contains("BUSYGROUP") || errorMsg.contains("already exists")) {
log.info("消費(fèi)者組已存在: stream={}, group={}", groupConfig.getKey(), groupConfig.getGroup());
} else {
log.warn("創(chuàng)建消費(fèi)者組失敗: stream={}, group={}, error={}",
groupConfig.getKey(), groupConfig.getGroup(), errorMsg);
// 即使創(chuàng)建失敗，也繼續(xù)處理其他消費(fèi)組
}
}
}
}
/**
* 啟動(dòng)單個(gè)消費(fèi)組的消費(fèi)者
*
* @param groupConfig 消費(fèi)組配置
*/
private void startConsumer(StreamGroupProperties.StreamGroupConfig groupConfig) {
// 為每個(gè)實(shí)例生成唯一消費(fèi)者名稱
String consumerName = generateUniqueConsumerName(groupConfig.getConsumerPrefix());
// 啟動(dòng)消費(fèi)者線程
consumerExecutor.submit(() -> {
consumeMessages(groupConfig, consumerName);
});
log.info("Redis Stream消費(fèi)者啟動(dòng)成功 - Stream: {}, Group: {}, Consumer: {}",
groupConfig.getKey(), groupConfig.getGroup(), consumerName);
}
/**
* 消費(fèi)消息的循環(huán)方法
*
* @param groupConfig 消費(fèi)組配置
* @param consumerName 消費(fèi)者名稱
*/
private void consumeMessages(StreamGroupProperties.StreamGroupConfig groupConfig, String consumerName) {
String streamKey = groupConfig.getKey();
String groupName = groupConfig.getGroup();
RStream<String, String> stream = redissonClient.getStream(streamKey);
log.info("開始消費(fèi)消息 - Stream: {}, Group: {}, Consumer: {}", streamKey, groupName, consumerName);
while (!Thread.currentThread().isInterrupted()) {
try {
// 讀取消息（從未確認(rèn)的消息開始，讀取最多10條）
Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
groupName,
consumerName,
10
);
if (messages != null && !messages.isEmpty()) {
// 處理每條消息
for (Map.Entry<StreamMessageId, Map<String, String>> entry : messages.entrySet()) {
StreamMessageId messageId = entry.getKey();
Map<String, String> messageBody = entry.getValue();
// 調(diào)用消費(fèi)者服務(wù)處理消息
messageConsumerService.onMessage(streamKey, messageId.toString(), messageBody, groupName);
}
} else {
// 沒有消息時(shí)，短暫休眠避免CPU空轉(zhuǎn)
Thread.sleep(100);
}
} catch (InterruptedException e) {
log.info("消費(fèi)者線程被中斷 - Stream: {}, Group: {}, Consumer: {}", streamKey, groupName, consumerName);
Thread.currentThread().interrupt();
break;
} catch (Exception e) {
log.error("消費(fèi)消息異常 - Stream: {}, Group: {}, Consumer: {}",
streamKey, groupName, consumerName, e);
// 發(fā)生異常時(shí)短暫休眠，避免快速重試導(dǎo)致CPU占用過(guò)高
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
log.info("消費(fèi)者線程結(jié)束 - Stream: {}, Group: {}, Consumer: {}", streamKey, groupName, consumerName);
}
/**
* 生成唯一消費(fèi)者名稱（支持多實(shí)例部署的關(guān)鍵）
* 使用IP+進(jìn)程ID確保集群環(huán)境下唯一性
*
* @param prefix 消費(fèi)者名稱前綴
* @return 唯一的消費(fèi)者名稱
*/
private String generateUniqueConsumerName(String prefix) {
try {
String hostAddress = InetAddress.getLocalHost().getHostAddress();
String processId = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
return (prefix != null ? prefix : "consumer_") + hostAddress + "_" + processId + "_" + System.currentTimeMillis();
} catch (Exception e) {
// fallback：使用UUID
return (prefix != null ? prefix : "consumer_") + UUID.randomUUID().toString().substring(0, 8);
}
}
/**
* 創(chuàng)建專用線程池
* 用于處理Redis Stream消息消費(fèi)
*
* @return 線程池執(zhí)行器
*/
private ExecutorService createConsumerThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2); // 核心線程數(shù)
executor.setMaxPoolSize(10); // 最大線程數(shù)
executor.setQueueCapacity(100); // 隊(duì)列容量
executor.setThreadNamePrefix("redis-stream-consumer-"); // 線程名前綴
executor.setDaemon(false); // 非守護(hù)線程，確保應(yīng)用關(guān)閉時(shí)能正常結(jié)束
executor.initialize();
return executor.getThreadPoolExecutor();
}
}7. 定時(shí)處理消息告警、重試pending消息
package com.lijw.mp.config.redisstream;
import com.lijw.mp.event.MessageConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* 待處理消息重試服務(wù)
* 定時(shí)檢查并處理Redis Stream中未被確認(rèn)（ACK）的消息
*
* 生產(chǎn)環(huán)境特性：
* 1. 定時(shí)掃描pending消息
* 2. 讀取pending消息并重新處理
* 3. 復(fù)用MessageConsumerService進(jìn)行業(yè)務(wù)處理
* 4. 支持消息空閑時(shí)間檢查和告警
*/
@Component
@Slf4j
public class PendingMessageRetryService {
/**
* Redisson客戶端
*/
@Autowired
private RedissonClient redissonClient;
/**
* Stream消費(fèi)組配置屬性
*/
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 消息消費(fèi)者服務(wù)（用于重新處理pending消息）
*/
@Autowired
private MessageConsumerService messageConsumerService;
/**
* 定時(shí)監(jiān)控各個(gè)消費(fèi)隊(duì)列的數(shù)量
* 執(zhí)行時(shí)機(jī)：每10秒執(zhí)行一次
*/
@Scheduled(fixedDelay = 10000)
public void monitorQueueStats() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
return;
}
long totalMessages = 0;
long totalPendingMessages = 0;
StringBuilder statsBuilder = new StringBuilder("消息隊(duì)列監(jiān)控統(tǒng)計(jì) | ");
// 遍歷所有配置的消費(fèi)組
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
// 獲取總消息數(shù)
Long messageCount = stream.size();
messageCount = messageCount != null ? messageCount : 0L;
// 獲取pending消息數(shù)
long pendingCount = getPendingMessageCount(groupConfig);
// 計(jì)算已處理消息數(shù)
long processedCount = messageCount - pendingCount;
totalMessages += messageCount;
totalPendingMessages += pendingCount;
// 構(gòu)建單條日志信息
statsBuilder.append(String.format("[%s/%s: 總數(shù)=%d, Pending=%d, 已處理=%d] ",
groupConfig.getKey(),
groupConfig.getGroup(),
messageCount,
pendingCount,
Math.max(0, processedCount)));
} catch (Exception e) {
log.error("監(jiān)控Stream統(tǒng)計(jì)信息失敗 - Stream: {}, Group: {}",
groupConfig.getKey(), groupConfig.getGroup(), e);
statsBuilder.append(String.format("[%s/%s: 統(tǒng)計(jì)失敗] ",
groupConfig.getKey(), groupConfig.getGroup()));
}
}
// 添加總計(jì)信息
statsBuilder.append(String.format("| 總計(jì): 總數(shù)=%d, Pending=%d, 已處理=%d",
totalMessages,
totalPendingMessages,
Math.max(0, totalMessages - totalPendingMessages)));
// 打印單條日志
log.info(statsBuilder.toString());
}
/**
* 定時(shí)處理未確認(rèn)的消息
* 執(zhí)行時(shí)機(jī)：每30秒(30000)執(zhí)行一次（通過(guò)@Scheduled注解配置）
*/
@Scheduled(fixedDelay = 30000)
public void retryPendingMessages() {
if (streamGroupProperties.getGroups() == null || streamGroupProperties.getGroups().isEmpty()) {
return;
}
// 遍歷所有配置的消費(fèi)組
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
// 獲取待處理消息數(shù)量（通過(guò)讀取pending消息來(lái)統(tǒng)計(jì)）
long totalPending = getPendingMessageCount(groupConfig);
if (totalPending > 0) {
log.info("發(fā)現(xiàn)待處理消息 - Stream: {}, Group: {}, 待處理數(shù)量: {}",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
// 根據(jù)pending消息數(shù)量進(jìn)行告警
if (totalPending > 1000) {
log.error("【嚴(yán)重告警】待處理消息數(shù)量過(guò)多 - Stream: {}, Group: {}, 數(shù)量: {}, " +
"建議：1.檢查消費(fèi)者是否正常運(yùn)行 2.增加消費(fèi)實(shí)例 3.檢查消息處理邏輯",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
} else if (totalPending > 100) {
log.warn("【警告】待處理消息數(shù)量較多 - Stream: {}, Group: {}, 數(shù)量: {}, " +
"建議：考慮增加消費(fèi)實(shí)例或檢查消息處理性能",
groupConfig.getKey(), groupConfig.getGroup(), totalPending);
}
// 讀取并處理pending消息
readAndProcessPendingMessages(groupConfig);
}
} catch (Exception e) {
log.error("處理待處理消息異常 - Stream: {}, Group: {}",
groupConfig.getKey(), groupConfig.getGroup(), e);
}
}
}
/**
* 獲取待處理消息數(shù)量
* 通過(guò)讀取pending消息來(lái)統(tǒng)計(jì)數(shù)量
*
* @param groupConfig 消費(fèi)組配置
* @return 待處理消息數(shù)量
*/
private long getPendingMessageCount(StreamGroupProperties.StreamGroupConfig groupConfig) {
try {
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
String consumerName = "count-consumer";
// 嘗試讀取pending消息（使用0作為起始ID，讀取最多1000條來(lái)統(tǒng)計(jì)）
Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
groupConfig.getGroup(),
consumerName,
1000,
new StreamMessageId(0, 0)
);
return messages != null ? messages.size() : 0;
} catch (Exception e) {
log.warn("獲取pending消息數(shù)量失敗 - Stream: {}, Group: {}",
groupConfig.getKey(), groupConfig.getGroup(), e);
return 0;
}
}
/**
* 讀取并處理pending消息
* 使用XREADGROUP命令讀取pending消息（使用0作為起始ID）
*
* 在Redis Stream中，使用XREADGROUP GROUP group consumer STREAMS key 0
* 可以讀取該消費(fèi)者組中所有pending的消息
*
* @param groupConfig 消費(fèi)組配置
*/
private void readAndProcessPendingMessages(StreamGroupProperties.StreamGroupConfig groupConfig) {
try {
// 使用固定的消費(fèi)者名稱來(lái)讀取pending消息
String retryConsumerName = "retry-consumer";
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
log.info("嘗試讀取pending消息 - Stream: {}, Group: {}, Consumer: {}",
groupConfig.getKey(), groupConfig.getGroup(), retryConsumerName);
// 讀取pending消息（使用0作為起始ID讀取所有pending消息，最多100條）
// 在Redis Stream中，0表示讀取所有pending消息
Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
groupConfig.getGroup(),
retryConsumerName,
100,
new StreamMessageId(0, 0)
);
log.info("讀取結(jié)果 - Stream: {}, Group: {}, Consumer: {}, 記錄數(shù)量: {}",
groupConfig.getKey(), groupConfig.getGroup(), retryConsumerName,
messages != null ? messages.size() : 0);
if (messages != null && !messages.isEmpty()) {
log.info("讀取到{}條pending消息，開始重新處理 - Stream: {}, Group: {}",
messages.size(), groupConfig.getKey(), groupConfig.getGroup());
int successCount = 0;
int failCount = 0;
// 復(fù)用MessageConsumerService的業(yè)務(wù)處理方法
for (Map.Entry<StreamMessageId, Map<String, String>> entry : messages.entrySet()) {
StreamMessageId streamMessageId = entry.getKey();
String messageId = streamMessageId.toString();
Map<String, String> messageBody = entry.getValue();
String lockKey = "pending:retry:lock:" + messageId;
try {
// 使用分布式鎖防止多實(shí)例重復(fù)處理
RBucket<String> lockBucket = redissonClient.getBucket(lockKey);
boolean lockAcquired = lockBucket.trySet("1", 5, TimeUnit.MINUTES);
if (!lockAcquired) {
// 鎖已被其他實(shí)例獲取，跳過(guò)處理
log.info("pending消息正在被其他實(shí)例處理，跳過(guò) - Stream: {}, Group: {}, MessageId: {}",
groupConfig.getKey(), groupConfig.getGroup(), messageId);
continue;
}
// 冪等性檢查：檢查消息是否已被處理
if (isMessageProcessed(messageId)) {
log.info("pending消息已處理，跳過(guò) - Stream: {}, Group: {}, MessageId: {}",
groupConfig.getKey(), groupConfig.getGroup(), messageId);
// 釋放鎖
lockBucket.delete();
continue;
}
log.info("開始處理pending消息 - Stream: {}, Group: {}, MessageId: {}",
groupConfig.getKey(), groupConfig.getGroup(), messageId);
log.debug("消息內(nèi)容: {}", messageBody);
// 復(fù)用MessageConsumerService的onMessage方法進(jìn)行業(yè)務(wù)處理
messageConsumerService.onMessage(groupConfig.getKey(), messageId, messageBody, groupConfig.getGroup());
successCount++;
log.info("pending消息重新處理成功 - Stream: {}, Group: {}, MessageId: {}",
groupConfig.getKey(), groupConfig.getGroup(), messageId);
// 處理成功后釋放鎖
lockBucket.delete();
} catch (Exception e) {
failCount++;
log.error("pending消息重新處理失敗 - Stream: {}, Group: {}, MessageId: {}, 錯(cuò)誤: {}",
groupConfig.getKey(), groupConfig.getGroup(), messageId,
e.getMessage(), e);
// 處理失敗后釋放鎖，允許重試
try {
RBucket<String> lockBucket = redissonClient.getBucket(lockKey);
lockBucket.delete();
} catch (Exception ex) {
log.warn("釋放鎖失敗: {}", ex.getMessage());
}
}
}
log.info("pending消息處理完成 - Stream: {}, Group: {}, 成功: {}, 失敗: {}",
groupConfig.getKey(), groupConfig.getGroup(), successCount, failCount);
} else {
log.warn("未讀取到pending消息 - Stream: {}, Group: {}, 可能原因：1.消息已被其他消費(fèi)者處理 2.消費(fèi)者名稱不匹配 3.消息已被ACK",
groupConfig.getKey(), groupConfig.getGroup());
}
} catch (Exception e) {
log.error("讀取pending消息失敗 - Stream: {}, Group: {}, 錯(cuò)誤類型: {}, 錯(cuò)誤消息: {}",
groupConfig.getKey(), groupConfig.getGroup(),
e.getClass().getSimpleName(), e.getMessage(), e);
}
}
/**
* 冪等性檢查：檢查消息是否已被處理
* 與MessageConsumerService使用相同的冪等性檢查機(jī)制
*
* @param messageId 消息ID
* @return 如果消息已處理返回true，否則返回false
*/
private boolean isMessageProcessed(String messageId) {
RBucket<String> bucket = redissonClient.getBucket("processed:" + messageId);
return bucket.isExists();
}
}8.REST控制器
package com.lijw.mp.controller;
import com.lijw.mp.config.redisstream.StreamGroupProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/test/stream")
@Slf4j
public class StreamTestController {
@Autowired
private RedissonClient redissonClient;
@Autowired
private StreamGroupProperties streamGroupProperties;
/**
* 獲取默認(rèn)的stream key（訂單stream）
*/
private String getDefaultStreamKey() {
if (streamGroupProperties.getGroups() != null && !streamGroupProperties.getGroups().isEmpty()) {
// 優(yōu)先查找order_stream，否則返回第一個(gè)
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
if (groupConfig.getKey().contains("order")) {
return groupConfig.getKey();
}
}
return streamGroupProperties.getGroups().get(0).getKey();
}
throw new RuntimeException("未配置任何消費(fèi)組");
}
/**
* 發(fā)送JSON格式復(fù)雜消息
*/
@PostMapping("/send-json")
public ResponseEntity<Map<String, Object>> sendJsonMessage(
@RequestBody Map<String, Object> messageData,
@RequestParam(required = false) String streamKey) {
try {
// 如果未指定stream key，使用默認(rèn)的
if (streamKey == null || streamKey.isEmpty()) {
streamKey = getDefaultStreamKey();
}
// 添加元數(shù)據(jù)
messageData.put("timestamp", System.currentTimeMillis());
messageData.put("messageType", "CUSTOM_EVENT");
// 轉(zhuǎn)換Map類型以適應(yīng)Redis Template
Map<String, String> stringMessage = new HashMap<>();
messageData.forEach((k, v) -> stringMessage.put(k, v.toString()));
RStream<String, String> stream = redissonClient.getStream(streamKey);
StreamMessageId messageId = stream.addAll(stringMessage);
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId.toString());
result.put("data", messageData);
log.info("JSON消息發(fā)送成功: {}", messageId);
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送JSON消息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(result);
}
}
/**
* 發(fā)送訂單事件測(cè)試消息
*/
@PostMapping("/send-order")
public ResponseEntity<Map<String, Object>> sendOrderMessage(
@RequestParam String orderId,
@RequestParam Double amount,
@RequestParam String status,
@RequestParam(required = false) String streamKey) {
try {
// 如果未指定stream key，使用默認(rèn)的訂單stream
if (streamKey == null || streamKey.isEmpty()) {
streamKey = getDefaultStreamKey();
}
Map<String, String> message = new HashMap<>();
message.put("eventType", "ORDER_" + status.toUpperCase());
message.put("orderId", orderId);
message.put("amount", amount.toString());
message.put("status", status);
message.put("eventTime", LocalDateTime.now().toString());
RStream<String, String> stream = redissonClient.getStream(streamKey);
StreamMessageId messageId = stream.addAll(message);
Map<String, Object> result = new HashMap<>();
result.put("success", true);
result.put("messageId", messageId.toString());
result.put("eventType", message.get("eventType"));
result.put("orderId", orderId);
log.info("訂單事件發(fā)送成功: {} - {}", message.get("eventType"), orderId);
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("發(fā)送訂單事件失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(result);
}
}
/**
* 查看所有配置的Stream信息
*/
@GetMapping("/info/all")
public ResponseEntity<Map<String, Object>> getAllStreamInfo() {
try {
Map<String, Object> result = new HashMap<>();
Map<String, Long> streams = new HashMap<>();
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
Long streamSize = stream.size();
streams.put(groupConfig.getKey(), streamSize);
} catch (Exception e) {
log.warn("獲取Stream信息失敗: {}", groupConfig.getKey(), e);
streams.put(groupConfig.getKey(), -1L);
}
}
}
result.put("streams", streams);
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("獲取所有Stream信息失敗", e);
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
}
}
/**
* 查詢消息隊(duì)列總數(shù)情況（監(jiān)控接口）
* 返回所有Stream的詳細(xì)信息，包括總消息數(shù)、pending消息數(shù)、消費(fèi)組信息等
*/
@GetMapping("/stats")
public ResponseEntity<Map<String, Object>> getStreamStats() {
try {
Map<String, Object> result = new HashMap<>();
List<Map<String, Object>> streamStats = new ArrayList<>();
long totalMessages = 0;
long totalPendingMessages = 0;
if (streamGroupProperties.getGroups() != null) {
for (StreamGroupProperties.StreamGroupConfig groupConfig : streamGroupProperties.getGroups()) {
try {
Map<String, Object> stats = getStreamStatistics(groupConfig);
streamStats.add(stats);
// 累計(jì)統(tǒng)計(jì)
totalMessages += (Long) stats.getOrDefault("messageCount", 0L);
totalPendingMessages += (Long) stats.getOrDefault("pendingCount", 0L);
} catch (Exception e) {
log.warn("獲取Stream統(tǒng)計(jì)信息失敗: {}", groupConfig.getKey(), e);
Map<String, Object> errorStats = new HashMap<>();
errorStats.put("streamKey", groupConfig.getKey());
errorStats.put("groupName", groupConfig.getGroup());
errorStats.put("error", e.getMessage());
streamStats.add(errorStats);
}
}
}
result.put("streams", streamStats);
// 創(chuàng)建匯總信息
Map<String, Object> summary = new HashMap<>();
summary.put("totalStreams", streamStats.size());
summary.put("totalMessages", totalMessages);
summary.put("totalPendingMessages", totalPendingMessages);
result.put("summary", summary);
result.put("timestamp", System.currentTimeMillis());
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("獲取Stream統(tǒng)計(jì)信息失敗", e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(result);
}
}
/**
* 獲取單個(gè)Stream的統(tǒng)計(jì)信息
*
* @param groupConfig 消費(fèi)組配置
* @return Stream統(tǒng)計(jì)信息
*/
private Map<String, Object> getStreamStatistics(StreamGroupProperties.StreamGroupConfig groupConfig) {
Map<String, Object> stats = new HashMap<>();
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
// 總消息數(shù)
Long messageCount = stream.size();
stats.put("streamKey", groupConfig.getKey());
stats.put("groupName", groupConfig.getGroup());
stats.put("messageCount", messageCount != null ? messageCount : 0L);
// Pending消息數(shù)（通過(guò)讀取pending消息來(lái)統(tǒng)計(jì)）
long pendingCount = getPendingMessageCount(groupConfig);
stats.put("pendingCount", pendingCount);
// 已處理消息數(shù)（總消息數(shù) - pending消息數(shù)）
long processedCount = (messageCount != null ? messageCount : 0L) - pendingCount;
stats.put("processedCount", Math.max(0, processedCount));
return stats;
}
/**
* 獲取待處理消息數(shù)量
*
* @param groupConfig 消費(fèi)組配置
* @return 待處理消息數(shù)量
*/
private long getPendingMessageCount(StreamGroupProperties.StreamGroupConfig groupConfig) {
try {
RStream<String, String> stream = redissonClient.getStream(groupConfig.getKey());
String consumerName = "stats-consumer";
// 嘗試讀取pending消息（使用0作為起始ID，讀取最多1000條來(lái)統(tǒng)計(jì)）
Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
groupConfig.getGroup(),
consumerName,
1000,
new StreamMessageId(0, 0)
);
return messages != null ? messages.size() : 0;
} catch (Exception e) {
log.warn("獲取pending消息數(shù)量失敗 - Stream: {}, Group: {}",
groupConfig.getKey(), groupConfig.getGroup(), e);
return 0;
}
}
/**
* 清除消息隊(duì)列（運(yùn)維接口）
* 注意：此操作會(huì)刪除Stream中的所有消息，請(qǐng)謹(jǐn)慎使用
*
* @param streamKey Stream鍵名（必填，防止誤操作）
* @param confirm 確認(rèn)參數(shù)，必須為"DELETE"才會(huì)執(zhí)行刪除操作
* @return 操作結(jié)果
*/
@DeleteMapping("/clear")
public ResponseEntity<Map<String, Object>> clearStream(
@RequestParam String streamKey,
@RequestParam(required = false, defaultValue = "") String confirm) {
try {
Map<String, Object> result = new HashMap<>();
// 安全檢查：必須提供確認(rèn)參數(shù)
if (!"DELETE".equals(confirm)) {
result.put("success", false);
result.put("error", "清除操作需要確認(rèn)，請(qǐng)?jiān)赾onfirm參數(shù)中傳入'DELETE'");
result.put("message", "此操作會(huì)刪除Stream中的所有消息，請(qǐng)謹(jǐn)慎使用");
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(result);
}
// 檢查stream是否存在
RStream<String, String> stream = redissonClient.getStream(streamKey);
Long sizeBefore = stream.size();
if (sizeBefore == null || sizeBefore == 0) {
result.put("success", true);
result.put("message", "Stream為空，無(wú)需清除");
result.put("streamKey", streamKey);
result.put("deletedCount", 0);
return ResponseEntity.ok(result);
}
// 刪除Stream（刪除整個(gè)Stream會(huì)清除所有消息）
// 注意：Redisson中刪除Stream需要使用Redis命令，這里使用trim到0的方式
// 或者直接刪除key
redissonClient.getKeys().delete(streamKey);
result.put("success", true);
result.put("message", "Stream清除成功");
result.put("streamKey", streamKey);
result.put("deletedCount", sizeBefore);
result.put("timestamp", System.currentTimeMillis());
log.warn("Stream已清除 - Stream: {}, 刪除消息數(shù): {}", streamKey, sizeBefore);
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("清除Stream失敗 - Stream: {}", streamKey, e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(result);
}
}
/**
* 清除指定Stream的pending消息（運(yùn)維接口）
* 只清除pending消息，保留已處理的消息
*
* @param streamKey Stream鍵名
* @param groupName 消費(fèi)組名稱
* @param confirm 確認(rèn)參數(shù)，必須為"DELETE_PENDING"才會(huì)執(zhí)行刪除操作
* @return 操作結(jié)果
*/
@DeleteMapping("/clear-pending")
public ResponseEntity<Map<String, Object>> clearPendingMessages(
@RequestParam String streamKey,
@RequestParam String groupName,
@RequestParam(required = false, defaultValue = "") String confirm) {
try {
Map<String, Object> result = new HashMap<>();
// 安全檢查：必須提供確認(rèn)參數(shù)
if (!"DELETE_PENDING".equals(confirm)) {
result.put("success", false);
result.put("error", "清除pending消息操作需要確認(rèn)，請(qǐng)?jiān)赾onfirm參數(shù)中傳入'DELETE_PENDING'");
result.put("message", "此操作會(huì)刪除Stream中所有pending消息，請(qǐng)謹(jǐn)慎使用");
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(result);
}
RStream<String, String> stream = redissonClient.getStream(streamKey);
// 讀取所有pending消息并確認(rèn)它們（通過(guò)ACK來(lái)清除pending狀態(tài)）
String consumerName = "clear-pending-consumer";
int clearedCount = 0;
int maxIterations = 100; // 防止無(wú)限循環(huán)
int iteration = 0;
while (iteration < maxIterations) {
Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
groupName,
consumerName,
100,
new StreamMessageId(0, 0)
);
if (messages == null || messages.isEmpty()) {
break;
}
// 確認(rèn)所有pending消息
for (StreamMessageId messageId : messages.keySet()) {
stream.ack(groupName, messageId);
clearedCount++;
}
iteration++;
}
result.put("success", true);
result.put("message", "Pending消息清除成功");
result.put("streamKey", streamKey);
result.put("groupName", groupName);
result.put("clearedCount", clearedCount);
result.put("timestamp", System.currentTimeMillis());
log.warn("Pending消息已清除 - Stream: {}, Group: {}, 清除數(shù)量: {}",
streamKey, groupName, clearedCount);
return ResponseEntity.ok(result);
} catch (Exception e) {
log.error("清除Pending消息失敗 - Stream: {}, Group: {}", streamKey, groupName, e);
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("error", e.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(result);
}
}
}8.1 查詢消息隊(duì)列總數(shù)情況（監(jiān)控接口）
GET http://localhost:8083/api/test/stream/stats
8.2 查看所有配置的Stream信息
GET http://localhost:8083/api/test/stream/info/all
8.3 發(fā)送JSON格式復(fù)雜消息
POST http://localhost:8083/api/test/stream/send-json
# PARAM
streamKey=order_stream
# Body
{
"orderId": "orderId123345123",
"amount": 500,
"messageType": "ORDER"
}
8.4 發(fā)送訂單事件測(cè)試消息
POST http://localhost:8083/api/test/stream/send-order # PARAM streamKey:order_stream orderId:orderId2025120701 amount:500 status:ok messageType:order
9.總結(jié)
上面的幾個(gè)接口已經(jīng)說(shuō)明了使用方式，下面截圖展示一下相應(yīng)的執(zhí)行日志：
消息的發(fā)送以及處理
消息的定時(shí)監(jiān)控
消息隊(duì)列監(jiān)控統(tǒng)計(jì) | [order_stream/order_group: 總數(shù)=212, Pending=0, 已處理=212] [payment_stream/payment_group: 總數(shù)=2, Pending=0, 已處理=2] [notification_stream/notification_group: 總數(shù)=2, Pending=0, 已處理=2] | 總計(jì): 總數(shù)=216, Pending=0, 已處理=216
到此這篇關(guān)于Spring Boot集成Redis Stream消息隊(duì)列從入門到實(shí)戰(zhàn)指南的文章就介紹到這了,更多相關(guān)Spring Boot集成Redis Stream消息隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- SpringBoot使用Redis Stream實(shí)現(xiàn)輕量消息隊(duì)列的示例代碼
- SpringBoot中實(shí)現(xiàn)Redis?Stream隊(duì)列的代碼實(shí)例
- redis之基于SpringBoot實(shí)現(xiàn)Redis stream實(shí)時(shí)流事件處理方式
- SpringBoot中使用Redis?Stream實(shí)現(xiàn)消息監(jiān)聽示例
- 如何使用redis的stream數(shù)據(jù)類型做消息隊(duì)列
- 解讀Redis秒殺優(yōu)化方案(阻塞隊(duì)列+基于Stream流的消息隊(duì)列)
- 一文弄懂Redis Stream消息隊(duì)列
相關(guān)文章
SpringMVC之簡(jiǎn)單的增刪改查示例（SSM整合）
本篇文章主要介紹了SpringMVC之簡(jiǎn)單的增刪改查示例（SSM整合），這個(gè)例子是基于SpringMVC+Spring+Mybatis實(shí)現(xiàn)的。有興趣的可以了解一下。2017-03-03
SpringBoot中@EnableAutoConfiguration和@Configuration的區(qū)別
這篇文章主要介紹了SpringBoot中@EnableAutoConfiguration和@Configuration的區(qū)別,@SpringBootApplication相當(dāng)于@EnableAutoConfiguration，@ComponentScan，@Configuration三者的集合,需要的朋友可以參考下2023-08-08
maven項(xiàng)目使用指定JDK版本打包的實(shí)現(xiàn)步驟
在使用多個(gè)JDK版本環(huán)境打包過(guò)程中,有時(shí)候懶得配置環(huán)境變量,使用maven打包的時(shí)候,會(huì)報(bào)版本對(duì)應(yīng)不上,本文就來(lái)介紹一下maven項(xiàng)目使用指定JDK版本打包的實(shí)現(xiàn)步驟,感興趣的可以了解一下2026-02-02
帶你詳細(xì)了解Spring Security的注解方式開發(fā)
這篇文章主要介紹了詳解spring security四種實(shí)現(xiàn)方式，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-08-08
idea2020中復(fù)制一個(gè)微服務(wù)實(shí)例的方法
這篇文章主要介紹了idea2020中復(fù)制一個(gè)微服務(wù)實(shí)例的方法,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-09-09
maven無(wú)法依賴spring-cloud-stater-zipkin的解決方案
這篇文章主要介紹了maven無(wú)法依賴spring-cloud-stater-zipkin如何解決,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2023-05-05
java.imageIo給圖片添加水印的實(shí)現(xiàn)代碼
最近項(xiàng)目在做一個(gè)商城項(xiàng)目, 項(xiàng)目上的圖片要添加水?、?添加圖片水印;②:添加文字水印;一下提供下個(gè)方法,希望大家可以用得著2013-07-07
這篇文章主要介紹了SpringBoot實(shí)體多層嵌套如何判空字段,最近在公司了接了個(gè)需求：需要開發(fā)一個(gè)中間系統(tǒng),進(jìn)行三方聯(lián)調(diào),文中通過(guò)代碼示例介紹的非常詳細(xì),需要的朋友可以參考下2024-09-09
解析Apache Dubbo的SPI實(shí)現(xiàn)機(jī)制
SPI全稱為Service Provider Interface，對(duì)應(yīng)中文為服務(wù)發(fā)現(xiàn)機(jī)制。SPI類似一種可插拔機(jī)制，首先需要定義一個(gè)接口或一個(gè)約定，然后不同的場(chǎng)景可以對(duì)其進(jìn)行實(shí)現(xiàn)，調(diào)用方在使用的時(shí)候無(wú)需過(guò)多關(guān)注具體的實(shí)現(xiàn)細(xì)節(jié)2021-06-06
這篇文章主要介紹了Java之格式化輸出過(guò)程,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2026-04-04