3、Redis Stream：Stream数据结构、消费者组、消息持久化、ACK机制
3.1 为什么需要Stream？
说实话，在Redis 5.0之前，消息队列这块一直是个痛点。
有人用List做队列，但消息确认得自己写。有人用Pub/Sub，但消息不持久化，一断连就全丢了。还有人用ZSet，但那是给排序用的，硬拿来当队列用，总感觉别扭。
我早期做游戏服务器时，就踩过Pub/Sub的坑。玩家下线再上线，中间发的消息全没了。那会儿我还纳闷，怎么排行榜数据对不上？查了半天才发现是Pub/Sub的锅。
Redis Stream的出现，说白了就是官方终于给出了一个正经的消息队列方案。它既有Kafka的消费者组概念，又有Redis一贯的简洁风格。
3.2 Stream数据结构长什么样？
Stream本质上是一个日志数据结构。每条消息都有一个唯一的ID，消息内容是键值对的形式。
你可以把它想象成一个只追加的日志文件。新消息永远加在末尾，老消息不会修改。
# 添加消息
XADD mystream * name "player1" score 100
# 读取消息
XRANGE mystream - +
# 获取Stream长度
XLEN mystream这里有个细节——消息ID。如果你用*，Redis会自动生成一个基于时间戳的ID，格式是时间戳-序号。比如1688888888888-0。
我个人习惯用自动生成的ID。为什么？因为时间戳天然有序，而且能避免ID冲突。你想想看，如果多个客户端同时发消息，手动指定ID很容易出问题。
3.3 消费者组——这才是核心
消费者组是Stream最强大的特性。它解决了两个关键问题：
- 消息分发：一条消息只被组内一个消费者处理
- 消息确认：处理完的消息需要ACK，否则会重新投递
我在项目中遇到过这样一个场景：游戏里有公会战，每场战斗会产生大量事件。如果用传统队列，一个消费者处理不过来。用消费者组，我可以开10个消费者并行处理，每条消息只分配给其中一个。
# 创建消费者组
XGROUP CREATE mystream mygroup 0
# 从消费者组读取消息
XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 5000 STREAMS mystream >注意那个>符号。它表示「只读取从未被投递过的消息」。如果你指定具体的消息ID，那就是读取历史消息。
重要概念：消费者组里的每个消费者，都有一个独立的待处理列表。消息投递后，会进入消费者的待处理列表（Pending List），直到你ACK它。
3.4 消息持久化——别担心丢消息
Stream的消息是持久化在Redis内存中的。但光靠内存不够，万一宕机怎么办？
Redis的RDB和AOF持久化机制，同样适用于Stream。你配置好AOF，就算机器重启，Stream里的消息也不会丢。
不过要注意一点：Stream的消息不会自动删除。你得自己管理消息的生命周期。
# 删除Stream中指定ID的消息
XDEL mystream 1688888888888-0
# 裁剪Stream，只保留最新的N条消息
XTRIM mystream MAXLEN 10000我曾经犯过一个错误——生产环境里Stream越积越大，最后占了几十G内存。排查时发现，是忘记做XTRIM了。嗯，这里要提醒你，一定要设置合理的MAXLEN。
警告：Stream的消息数量过多会占用大量内存。建议根据业务场景，定期裁剪或设置最大长度。游戏服务器里，我一般保留最近1小时的消息就够了。
3.5 ACK机制——确保消息不丢失
ACK（Acknowledgment）是消息队列的基石。消费者处理完消息后，必须显式地告诉Redis：这条消息我处理好了。
# 确认消息处理完成
XACK mystream mygroup 1688888888888-0如果不ACK，会发生什么？
消息会一直留在消费者的待处理列表里。下次你用XREADGROUP时，如果指定了BLOCK参数，Redis会优先投递待处理的消息。
说白了，ACK机制保证了「至少一次投递」。就算消费者在处理消息时崩溃了，重启后还能重新消费那些未ACK的消息。
小技巧：用XPENDING命令可以查看待处理的消息列表。我经常用它来排查消息积压的问题。
# 查看待处理消息
XPENDING mystream mygroup
# 查看某个消费者的待处理消息
XPENDING mystream mygroup - + 10 consumer13.6 实战：游戏服务器里的应用
拿我们游戏里的邮件系统举例。玩家完成任务后，系统要发奖励邮件。这个场景用Stream再合适不过。
// 生产者：任务完成时发送邮件消息
XADD mail_queue * user_id "10001" title "任务奖励" content "恭喜获得100金币"
// 消费者组：多个邮件服务实例并行处理
XGROUP CREATE mail_queue mail_group 0
// 消费者：处理邮件发送
XREADGROUP GROUP mail_group mail_worker_1 COUNT 1 BLOCK 5000 STREAMS mail_queue >这样做的好处很明显：
- 邮件发送慢？加几个消费者就行
- 某个消费者挂了？消息还在，重启后继续处理
- 想重试失败的消息？用XPENDING查出来，重新投递
3.7 避坑指南
最后分享几个我踩过的坑：
- 不要用Stream做广播：消费者组的设计是「一条消息只被一个消费者处理」。如果你需要广播，用Pub/Sub。
- 注意消息积压：消费者处理速度跟不上生产速度时，Stream会越来越大。监控XLEN，设置告警。
- ACK要及时：处理完消息立刻ACK，不要等。否则待处理列表会越来越大，影响性能。
我曾经有个项目，因为忘记ACK，待处理列表里积压了上百万条消息。每次重启消费者，都要重新处理这些历史消息，导致系统卡顿了好久。从那以后，我写代码时都会在try-finally里加上XACK。
好了，Stream的内容就讲到这里。下一章我们聊聊如何用Redis实现分布式锁，这可是游戏服务器里绕不开的话题。