★ FEATURED ARTICLE 1. 场景与选型先说说我为什么折腾这个。之前项目里要处理订单事件下单之后要发通知、扣库存、更新积分、生成报表。全都同步做响应时间直接爆炸。直接调一堆服务耦合度惨不忍睹下游一挂主流程跟着遭殃。后来想上消息队列但团队本来就维护着一套 Redis为了几个消息再搭一套 Kafka 或者 RabbitMQ运维成本受不了。后来试了试 Redis 从 5.0 带来的 Stream发现这玩意儿比想象中能打。日志采集也是类似场景。业务日志量不小要异步写到 ES 或者对象存储。日志允许延迟但绝对不能丢。Redis Stream 有 AOF 和 RDB 持久化加上消费确认机制重启之后消息还能捞回来正好满足。有人可能问Kafka 吞吐高、RabbitMQ 可靠为啥不用说实话Kafka 要管 Zookeeper现在 KRaft 也好点RabbitMQ 也得单独部署。对于中小规模系统Redis 本来就是基础设施Stream 相当于送的。单实例跑个十万级消息/秒没问题业务量不大完全够用。你要是每天几千万订单那还是老老实实上 Kafka别拿 Redis 硬扛。2. Redis Stream 的核心概念Redis Stream 本质上是一个日志结构的数据类型内部实现不是简单的 listpack而是用 listpack 存储消息数据用基数树rax来索引消息 ID这样按 ID 范围查询就非常高效。每条消息有一个自动生成的递增 ID通常长这样1700000000000-0。几个必须搞明白的概念Entry一条消息包含 ID 和一堆 field-value 对。Consumer Group消费组。同一个 stream 可以被多个消费组独立消费每个组内部一条消息只能被组内的一个消费者拿到。PELPending Entries List待确认消息列表。消费者读了一条消息这条消息的 ID 就会进到该消费者的 PEL直到它显式发XACK才会被移除。last_delivered_id消费组当前投递到的最后一个消息 ID。XREADGROUP消费组读消息的命令用表示只读新消息或者指定一个 ID 从 PEL 里读历史消息。XACK确认消息处理完成。PEL 是可靠性的核心可以说没有 PELStream 和普通发布订阅就没本质区别了。3. Spring Boot 接入封装一个 StreamTemplateSpring Data Redis 从 2.3 开始支持 Stream但 API 藏在opsForStream()里用起来比较啰嗦。我习惯自己包一层把发送、消费、确认、重投这些操作收敛到一个类里业务代码看着干净些。3.1 依赖和配置Spring Boot 2.7 直接用spring-boot-starter-data-redis里面带了 Spring Data Redis 3.xStream 支持已经很完整。dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependencyapplication.yml里常规配置就行单机版spring:data:redis:host:localhostport:6379timeout:5slettuce:pool:max-active:16max-idle:83.2 StreamTemplate 代码我直接贴代码注释我写清楚省得回头忘了。ComponentpublicclassStreamTemplate{privatefinalStringRedisTemplateredisTemplate;publicStreamTemplate(StringRedisTemplateredisTemplate){this.redisTemplateredisTemplate;}/** * 发送消息XADD * param key stream名称 * param map 消息字段 * return 消息ID */publicStringsend(Stringkey,MapString,Stringmap){returnredisTemplate.opsForStream().add(key,map);}/** * 创建消费组已存在就忽略 * 注意默认从当前最新位置开始消费如果要从头消费 * 可以用 createGroup(key, group, ReadOffset.from(0)) */publicvoidcreateGroup(Stringkey,Stringgroup){try{redisTemplate.opsForStream().createGroup(key,group);}catch(RedisSystemExceptione){// 组已存在会抛 BUSYGROUP这个不算错误if(!e.getMessage().contains(BUSYGROUP)){throwe;}}}/** * 读新消息XREADGROUP ... * param key stream名 * param group 消费组 * param consumer 消费者名最好用 主机名随机数 保证唯一 * param count 最多读多少条 */publicListMapRecordString,Object,ObjectreadNew(Stringkey,Stringgroup,Stringconsumer,intcount){returnredisTemplate.opsForStream().read(Consumer.from(group,consumer),StreamReadOptions.empty().count(count).block(Duration.ofSeconds(5)),StreamOffset.create(key,ReadOffset.latest())// 记住用 latest()代表 );}/** * 读PEL里的历史消息未确认的XREADGROUP ... 0 * 用于重启后先捞之前的遗留消息 */publicListMapRecordString,Object,ObjectreadPending(Stringkey,Stringgroup,Stringconsumer,intcount){returnredisTemplate.opsForStream().read(Consumer.from(group,consumer),StreamReadOptions.empty().count(count),StreamOffset.create(key,ReadOffset.from(0)));}/** * 确认消息XACK */publicvoidack(Stringkey,Stringgroup,String...recordIds){redisTemplate.opsForStream().acknowledge(key,group,recordIds);}/** * 查看消费组的pending信息 */publicPendingSummarypendingSummary(Stringkey,Stringgroup){returnredisTemplate.opsForStream().pending(key,group);}}注意几个坑ReadOffset.latest()对应的是命令里的只读新消息。ReadOffset.from(0)才是从 PEL 里捞。之前看有人用lastConsumed()那是拿当前消费者处理的最后一条 ID用来定位不是读新消息用的。createGroup默认从$当前最新开始如果想从头消费得用重载版本传ReadOffset.from(0)。如果 Redis 是集群版的Stream 的 key 本身会被分片消费组的状态跟着 key 走。但跨节点搞消费者组不行所以大规模场景得用多个 Stream 分片或者干脆让位给 Kafka。3.3 发消息订单事件下面是一个订单服务的发布者ServicepublicclassOrderEventPublisher{privatefinalStreamTemplatestreamTemplate;publicOrderEventPublisher(StreamTemplatestreamTemplate){this.streamTemplatestreamTemplate;}publicvoidpublishOrderCreated(Orderorder){MapString,StringeventnewHashMap();event.put(eventId,UUID.randomUUID().toString());event.put(type,ORDER_CREATED);event.put(orderId,order.getId());event.put(userId,order.getUserId());event.put(amount,order.getAmount().toPlainString());event.put(timestamp,String.valueOf(System.currentTimeMillis()));StringmessageIdstreamTemplate.send(order:events,event);log.info(订单事件发布成功messageId{},messageId);}}发送的时候不用指定 ID让 Redis 自己生成天然有序。3.4 消费消息定时拉取 手动 ACK消费者我用了个最简单的方案定时任务拉取。生产环境你要是追求实时性可以换成线程池 while(true)阻塞读但逻辑差不多。ComponentpublicclassOrderEventConsumer{privatestaticfinalStringSTREAM_KEYorder:events;privatestaticfinalStringGROUPorder-consume-group;privatefinalStreamTemplatestreamTemplate;privatefinalOrderEventHandlerorderEventHandler;publicOrderEventConsumer(StreamTemplatestreamTemplate,OrderEventHandlerorderEventHandler){this.streamTemplatestreamTemplate;this.orderEventHandlerorderEventHandler;}PostConstructpublicvoidinit(){streamTemplate.createGroup(STREAM_KEY,GROUP);}Scheduled(fixedDelay1000)publicvoidconsumeNewMessages(){// 消费者名字必须保持唯一否则同一实例重启后会被当成不同消费者StringconsumerString.format(%s-%d,InetAddress.getLocalHost().getHostName(),UUID.randomUUID().getMostSignificantBits()Long.MAX_VALUE);ListMapRecordString,Object,ObjectrecordsstreamTemplate.readNew(STREAM_KEY,GROUP,consumer,10);for(MapRecordString,Object,Objectrecord:records){try{booleansuccessorderEventHandler.handle(record.getValue());if(success){streamTemplate.ack(STREAM_KEY,GROUP,record.getId().getValue());}else{// 业务处理明确返回失败说明这消息本身可能有问题丢死信然后 ack否则会卡在 PEL 里sendToDeadLetter(record,business failure);streamTemplate.ack(STREAM_KEY,GROUP,record.getId().getValue());}}catch(Exceptione){// 没 ack消息留在 PEL等会儿再捞log.error(消费异常暂不确认ID{},record.getId(),e);}}}}关键点只有XACK了Redis 才认为这条消息处理完成了。假如消费过程中抛了异常别 ack消息就会留在 PEL 里之后你能通过XAUTOCLAIM或者手动读 PEL 重新捞起来。4. 消费组里的竞争和负载均衡同一个 Stream 能被多个消费组独立消费互不干扰。一个组里可以注册多个消费者。那 Redis 怎么分配消息呢我觉得**“竞争”这个词比“分区”更贴切**。Kafka 是把 Topic 分成多个 Partition每个消费者负责几个分区这是静态的。Redis 消费组里多个消费者同时执行XREADGROUP ... 的时候Redis 内部会维护一个last_delivered_id按顺序把新消息逐个分给当前正在等待的消费者。每个消息只给一个消费者不会重复。谁有本事谁先抢不是严格轮询但整体上是均衡的。比如酒店预订系统三个实例都跑着相同的消费者代码消费者名各不相同redis 就会把消息分发到不同的消费者实现扩展。想提高吞吐增加消费者实例就行。不过有一点要注意Redis 的分配不会像 Kafka 那样做 rebalance如果某个消费者处理特别慢它也没办法把它的负载分担出去。所以设计消费任务时尽量保证耗时相近或者干脆别搞太重的业务逻辑毕竟 Stream 不是批处理引擎。5. 消息幂等去重避不开的坑正常情况下一条消息只会在一个消费者的 PEL 里挂着。但是消费者故障转移的时候消息可能被XAUTOCLAIM分配到另一个消费者重新处理。所以业务侧必须做幂等。最简单的做法是拿消息里的业务 ID比如 eventId去重。可以用 Redis 的 SetpublicbooleanisProcessed(StringeventId){returnredisTemplate.opsForSet().isMember(processed:event:ids,eventId);}publicvoidmarkProcessed(StringeventId){redisTemplate.opsForSet().add(processed:event:ids,eventId);}不过 Redis 的 Set 有丢失风险不是数据丢失是可能被淘汰更稳妥的方式是业务表加唯一索引。比如消费订单事件插入一张order_event表event_id 设为主键。先插数据如果主键冲突说明处理过了直接 ack。TransactionalpublicbooleanhandleOrderCreated(MapString,Stringdata){StringeventIddata.get(eventId);try{eventMapper.insertEvent(eventId,data);}catch(DuplicateKeyExceptione){returntrue;// 已经处理过没必要再处理}// 真正的业务逻辑...returntrue;}我的建议是Redis Set 做前置过滤数据库唯一键兜底两者结合效果最好。如果你只在乎不丢不在乎重复那上面都不用了。6. 可靠性PEL、XAUTOCLAIM 和死信6.1 PEL 到底干了啥消费者读取消息后消息 ID 会进入消费者的 PEL。PEL 保存在 Redis 里所以消费者即使宕机了消息也不会丢只是没人处理。要恢复就得靠重投。6.2 消费者重启后先处理 PEL假设消费者 A 处理消息到一半突然宕机了。重启后它自己内存里的东西全没了但 Redis 里 PEL 还在。这时候可以先用readPending方法把ID0开始的历史消息读出来重新处理。注意这要求业务处理必须幂等——之前可能已经处理一半了所以重来不能出事。6.3 XAUTOCLAIM自动转移超时消息如果消费者 A 一直没起来PEL 里的消息就得有人接管。Redis 6.2 开始提供了XAUTOCLAIM命令可以自动把空闲时间超过指定阈值的消息从别的消费者名下抢过来同时递增投递次数。Spring Data Redis 提供了封装publicListMapRecordString,Object,ObjectautoClaim(Stringkey,Stringgroup,Stringconsumer,DurationminIdleTime,longcount){returnredisTemplate.opsForStream().autoClaim(key,group,consumer,ClaimOptions.minIdle(minIdleTime),StreamReadOptions.empty().count(count)).getRecords();}定时任务认领Scheduled(cron0 0/1 * * * ?)publicvoidclaimTimeoutMessages(){Stringconsumerclaim-worker-instanceId;ListMapRecordString,Object,ObjectrecordsstreamTemplate.autoClaim(STREAM_KEY,GROUP,consumer,Duration.ofMinutes(5),100);for(MapRecordString,Object,Objectrecord:records){// 照常处理然后 ACK}}这里要注意autoClaim返回的记录里可能会有投递次数信息取决于版本你可以拿出来判断是不是该扔进死信队列了。6.4 死信队列死信队列DLQ是个老话题。Redis Stream 本身没有内置死信概念需要自己建一个 Stream 专门接收处理失败的消息。不过怎么判断一条消息是否该死信我一般维护一个计数比如用 Redis 的 hashStringattemptKeyattempt:record.getId().getValue();LongattemptredisTemplate.opsForValue().increment(attemptKey);if(attempt5){sendToDeadLetter(record);streamTemplate.ack(STREAM_KEY,GROUP,record.getId().getValue());redisTemplate.delete(attemptKey);}投递次数上限自己定超过就 ack 掉然后丢进order:events:dead-letter。这样既不会无限重试也不会阻塞组里的新消息。7. 消费者故障和集群下的可靠性7.1 消费者实例崩溃就一个消费者崩溃了只要其他消费者还在那 PEL 里的消息依赖XAUTOCLAIM就能被接管。这里有个前提业务处理必须无状态、幂等否则可能会出岔子。7.2 Redis 主从切换Redis Sentinel 或者 Cluster 模式Stream 数据会随着主从复制到从节点上消费组的状态也会同步。主节点宕机后从节点晋升Stream 数据不会丢——前提是你开了 AOF且配置合理。我一般建议appendfsync everysec最多丢一秒可以接受如果非要严格不丢用always但性能会打折。另外客户端这边 Spring Data Redis 的 Lettuce 会自动感知拓扑变化你只需要在操作时做好异常重试就行。7.3 Redis Cluster 下的 StreamRedis Cluster 里一个 Stream 对应一个 key会分片到某个 slot。也就是说一个 Stream 的所有消息都在同一个节点上没法像 Kafka 那样分区到多台机器。想要横向扩展只能手动建多个 Stream比如order:events:0、order:events:1然后客户端按订单号 hash 路由。这样搞确实麻烦所以如果有数据量大到单节点撑不住建议直接换 Kafka别在 Redis 上死磕。7.4 优雅停机消费者服务要重启时最好不要直接强杀。在PreDestroy里把运行的标志位设为 false让消费循环自然退出。此时有些消息可能已经被拉下来了但还没处理完别 ack留在 PEL 里别的消费者会通过XAUTOCLAIM把它们接走。这样基本可以做到“零丢失”。8. 跟 RabbitMQ、Kafka 的选型对比网上有太多表格我不爱做表说点实在的。Redis Stream就是轻量级的选择运维成本最低性能也够用。缺点是没有死信队列、延迟队列这些内置组件都得自己造轮子而且消息是存在内存里的虽然可以持久化但内存成本高。适合消息量不大十万级/秒以下、团队不想维护额外中间件的场景。RabbitMQ是老牌专业选手路由灵活死信、延迟队列都是现成的可靠性很高。不过吞吐量受限于 Erlang VM一般也就万级/秒做企业级业务集成很合适。运维要管 broker、vhost还有镜像队列比 Redis 麻烦很多。Kafka就不用多说了百万级/秒的吞吐分区顺序消息回溯流式处理都是它的强项。但运维成本高要管理 Zookeeper 或者 KRaft还得调参数一般不打算上大数据场景根本没必要惹它。用一句话概括如果你手里已经有 Redis而且业务消息量不大那就用 Stream 顶上去别额外造出一个要运维的“动物园”来。9. 监控和容量那些事上了 Stream 之后你得盯着点它的健康状态线上出了积压可不是闹着玩的。消息积压XLEN看看 Stream 里有多少条消息。如果积压数量一直上涨说明消费者来不及处理。消费组状态用XINFO GROUPS看每个消费组的未确认消息数就是 PEL 里的数量这个数字越大说明消费者越不给力。消费者状态XINFO CONSUMERS可以查看每个消费者 PEL 大小和空闲时间空闲时间过长说明这个消费者可能挂了。Redis 本身关注内存使用、AOF rewrite 频率别让 Redis 因为内存淘汰把消息给删了maxmemory-policy建议设置成noeviction宁可直接报错也不能丢消息。如果你接了 Prometheusredis_exporter 会把 Stream 这些指标都导出来自己写几条告警规则。比如-alert:RedisStreamMsgBacklogexpr:redis_stream_consumers_lag1000for:5m容量估算一条消息按 200 字节算每天 1000 万条就是 2GB 左右加上 PEL 的元数据大概 3GB。如果 Redis 内存紧张先想想业务上能不能剪裁消息格式或者是不是该换 Kafka 了。10. 收个尾Redis Stream 这东西本质上是用 Redis 的数据结构模拟了一套可靠消息机制。它不像 Kafka 那样宏大也不像 RabbitMQ 那样“专业”但在很多业务场景下它就是最合适的那个——前提是你已经会 Redis并且不想为了几个队列再增加一套运维负担。集成到 Spring Boot 也不难自己封装一个StreamTemplate就能用得很顺手。但你要记住三个原则第一业务必须幂等别拿“不会重复”当不做幂等的借口第二持久化要开 AOF别拿“Redis 很快”来对抗宕机第三盯住 PEL 的积压和消费者的空闲时间别等问题爆发了才想起来看监控。代码我就不贴完整工程了上面这些足够你起步。要是真用起来遇到什么玄学问题欢迎评论区聊聊。官网https://farerboy.com/
阅读完成 · 觉得有帮助？
