380 lines
9.3 KiB
Markdown
380 lines
9.3 KiB
Markdown
# Spring Boot Redis Stream Demo
|
||
|
||
这是一个基于 Spring Boot 和 Redis Stream 的端到端消息收发演示项目，实现了生产者→Redis Stream→消费者的完整流程。
|
||
|
||
## 功能特性
|
||
|
||
- ✅ 使用 Redis 5+ 的 Stream 数据结构（非 Pub/Sub）
|
||
- ✅ 支持两种消费者模式：
|
||
- **StreamListener 模式**：自动监听，实时处理，事件驱动
|
||
- **Manual Ack 模式**：手动拉取，按需处理，精确控制
|
||
- ✅ 完整的单元测试和集成测试
|
||
- ✅ REST API 接口用于测试
|
||
- ✅ 支持批量消息发送（100条消息）
|
||
- ✅ 消息确认机制（ACK）
|
||
- ✅ 消费者组支持
|
||
- ✅ 错误处理和重试机制
|
||
- ✅ 监控和统计功能
|
||
- ✅ 灵活的配置管理
|
||
|
||
## 技术栈
|
||
|
||
- **Spring Boot 2.7.18**
|
||
- **Spring Data Redis**
|
||
- **Lettuce Redis Client**
|
||
- **Jackson JSON**
|
||
- **Lombok**
|
||
- **Testcontainers** (用于集成测试)
|
||
- **Redis 7+**
|
||
|
||
## 项目结构
|
||
|
||
```
|
||
src/
|
||
├── main/
|
||
│ ├── java/com/example/
|
||
│ │ ├── config/ # 配置类
|
||
│ │ │ ├── RedisStreamConfig.java
|
||
│ │ │ └── RedisStreamProperties.java
|
||
│ │ ├── controller/ # REST 控制器
|
||
│ │ │ └── StreamController.java
|
||
│ │ ├── model/ # 数据模型
|
||
│ │ │ └── Message.java
|
||
│ │ ├── service/ # 业务服务
|
||
│ │ │ ├── MessageProducerService.java
|
||
│ │ │ ├── StreamListenerConsumerService.java
|
||
│ │ │ └── ManualAckConsumerService.java
|
||
│ │ └── RedisStreamApplication.java
|
||
│ └── resources/
|
||
│ └── application.yml
|
||
└── test/
|
||
└── java/com/example/
|
||
├── RedisStreamIntegrationTest.java
|
||
└── service/
|
||
└── MessageProducerServiceTest.java
|
||
```
|
||
|
||
## 快速开始
|
||
|
||
### 1. 环境要求
|
||
|
||
- Java 8+
|
||
- Maven 3.6+
|
||
- Redis 5+ (推荐 Redis 7+)
|
||
|
||
### 2. 启动 Redis
|
||
|
||
```bash
|
||
# 使用 Docker 启动 Redis
|
||
docker run -d --name redis -p 6379:6379 redis:7-alpine
|
||
|
||
# 或使用本地 Redis
|
||
redis-server
|
||
```
|
||
|
||
### 3. 运行应用
|
||
|
||
```bash
|
||
# 编译项目
|
||
mvn clean compile
|
||
|
||
# 运行应用
|
||
mvn spring-boot:run
|
||
```
|
||
|
||
应用将在 `http://localhost:8080` 启动。
|
||
|
||
### 4. 运行测试
|
||
|
||
#### 方式一：使用测试脚本（推荐）
|
||
|
||
```bash
|
||
# Windows
|
||
test-simple.bat
|
||
|
||
# Linux/Mac
|
||
chmod +x test.sh
|
||
./test.sh
|
||
```
|
||
|
||
#### 方式二：手动测试
|
||
|
||
```bash
|
||
# 1. 发送100条消息
|
||
curl -X POST "http://localhost:8080/api/stream/send-batch?count=100"
|
||
|
||
# 2. 使用Manual Ack消费消息
|
||
curl -X POST "http://localhost:8080/api/stream/consume-manual?count=100"
|
||
|
||
# 3. 查看Stream信息
|
||
curl http://localhost:8080/api/stream/info
|
||
```
|
||
|
||
#### 方式三：运行单元测试
|
||
|
||
```bash
|
||
# 运行所有测试（需要本地Redis）
|
||
mvn test
|
||
|
||
# 运行集成测试
|
||
mvn test -Dtest=RedisStreamIntegrationTest
|
||
|
||
# 运行单元测试
|
||
mvn test -Dtest=MessageProducerServiceTest
|
||
```
|
||
|
||
## API 接口
|
||
|
||
### 发送消息
|
||
|
||
```bash
|
||
# 发送单条消息
|
||
curl -X POST http://localhost:8080/api/stream/send \
|
||
-H "Content-Type: application/json" \
|
||
-d '{
|
||
"content": "测试消息",
|
||
"type": "TEST",
|
||
"sender": "test-user"
|
||
}'
|
||
|
||
# 批量发送 100 条消息
|
||
curl -X POST "http://localhost:8080/api/stream/send-batch?count=100"
|
||
```
|
||
|
||
### 消费消息
|
||
|
||
```bash
|
||
# 使用 Manual Ack 模式消费 10 条消息
|
||
curl -X POST "http://localhost:8080/api/stream/consume-manual?count=10"
|
||
|
||
# 拉取单条消息（Manual Ack 模式）
|
||
curl http://localhost:8080/api/stream/poll-single
|
||
|
||
# 处理单条消息（Manual Ack 模式）
|
||
curl -X POST http://localhost:8080/api/stream/process-single?recordId=1234567890-0
|
||
|
||
# 控制 StreamListener 状态
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/start # 启动
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/stop # 停止
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/status # 状态
|
||
```
|
||
|
||
### 查看信息
|
||
|
||
```bash
|
||
# 获取 Stream 信息
|
||
curl http://localhost:8080/api/stream/info
|
||
|
||
# 健康检查
|
||
curl http://localhost:8080/api/stream/health
|
||
```
|
||
|
||
### 管理操作
|
||
|
||
```bash
|
||
# 重置计数器
|
||
curl -X POST http://localhost:8080/api/stream/reset
|
||
|
||
# 清空 Stream
|
||
curl -X POST http://localhost:8080/api/stream/clear
|
||
```
|
||
|
||
## 两种消费者模式详解
|
||
|
||
### 1. StreamListener 模式
|
||
|
||
**特点：**
|
||
- 自动监听 Redis Stream，实时接收消息
|
||
- 基于事件驱动的消息处理
|
||
- 自动确认消息（ACK）
|
||
- 支持消费者组和负载均衡
|
||
- 适合高并发、实时性要求高的场景
|
||
|
||
**使用方式：**
|
||
```java
|
||
// 自动启动监听（应用启动时）
|
||
@PostConstruct
|
||
public void init() {
|
||
streamListenerConsumerService.startListening();
|
||
}
|
||
|
||
// 手动控制监听状态
|
||
streamListenerConsumerService.startListening(); // 启动
|
||
streamListenerConsumerService.stopListening(); // 停止
|
||
boolean isListening = streamListenerConsumerService.isListening(); // 检查状态
|
||
```
|
||
|
||
**API 接口：**
|
||
```bash
|
||
# 控制 StreamListener 状态
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/start # 启动
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/stop # 停止
|
||
curl -X POST http://localhost:8080/api/stream/stream-listener/status # 状态
|
||
```
|
||
|
||
### 2. Manual Ack 模式
|
||
|
||
**特点：**
|
||
- 手动拉取消息，按需处理
|
||
- 完全控制消息确认时机
|
||
- 支持批量处理
|
||
- 适合需要精确控制处理流程的场景
|
||
- 可以实现复杂的重试和错误处理逻辑
|
||
|
||
**使用方式：**
|
||
```java
|
||
// 拉取单条消息
|
||
MapRecord<String, Object, Object> message = manualAckConsumerService.pollSingleMessage();
|
||
if (message != null) {
|
||
boolean success = manualAckConsumerService.processSingleMessage(message);
|
||
}
|
||
|
||
// 批量处理消息
|
||
manualAckConsumerService.batchProcessMessages(10);
|
||
|
||
// 拉取并处理消息
|
||
manualAckConsumerService.pollAndProcessMessages();
|
||
```
|
||
|
||
**API 接口：**
|
||
```bash
|
||
# 拉取单条消息
|
||
curl http://localhost:8080/api/stream/poll-single
|
||
|
||
# 处理单条消息
|
||
curl -X POST http://localhost:8080/api/stream/process-single?recordId=1234567890-0
|
||
|
||
# 批量处理消息
|
||
curl -X POST http://localhost:8080/api/stream/consume-manual?count=10
|
||
```
|
||
|
||
## 核心组件说明
|
||
|
||
### 1. MessageProducerService
|
||
|
||
消息生产者服务，负责向 Redis Stream 发送消息：
|
||
|
||
- `sendMessage(Message message)`: 发送单条消息
|
||
- `sendBatchMessages(int count)`: 批量发送消息
|
||
- `getStreamLength()`: 获取 Stream 长度
|
||
- `clearStream()`: 清空 Stream
|
||
|
||
### 2. StreamListenerConsumerService
|
||
|
||
基于 StreamListener 的消费者服务：
|
||
|
||
- 实现 `StreamListener` 接口
|
||
- 自动监听 Redis Stream 中的新消息
|
||
- 自动处理消息并发送 ACK
|
||
- 提供消息统计功能
|
||
- 支持启动/停止监听控制
|
||
- 包含错误处理和重试机制
|
||
|
||
### 3. ManualAckConsumerService
|
||
|
||
基于手动拉取的消费者服务：
|
||
|
||
- `pollAndProcessMessages()`: 手动拉取并处理消息
|
||
- `pollSingleMessage()`: 拉取单条消息
|
||
- `processSingleMessage()`: 处理单条消息
|
||
- `batchProcessMessages(int count)`: 批量处理指定数量的消息
|
||
- 手动发送 ACK 确认
|
||
- 提供消息统计功能
|
||
- 支持处理状态检查
|
||
|
||
### 4. RedisStreamConfig
|
||
|
||
Redis Stream 配置类：
|
||
|
||
- 配置 RedisTemplate
|
||
- 创建 StreamMessageListenerContainer
|
||
- 管理监听容器的生命周期
|
||
|
||
## 测试用例
|
||
|
||
### 集成测试 (RedisStreamIntegrationTest)
|
||
|
||
- `testStreamListenerEndToEnd()`: StreamListener 端到端测试
|
||
- `testManualAckEndToEnd()`: Manual Ack 端到端测试
|
||
- `testSingleMessageProductionAndConsumption()`: 单条消息测试
|
||
- `testConcurrentProductionAndConsumption()`: 并发测试
|
||
- `testStreamLengthAndClear()`: Stream 管理测试
|
||
|
||
### 单元测试 (MessageProducerServiceTest)
|
||
|
||
- `testSendSingleMessage()`: 单条消息发送测试
|
||
- `testSendBatchMessages()`: 批量消息发送测试
|
||
- `testSendLargeBatchMessages()`: 大批量消息测试
|
||
- `testClearStream()`: Stream 清空测试
|
||
- `testGetStreamLength()`: Stream 长度测试
|
||
|
||
## 配置说明
|
||
|
||
### application.yml
|
||
|
||
```yaml
|
||
spring:
|
||
redis:
|
||
host: localhost
|
||
port: 6379
|
||
database: 0
|
||
|
||
redis:
|
||
stream:
|
||
key: "message-stream"
|
||
consumer-group: "message-consumer-group"
|
||
consumer-name: "message-consumer"
|
||
```
|
||
|
||
## 运行示例
|
||
|
||
### 1. 启动应用
|
||
|
||
```bash
|
||
mvn spring-boot:run
|
||
```
|
||
|
||
### 2. 发送 100 条消息
|
||
|
||
```bash
|
||
curl -X POST "http://localhost:8080/api/stream/send-batch?count=100"
|
||
```
|
||
|
||
### 3. 查看 Stream 信息
|
||
|
||
```bash
|
||
curl http://localhost:8080/api/stream/info
|
||
```
|
||
|
||
### 4. 使用 Manual Ack 消费消息
|
||
|
||
```bash
|
||
curl -X POST "http://localhost:8080/api/stream/consume-manual?count=50"
|
||
```
|
||
|
||
### 5. 运行测试验证
|
||
|
||
```bash
|
||
mvn test -Dtest=RedisStreamIntegrationTest#testStreamListenerEndToEnd
|
||
```
|
||
|
||
## 注意事项
|
||
|
||
1. **Redis 版本**: 需要 Redis 5.0+ 支持 Stream 功能
|
||
2. **消费者组**: 项目使用消费者组模式，确保消息的可靠传递
|
||
3. **ACK 机制**: 两种消费者模式都实现了手动 ACK，确保消息处理完成
|
||
4. **并发安全**: 使用 AtomicLong 确保计数器的线程安全
|
||
5. **资源管理**: 监听容器会在应用关闭时自动清理资源
|
||
|
||
## 扩展建议
|
||
|
||
1. **消息持久化**: 可以添加消息持久化到数据库
|
||
2. **死信队列**: 实现消息处理失败的死信队列机制
|
||
3. **监控指标**: 集成 Micrometer 提供监控指标
|
||
4. **消息重试**: 实现消息处理失败的重试机制
|
||
5. **集群支持**: 支持 Redis 集群模式
|
||
|
||
## 许可证
|
||
|
||
MIT License
|