计数服务 + BufferTrigger 批量聚合——千条点赞消息，一次 Redis 写入

UGC 产品里到处都是计数笔记点赞数、关注数、粉丝数、收藏数。流量一来同一篇笔记一秒被几百人点赞逐条HINCRBY能把 Redis 和 DB 全打满。BufferTrigger 给计数服务加了一个内存缓冲层——零散的计数消息进来先攒着攒够一批按 noteId 分组聚合最后一次性写入。千条消息变几次 Redis 操作DB 写入量也降两个数量级。计数服务是什么为什么单独拆出来以笔记点赞为例。用户点一下赞数据库里要记一条点赞记录同时笔记的点赞总数要 1。点赞总数这个数字如果每次都SELECT COUNT(*) FROM note_like WHERE note_id 100——一篇文章几千个赞每次都要扫全表DB 一秒就挂。所以计数被单独抽成了一个计数服务。整个链路是这样① 发 MQ② 消费③ HINCRBY用户点赞业务服务RocketMQ消息: noteId100, like1计数服务Redis Hashnote:100 → {like: 280}④ 前端读点赞数业务操作触发 MQ计数服务消费 MQ在 Redis 里做原子加减。前端要展示点赞数的时候直接从 Redis 读不碰数据库。这个设计在低流量下一点问题没有——来一条消费一条HINCRBY一下完事。问题出在流量上来的时候。热门笔记一秒可能被几百人同时点赞MQ 里瞬间堆积几百条noteId100, like1。如果计数服务来一条就HINCRBY一次Redis计数服务RocketMQRedis计数服务RocketMQ... 同一个 key 被操作了 500 次 ...noteId100, like1HINCRBY note:100 like 1noteId100, like1HINCRBY note:100 like 1noteId100, like1HINCRBY note:100 like 1Redis 本身不慢但 500 次独立的网络往返是白花的。更严重的是这 500 条消息后面还要逐条写 DB 落库——500 次 SQL 和一次UPDATE SET like like 500结果一样前者把连接池打满后者瞬间完成。所以关键不是计数服务本身而是计数服务消费消息的时候怎么聚合。同一个 key 的多次操作应该合并成一次——这句话就是 BufferTrigger 的全部价值。BufferTrigger 怎么干的——攒一批聚合一次写BufferTriggercom.github.phantomthief:buffer-trigger就是一个内存缓冲层。你往里面enqueue(item)它帮你攒着满足条件时自动调你的消费函数把这一批Listitem交给你处理。攒满 1000 条或超过 1 秒enqueue(item)内存缓冲池触发 flushconsumeMessage(List)̨分组 → 聚合 → 一次写 Redis三个控制条件参数含义batchSize攒够多少条就 flushlinger超过多长时间没攒够也 flushbufferSize缓冲池最多存多少条超过按模式处理阻塞 / 丢弃 / 无限回到点赞的场景。没有 BufferTrigger 的时候计数服务来一条写一次——500 次 Redis 操作。加了 BufferTrigger 之后RedisBufferTriggerRocketMQRedisBufferTriggerRocketMQ触发 flush按 noteId 分组聚合noteId100: 300noteId200: 200500 次变 2 次noteId100, like1noteId100, like1noteId200, like1...攒了 500 条HINCRBY note:100 like 300HINCRBY note:200 like 200真实代码——从构建到聚合到落库依赖dependencygroupIdcom.github.phantomthief/groupIdartifactIdbuffer-trigger/artifactIdversion0.2.21/version/dependencyConsumer 端完整写法ComponentRocketMQMessageListener(consumerGroupgroup_count_note_like,topicMQConstants.TOPIC_COUNT_NOTE_LIKE)Slf4jpublicclassCountNoteLikeConsumerimplementsRocketMQListenerString{ResourceprivateRedisTemplateString,ObjectredisTemplate;ResourceprivateRocketMQTemplaterocketMQTemplate;// ① 构建 BufferTriggerprivateBufferTriggerStringbufferTriggerBufferTrigger.StringbatchBlocking().bufferSize(50000)// 最多攒 5 万条.batchSize(1000)// 满 1000 条就 flush.linger(Duration.ofSeconds(1))// 不满 1000 条1 秒后也 flush.setConsumerEx(this::consumeMessage)// flush 时调用这个方法.build();OverridepublicvoidonMessage(Stringbody){// ② 收到消息不直接处理扔进 BufferTriggerbufferTrigger.enqueue(body);}/** * ③ BufferTrigger 攒够一批后回调——在这里做聚合和写入 */privatevoidconsumeMessage(ListStringbodys){log.info( 聚合消息, size: {},bodys.size());// 3.1 JSON → DTOListCountLikeUnlikeNoteMqDTOdtosbodys.stream().map(body-JsonUtils.parseObject(body,CountLikeUnlikeNoteMqDTO.class)).toList();// 3.2 按 noteId 分组MapLong,ListCountLikeUnlikeNoteMqDTOgroupMapdtos.stream().collect(Collectors.groupingBy(CountLikeUnlikeNoteMqDTO::getNoteId));// 3.3 每组聚合点赞 1取消点赞 -1算出净变化ListAggregationCountLikeUnlikeNoteMqDTOresultLists.newArrayList();for(varentry:groupMap.entrySet()){LongnoteIdentry.getKey();LongcreatorIdnull;intfinalCount0;for(vardto:entry.getValue()){creatorIddto.getNoteCreatorId();finalCountdto.getType().equals(LikeUnlikeNoteTypeEnum.LIKE.getCode())?1:-1;}result.add(AggregationCountLikeUnlikeNoteMqDTO.builder().noteId(noteId).creatorId(creatorId).count(finalCount).build());}// 3.4 聚合后的结果一次写 Redis只写几个 key不再是几百次result.forEach(item-{StringkeyRedisKeyConstants.buildCountNoteKey(item.getNoteId());if(redisTemplate.hasKey(key)){redisTemplate.opsForHash().increment(key,likeTotal,item.getCount());}});// 3.5 聚合后的批量落库同样是批量 MQ 消费MessageStringmessageMessageBuilder.withPayload(JsonUtils.toJsonString(result)).build();rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE_2_DB,message,callback);}}数据流enqueue × 500触发 flushRocketMQ 推送 500 条BufferTriggerconsumeMessage(List)按 noteId 分组聚合noteId100: 300, -20 → 净 280noteId200: 150, -10 → 净 140Redis HINCRBY × 2MQ → 计数落库批量500 条原始消息聚合后只操作了 2 个 Redis key落库也变成批量。一条链路上每一环都被优化了。BufferTrigger 的三种构建模式// 阻塞模式队列满了生产者等——一条都不想丢BufferTrigger.StringbatchBlocking().bufferSize(50000).batchSize(1000).linger(Duration.ofSeconds(1)).setConsumerEx(this::consumeMessage).build();// 非阻塞模式队列满了丢弃新来的——可以接受少量丢失BufferTrigger.Stringbatch().bufferSize(50000).batchSize(1000).linger(Duration.ofSeconds(1)).setConsumerEx(this::consumeMessage).build();// 简单模式无界队列——生产速度可控BufferTrigger.Stringsimple().batchSize(1000).linger(Duration.ofSeconds(1)).setConsumerEx(this::consumeMessage).build();MQ 消费端用batchBlocking——消息不能丢。计数服务也一样。什么时候该用、什么时候不该用该用同一批 key 被频繁操作能合并——点赞、收藏、关注等计数场景可以接受秒级批量延迟——用户不关心点赞数晚一秒更新下游存储扛不住高频写入不该用每条消息必须立刻生效——支付回调、库存扣减消息量本身就小攒不满攒着反而增加延迟没有聚合空间——每条消息操作的是完全不同的 key