从使用者痛点来看。消息丢失的隐形杀手

• List/Set 队列LPUSH/RPOP 或 RPUSH/LPOP 的消息一旦被弹出，就算未处理完成也会从队列中移除。怎么说呢，服务崩溃后重启时这些未处理的消息永远无法恢复。

• Pub/Sub 订阅发布者推送消息时若订阅者不在线。消息直接丢失，没有任何补救机制。

  

• 传统数据库表队列虽然能保存消息。但需要额外开发状态机、锁定机制等复杂逻辑，且性能难以匹敌专业队列程序。

. Stream 主要命令全解析

. 基础写入与读取

XADD key * field value

向 Stream 中追加一条消息。 其实，* 表示自动生成 ID。也可手动指定 ID，

XREAD COUNT num STREAMS key id

- COUNT num: 每次最多返回 num 条消息。- $: 从最新未读开始；- : 阻塞等待新消息。

. 使用者组模式

. 创建/管理组：

XGROUP CREATE key groupId $ MKSTREAM CONSUMERID auto | manual_id XGROUP SETID key groupId id XGROUP DELCONSUMER key groupId consumer_id XGROUP DELGROUP key groupId . 消费流程关键命令：

• . 获取待确认的偏移量：
•  XREADGROUP GROUP COUNT STREAMS >
• . ACK确认或NACK失败标记：
•  XACK <$id>
•  XCLAIM <$id> -1 -1 true true
• . 检查积压与偏移量问题：
• ▶ XLEN streamkey - XINFO JUSTIFY streamkey - XPENDING stream_key group

. Python实战案例：建立高可靠日志收集程序

. 生产者端代码

python import redis

r = redis.Redis

def produce_log: # 自动生成ID并追加到stream r.xadd

. 使用者端代码

python def consumelogs: while True: 说到try。# 阻塞获取最新1条未处理过的日志 msg = r.xreadgroup('MY-GROUP','consumer-001',streams={'logstream': '>'},count=1,block=500)

 if msg:
msg_id,data = next))
print
# 模拟处理...
time.sleep
# ACK确认已处理
r.xack
except Exception as e:
print

. 高级调整技巧

场景挑战	Stream方法
大量短连接导致连接池耗尽	使用Redis Cluster分片或Pipeline批量操作减少网络开销

积压太多导致内存爆炸

- XTRIM MAXLEN/MINID 自动清理旧数据
- 按天分流程降低单Stream压力

跨区域部署延迟高

- Redis复制组保持本地读写
- Raft协议同步替代单纯异步复制

严格顺序要求场景

- 不使用consumer groups
- 自定义+循环检查机制

<="">