dev

Author: chan-max

src/queue/IMPLEMENTATION.md

@@ -0,0 +1,114 @@
+# 消息队列实现方案总结
+
+## 实现概述
+
+提供了**两种实现方案**，系统会自动检测 Redis 版本并选择最佳实现：
+
+### 方案 1: QueueService（基于 List + Sorted Set）
+- **Redis 版本要求**: 2.0+
+- **数据结构**: List + Sorted Set + String
+- **特点**: 兼容性好，实现简单
+- **适用场景**: Redis 版本较低或需求简单
+
+### 方案 2: QueueStreamsService（基于 Redis Streams）
+- **Redis 版本要求**: 5.0+
+- **数据结构**: Stream + Sorted Set（延迟消息）
+- **特点**: 功能丰富，原生支持消费者组和 ACK
+- **适用场景**: Redis 5.0+，需要更强大的功能
+
+## 自动选择机制
+
+系统通过 `QueueFactoryService` 自动检测：
+
+1. **启动时检查** Redis 版本
+2. **测试** Streams 功能是否可用
+3. **自动选择** 最佳实现
+
+## 使用方式
+
+### 方式 1: 直接使用 QueueService（推荐）
+
+```typescript
+import { QueueService } from '../queue/queue.service';
+
+@Injectable()
+export class YourService {
+  constructor(private readonly queueService: QueueService) {}
+  
+  async someMethod() {
+    // 生产消息
+    await this.queueService.produce('queue-name', 'type', data);
+    
+    // 消费消息
+    const messages = await this.queueService.consume('queue-name');
+    
+    // 确认消息
+    await this.queueService.ack('queue-name', messageId);
+  }
+}
+```
+
+系统会自动使用最佳实现，你无需关心底层细节。
+
+### 方式 2: 使用 QueueFactoryService
+
+```typescript
+import { QueueFactoryService } from '../queue/queue-factory.service';
+
+@Injectable()
+export class YourService {
+  constructor(private readonly factory: QueueFactoryService) {}
+  
+  async someMethod() {
+    const queueService = await this.factory.getService();
+    const isUsingStreams = this.factory.isUsingStreams();
+    
+    // 使用 queueService...
+  }
+}
+```
+
+## Redis Streams 命令说明
+
+如果使用 Streams 实现，底层使用以下 Redis 命令：
+
+- **XADD**: 添加消息到 Stream
+- **XREADGROUP**: 从消费者组读取消息
+- **XACK**: 确认消息处理完成
+- **XPENDING**: 查看 Pending List
+- **XDEL**: 删除消息
+- **XLEN**: 获取 Stream 长度
+
+## 数据结构对比
+
+### List 实现
+```
+queue:{queue}                    - List/Sorted Set，存储消息ID
+queue:delay:{queue}              - Sorted Set，延迟消息
+queue:processing:{queue}        - List，处理中消息
+queue:message:{queue}:{id}      - String，消息详情
+```
+
+### Streams 实现
+```
+stream:{queue}                   - Stream，消息流
+queue:delay:{queue}              - Sorted Set，延迟消息
+queue:message:{queue}:{id}      - String，消息详情（用于查询）
+```
+
+## 性能对比
+
+| 操作 | List 实现 | Streams 实现 |
+|------|-----------|--------------|
+| 生产消息 | O(1) | O(1) |
+| 消费消息 | O(1) | O(1) |
+| ACK | O(1) | O(1) |
+| 查询 Pending | O(N) | O(1) |
+| 消费者组 | 需手动实现 | 原生支持 |
+
+## 推荐使用
+
+- **直接注入 QueueService**：系统会自动选择最佳实现
+- **接口完全兼容**：两种实现接口一致，可以无缝切换
+- **无需关心底层**：系统会自动处理版本检测和实现选择
+