有不少团队把Redis当消息队列用最早都是从List起步的一队业务方LPUSH扔消息另外几个消费者BRPOP捞消息跑起来简单粗暴前期确实省了不少中间件选型的事。但等业务稍微复杂一点消息要分组消费、要确认重试、要查看积压情况的时候List那套就有点撑不住了。Redis 5.0里引入的Stream本质就是Redis官方给消息队列场景设计的数据结构既保留了Redis一贯的轻量、低延迟又把消费组、消息确认、Pending重投这些真正生产级的能力补齐了。这篇文章我不打算给你念文档而是按照实际项目里怎么接入、怎么踩坑、怎么调优的路子把Stream从头到尾掰开讲一遍适合正在用Redis做异步消息、又不想为了一个轻量队列引入整套Kafka/RabbitMQ重组件的团队。1. 用List做队列时那些说不出口的痛点在没有Stream的日子里我见过不少项目组用List实现队列最典型的是LPUSH配BRPOP组合。数据先塞进List尾部消费者用阻塞的BRPOP从头部弹出看起来丝般顺滑直到线上出现第一单事故。1.1 取出即删除的“丢消息”隐患BRPOP弹出消息的同时消息就从List里消失了。如果这个消费者进程在拿到消息之后、处理业务之前突然崩溃这条消息就再也找不回来了。后来有人想到用BRPOPLPUSH先把消息转到一个备份List里处理成功后再从备份List里LREM删掉等于手工实现了一个“保险箱”。这确实能缓解消息丢失但代价是每一条消息的处理流程都要额外操作两次Redis而且LREM还得按值精确匹配要是消息内容带特殊情况删除逻辑就非常别扭。说白了List队列本身没有“未确认消息”的概念每条消息被弹出去就默认为已消费服务端不保留任何痕迹。1.2 消费组和重复消费全靠自己造业务升级后一个队列往往需要多个消费者分担压力。大家不约而同地用上了分布式锁拿到锁的消费者从List里取消息处理完再释放锁。逻辑上没问题但分布式锁本身有一堆边界问题比如锁过期、线程阻塞、客户端失联一旦锁意外释放同一个时刻会有两个消费者同时取到同一条消息。更麻烦的是List没有“组内游标”的概念也没有ACK机制你想知道某个消费者到底消费到哪条了几乎没法从Redis层面拿到精确答案只能把自己业务表里的处理状态捞出来反推。这些本应是消息队列的基本能力在List上都要从零搭。1.3 排序、延迟消息和消息回溯的无力感List本质上是一个线性表天然按入队顺序排列只能严格FIFO。想做一个优先级队列想实现一个延迟队列要么加多个List然后按优先级依次BRPOP要么引入ZSet按时间戳排序再轮询逻辑一多就乱了。反过来如果某个消费者落后了你想让它从某个时间点重新消费历史消息List也完全做不到因为消息消费完就从结构上删除了。我印象特别深的一次同事想把出错的消息从备份队列里恢复出来发现数据都还在但已经分辨不出先后顺序最后只能人工一条条挑。整个链路上Redis只是一个“临时暂存器”队列管理和消息追溯靠的是业务代码的自我修养这种模式时间一长必然出问题。所以Stream的出现不是锦上添花而是把原来散落在业务代码里的各种补丁和妥协收敛成一套原生数据结构。2. Stream的底层结构XADD写入与XREAD读取到底做了什么Stream一上来给人的观感像Redis里多了一张“追加日志表”每条消息都有一个单调递增的ID还有一堆键值对形式的字段。但它的内部并不真是表而是经过专门设计的紧凑结构。2.1 认识Stream的三个核心维度要理解Stream先抓住三个东西消息ID、消息内容、消费位置。消息ID默认是毫秒时间戳-序号的格式比如1751311234567-0Redis保证在同一个Stream中严格递增且唯一。消息ID允许客户端自定义只要比当前最大ID大就行。消息内容一条消息是一组field-value对类似一个小Hash结构。生产者和消费者通过约定字段名来解析业务数据。消费位置每个消费者组各自维护一个游标游标就是一条消息ID。XREAD按游标取后面没读过的消息XACK则是把已处理完的消息标记为完成。这三个维度把“日志结构”“数据载荷”“消费进度”完全解耦了。拿到一条消息你既可以用它做普通的点对点通信也可以把它交给多个消费组分别处理各组互不干扰。2.2 XADD写入时ID生成的两种方式XADD往Stream里追加一条消息最简单的写法是XADD order:events * orderId 20250101001 eventType NOTIFY这里的*表示让Redis自动生成消息ID。自动ID用的是服务器本地的毫秒时间戳加序号同一毫秒内可以追加多条消息序号从0开始自增因此哪怕同一毫秒也能保证全局有序。上游如果是多实例并发写入时间戳都由Redis自身决定不会出现客户端时钟偏移导致的乱序问题。如果你想用业务侧的时间或流水号过来也可以显式指定ID比如XADD order:events 1751311234500-0 orderId ...但前提是这个ID必须大于Stream当前最大ID。显式ID最大的好处是方便在业务表里做消息幂等去重你拿到一条消息就能按消息ID去查本地记录。不过也没必要每个项目都这样做大多数人直接用*就够了。补充一个细节XADD带着MAXLEN参数可以限制Stream长度比如XADD order:events MAXLEN 10000 * orderId xxxStream超过1万条后就自动挤出最早的旧消息避免无界增长撑爆内存。这个参数后面讲运维时再展开。2.3 XREAD读取的三种姿势与阻塞机制写入是追加读取就是按ID扫过去了。XREAD命令的基础语法是XREAD COUNT 10 BLOCK 5000 STREAMS order:events 0这句的意思是从order:events这条流里从ID0开始往后读最多读10条如果没有新消息就最多阻塞5000毫秒再返回。这里有几个关键语义起始ID填0表示从头开始消费所有消息填$表示只读未来新增的消息。{这里需要注意0和$都是XREAD的规则不代表具体消息ID}BLOCK参数的单位是毫秒BLOCK 0表示无限期阻塞直到有消息进入。XREAD一次可以同时读多个StreamXREAD STREAMS queueA queueB 0 0Redis会从每个Stream各自游标位置向后取消息按Stream数据量分发。XREAD的使用场景更像“我自己来控制消费位置”适合做消息导出、数据备份、任意时间戳回放。但生产环境的消费者通常不会直接用XREAD因为根本没有“组”的概念只有XREADGROUP才能真正分摊消息。3. 消费组机制从XREADGROUP到XACK的完整消息生命周期Stream和List最本质的区别在于它自带Consumer Group机制。一个Stream下可以挂多个消费组每个组里有多个消费者一条消息投递给某个组时组内只会有一个消费者吃到这条消息。多个消费组之间则各自维护独立游标相当于同一个事件可以被不同业务模块分别完整处理。这套模型和Kafka的消费组很接近理解起来不费劲。3.1 创建消费组与首次读取的游标规则通过XGROUP CREATE命令创建一个消费组XGROUP CREATE order:events notifyGroup 0第三个参数是起始消息ID。传0表示这个新组从头开始消费这个Stream里留存的历史消息如果只想消费创建组之后的新消息就传$。我建议这个参数要按业务想清楚再定。我见过一个同学把组起始ID直接填$以为这样省事结果上游已经产生了大量待处理数据这些消息全都被跳过了等发现时已经造成业务延迟。如果一开始不确定宁可先填0在测试环境看全量消息再决定。创建好组之后消费者用XREADGROUP读取XREADGROUP GROUP notifyGroup consumerA COUNT 10 STREAMS order:events 注意STREAMS后面的这个它表示“从该组当前游标向后分配新消息”。这里有几个容易绕晕的地方是XREADGROUP专用的特殊位置告诉Redis我不要手动指定起始ID而是按组内进度来。如果你在这里填了一个具体ID那语义会变成查看Pending消息而不是取新消息后面排错时经常搞混。3.2 Pending机制一条消息从投递到确认的完整旅程消费者从组里读到一组消息后消息并没有被删除而是进入该消费者的Pending Entries List简称PEL。PEL记录的是“已投递但还没收到ACK”的消息。业务流程处理完一条消息后务必用XACK把它标记为完成XACK order:events notifyGroup 1751311234567-0只有执行了XACKRedis才会把这条消息从该组的PEL中移除。如果消费者处理过程中宕机、抛异常、一直没有ACK这条消息会一直留在PEL里并且投递计数增加。经历过List队列的朋友看到这里应该很感慨原来在List里用BRPOPLPUSH手工实现的备份重试逻辑现在变成了一等公民的基础能力。每条消息的“已投递未确认”状态都在Redis里可见谁消费了、哪条卡住了、卡了多久全部有据可查。3.3 XCLAIM与XAUTOCLAIM异常消息的重新分配PEL里挂着的消息需要被重新投递。传统做法是XCLAIM把某条指定ID的消息从一个消费者名下转移给另一个消费者。XCLAIM order:events notifyGroup consumerB 60000 1751311234567-0这个命令表示把我指定的消息ID转移给consumerB前提是这条消息在队列里闲置了至少60000毫秒也就是1分钟。转移成功后新的消费者就可以重新处理这条消息。Redis 6.2开始多了XAUTOCLAIM命令可以自动扫描PEL中超时的消息并批量转移给当前消费者。对于我这种懒人XAUTOCLAIM的感知价值很大因为不用逐一列举消息ID一条命令就把一批“死消息”捞出来重投XAUTOCLAIM order:events notifyGroup consumerB 60000 0这里的60000是闲置时间阈值0表示从该组PEL的起始位置开始扫描。3.4 消费语义定位与幂等性要求对Stream的消息语义业内普遍的定位是“at least once”而不是“exactly once”。原因很简单XACK和业务处理不是原子的。你处理完了业务但在执行XACK之前进程崩溃这条消息就会被重新投递另一个消费者会再处理一次。反过来如果XACK先执行而业务后提交一旦业务提交失败消息已经确认完成也就不会重发了。所以一个铁律是消费端必须对消息做幂等处理。最常用的办法是拿业务唯一键查一次本地表如果已经处理过就不再重复执行。在我的项目里通常会在XACK之前多做一步“处理结果落库”把消息ID或者业务单号作为唯一键写进去即使收到重复消息也能直接判断为已完成。这个幂等设计是Stream落地的必要条件绝不是可选项。4. 生产环境接入案例消费者宕机与消息积压的完整排查链路前面讲了一堆命令和概念现在落到一个具体项目里看看怎么用。这套逻辑我在一个模拟项目X里实践过业务场景是用户在商城下了订单后系统要异步给用户发积分、发短信通知。原来我们是两个独立线程池各干各的后来把这两个动作统一放进一个订单事件Stream里由消费组负责分发处理。4.1 生产端与消费端代码落地生产端比较直接订单创建成功后向Stream里写一条事件MapString, String body new HashMap(); body.put(orderId, orderId); body.put(userId, userId); body.put(eventType, ORDER_CREATED); redisTemplate.opsForStream() .add(StreamRecords.mapBacked(body) .withStreamKey(order:events));消费端用独立的消费组跑循环。为了方便说明我用Spring Data Redis的StreamOperations来写一个简化版while (running) { ListMapRecordString, Object, Object records redisTemplate .opsForStream() .read(Consumer.from(notifyGroup, consumerA), StreamReadOptions.empty().count(20).block(Duration.ofSeconds(5)), StreamOffset.create(order:events, ReadOffset.lastConsumed())); for (MapRecordString, Object, Object record : records) { try { String orderId String.valueOf(record.getValue().get(orderId)); // 幂等校验本地处理记录表查一下 if (alreadyProcessed(orderId)) { redisTemplate.opsForStream().acknowledge(order:events, notifyGroup, record.getId()); continue; } sendPointsToUser(orderId); sendSmsToUser(orderId); saveProcessedRecord(orderId); redisTemplate.opsForStream().acknowledge(order:events, notifyGroup, record.getId()); } catch (Exception e) { // 不Ack消息会留在Pending里等待后续重新投递 log.error(consume error, messageId: {}, record.getId(), e); } } }大概结构就是循环拉消息、处理逻辑、成功ACK、失败不ACK。跑了一段时间这套框架确实稳定没有出现List模式里那种莫名其妙的乱序和重复竞争。4.2 线上告警消费延迟半小时Pending持续上升上线后第三周监控突然报警订单通知的消费延迟超过30分钟Redis里order:events的长度不断上涨。我先看XPENDING确认是不是有消息没有被确认XPENDING order:events notifyGroup返回结果里显示Pending消息有305条其中有一条消息闲置时间超过1900秒。当时直觉判断是某个消费者实例出了问题。再用XINFO查看消费组里各个消费者的状态XINFO CONSUMERS order:events notifyGroup结果发现有两个消费者consumerA的pending数是5条consumerB的pending数是300条。问题基本锁定在consumerB上它接收了大量消息但既没有处理完也没有触发重平衡。查服务日志发现consumerB实例因为外部短信接口超时线程全部卡在等待响应上导致消息处理循环没法继续ACK。4.3 用XCLAIM做消息转移与恢复要恢复积压最直接的办法是把consumerB名下Pending超时的消息转移给consumerA。我先把消费者B的实例摘掉然后用XAUTOCLAIM把超时消息批量捞回来XAUTOCLAIM order:events notifyGroup consumerA 60000 0这条命令把闲置超过1分钟的消息从consumerB名下转移到consumerA名下转移成功后consumerA会继续消费这些消息并执行处理流程。转移消息本身不会自动执行XACK只是一次重新投递因此我在消费代码里的幂等校验又兜住了一遍底不会被重复处理影响业务。类似这样的问题我还遇到过更隐蔽的消费者自己没有宕机但因为没有设置合理的max-delivery-count某些毒消息被无限次重新投递每次投递时刚好卡在异常点等于死循环。Redis Stream没有像一些商业MQ那样内置“最大消费次数”的概念必须自己在XCLAIM/XAUTOCLAIM拉取消息时对delivery count做判断超过阈值就丢进DLT队列留人工处理。这个逻辑看着简单但线上真正遇到毒消息时没有它是真的烧Redis CPU和内存。整个过程走完把消费者B修复后重新接回组里积压量很快就消化了。这个排错链路给我最大的感受是因为所有消息状态都能通过XPENDING、XINFO看到故障定位不是“猜”而是照着数据一步步跟踪。5. 运维视角Stream的backlog淘汰、内存控制与日常监控Stream写起来容易但如果不做限制它就是一个会无限增长的结构迟早把Redis内存吃穿。运维上的核心命题是消息在Stream里留多久积压到什么程度该报警怎么把积压清理掉。5.1 用MAXLEN和MINID控制消息保留Stream的裁剪方式主要有两种按长度裁剪或者按消息ID实际是按时间裁剪。按长度裁剪XTRIM order:events MAXLEN ~ 10000MAXLEN后面的~表示允许近似裁剪。若不写~Redis会精确裁到10000条代价是逐条删除性能在超大Stream上非常差写上~之后Redis按节点单位批量删除效率高很多但裁剪后的长度可能稍微超过设定的值。日常使用中只要不是对精确条数有硬性要求一律建议加~。按时间裁剪XTRIM order:events MINID 1751311234567MINID表示删除掉所有小于指定消息ID的旧消息。配合自动ID格式这个消息ID的前半段就是毫秒时间戳。所以你可以直接用“当前时间减去保留窗口”算出最小的ID执行一条命令就把超时的历史事件全部清掉。5.2 内存模型与节点配置Stream的底层在Redis内部是由listpack节点和rax索引组成的可持久化结构。简单理解它把邻近的消息打包进一个个小节点每个节点默认最多存放100条消息或大约100字节的键值数据这两个参数分别由stream-node-max-entries和stream-node-max-bytes控制。消息写多了节点会不断分裂形成类似有序链表的组织方式。h这个设计带来的直观影响是消息量较少时一个节点打包几百条消息内存占用非常紧凑消息大量积压之后节点数量变多拆分和索引维护的额外开销上升。所以日常使用中不建议把一个Stream无限拉长更不要把它当成永久事件存储。把消息生命周期控制在业务可接受的范围内比如保留3天或7天才是合理的姿势。5.3 监控指标与告警阈值Stream相关的监控不建议只看内存几个更有价值的指标是Pending数量用XPENDING查看当前组内未确认消息数如果持续上涨且下不去说明有消费者处理不过来。队列长度用XLEN查看当前Stream总消息数和业务量曲线对比就能看出消费是否跟上生产。消费延迟取最新一条消息ID中的时间戳和当前时间做差得到大致消费延迟。延迟超过业务容忍阈值就告警。我自己的监控配置大概是pending数连续5分钟超过1000就告警队列长度超过设定上限触发XTRIM预案延迟超过10分钟就进入人工排查流程。不要把阈值拍脑袋乱定一定要根据平时正常波动留出足够余量不然半夜一个误报够你受的。5.4 持久化与故障切换的几个提醒Stream当然会走Redis的RDB和AOF持久化宕机重启后消息还在。但有几个细节需要注意因为Stream是追加型结构AOF持久化的重写频率会直接影响恢复速度建议在业务允许的情况下开启合理的AOF rewrite策略。另外使用Redis Stream时如果部署模式是主从或者分片集群消费者组和Pending状态是保存在单个Redis节点上的副节点升级为主节点后Stream数据和新追加的消息不会丢失但正在执行的消费游标可能因为主从切换的时间差产生少量重复投递。消息重复的概率确实存在再加上前面说的幂等设计两者配合才能放心。6. Stream与Kafka、RabbitMQ的选型取舍Redis Stream出来之后消息队列的选型又多了一个选项。它不是要替代Kafka和RabbitMQ而是填补了轻量队列到重型MQ之间的空隙。下面的对比表是我根据自己项目中的实际使用体会整理的不能代替你压测但能帮你少走弯路。维度Redis StreamKafkaRabbitMQRedis Pub/Sub消息持久化随Redis持久化磁盘日志保留可配置持久化队列不持久化客户端断开即丢消费确认有Pending和XACK机制有offset提交机制有ACK和requeue机制无消息回溯支持按范围读取和重新消费支持按offset回溯有死信和重新入队不支持多消费组原生支持原生支持通过交换机绑定广播模式无组概念典型吞吐量单节点中等低延迟高吞吐大数据中高吞吐路由灵活较高但不保证可靠性运维复杂度极低随Redis走高需维护集群中等略显笨重极低适用场景微服务内部异步事件、短时积压、低延迟通知日志管道、大规模事件驱动、实时数仓复杂路由、多协议、企业服务总线实时广播、弱业务一致性场景具体场景怎么选我的建议是这样的如果业务规模不大团队成员对Redis已经很熟系统内部只需要一个“能保证至少有一个人处理这条消息”的异步通道Stream是非常合适的选择。比如订单通知、积分变动、缓存刷新这类消息量不会爆炸积压期限也不需要太长运维上只要管Redis本身就够了。如果系统对消息有明确的跨团队契约、需要严格的分区顺序、希望消息在磁盘上保留很久用于事故复盘或者每秒写入了上百万条事件Kafka依然是首选。Kafka的partition机制、批量压缩传输能力是Stream目前无法替代的。如果业务上有复杂的路由关系一份消息要根据不同交换机发给不同队列RabbitMQ更顺手。它的协议生态和插件体系也成熟得多适合做跨语言企业系统集成。而Redis Pub/Sub只适合“发出去听不听到随缘”的场景比如前端页面收到一条新订单提示它不需要可靠消费订阅者离线也不该留消息这种情况下Pub/Sub反而更轻。网格再回到Stream哪怕是在生产环境我也不会把一个Stream当作无限缓冲池来用。它更适合作为业务系统里一条“快速且可控”的异步支线承接不大不小的事件流。消息在Stream里待多久、积压多少、会不会重复这些用一个成熟的消费组方案加上指数退避重试就能解决不需要引入一整套新的中间件基础设施。最后再分享一个小技巧Stream的消费代码里一定要把XACK和“处理结果落库”设计成同一个事务单元而不是分开随便调。先写业务结果表再执行XACK能最大程度降低“处理成功但消息没确认”的模糊区。这个顺序我调整之前系统总是出现一些莫名其妙的重复任务调整之后再配合XAUTOCLAIM把真正的超时消息捞回来重试消息队列这块基本不用再人肉救火了。
