4. Redis Streams 详解：Stream 数据结构、消费者组、消息持久化与 PUB/SUB 对比

Redis Streams 是 Redis 5.0 引入的核心数据结构，专为消息队列与事件流场景设计。它弥补了 PUB/SUB 无持久化、无回溯消费的短板，同时提供了比 List 更丰富的消费者组模型。本节从数据结构、消费者组、持久化机制三个维度深入剖析，并与 PUB/SUB 做系统性对比。

4.1 Stream 数据结构核心模型

Stream 本质上是一个仅追加的日志结构，每条消息由唯一的 ID 和 键值对列表组成。ID 默认基于毫秒时间戳和序列号生成，保证全局有序且唯一。

# 添加消息到 Stream
XADD mystream * sensor-id 1234 temperature 19.8
# 输出: "1712345678000-0"

# 读取最新消息（阻塞模式）
XREAD COUNT 1 BLOCK 5000 STREAMS mystream 0

# 按时间范围查询
XRANGE mystream 1712345678000 1712345679000

4.2 消费者组（Consumer Group）机制

消费者组是 Stream 最强大的特性，它允许多个消费者协同消费同一个消息流，同时保证消息不丢失、负载均衡和故障转移。

# 创建消费者组
XGROUP CREATE mystream mygroup $

# 消费者 A 读取消息（阻塞模式）
XREADGROUP GROUP mygroup consumerA COUNT 1 BLOCK 5000 STREAMS mystream >

# 消费者 B 读取消息
XREADGROUP GROUP mygroup consumerB COUNT 1 BLOCK 5000 STREAMS mystream >

# 确认消息
XACK mystream mygroup 1712345678000-0

# 查看待处理消息
XPENDING mystream mygroup

4.3 消息持久化与可靠性保障

Stream 的消息持久化依赖 Redis 的 RDB 和 AOF 持久化机制。与 PUB/SUB 不同，Stream 的消息写入后即持久化到磁盘，重启后可恢复。

4.4 Streams 与 PUB/SUB 的全面对比

PUB/SUB 是经典的发布-订阅模式，而 Streams 是消息队列与事件流的融合体。两者在消息模型、持久化、消费方式上存在本质差异。

// PUB/SUB 示例（无持久化）
SUBSCRIBE channel:alerts
// 发布者
PUBLISH channel:alerts "Server down!"

// Stream 示例（有持久化 + 消费者组）
XADD stream:alerts * event "server_down" host "web-01"
XREADGROUP GROUP alert-group worker1 STREAMS stream:alerts >

4.5 实践案例：使用 Streams 构建可靠任务队列

以下示例展示如何使用 Streams 实现一个生产者-消费者任务队列，包含失败重试和死信处理。

// 生产者：添加任务
XADD task:queue * type "email" to "user@example.com" subject "Welcome"

// 消费者组：创建组
XGROUP CREATE task:queue workers $

// 消费者：循环处理任务
while (true) {
    // 阻塞读取任务
    result = XREADGROUP GROUP workers worker1 COUNT 1 BLOCK 5000 STREAMS task:queue >
    if (result) {
        message = result[0][1][0]
        id = message[0]
        data = message[1]
        
        // 处理任务
        success = processTask(data)
        
        if (success) {
            // 确认消息
            XACK task:queue workers id
        } else {
            // 检查重试次数
            pending = XPENDING task:queue workers - + 10
            if (pending[id].deliveryCount > 3) {
                // 移入死信队列
                XADD task:queue:dead * original_id id error "max_retries"
                XACK task:queue workers id
            }
            // 否则等待 XCLAIM 重新投递
        }
    }
}