如何基于Kafka与Redis实现按消息数或时长阈值的消息延时聚合处理

你之前选型Redis pub/sub不符合需求的核心原因是其为无持久化的即时广播模型，无订阅者时消息会直接丢失，且不支持消息堆积缓存，推荐以下两种落地性高的实现方案：

方案1：基于Redis的轻量实现（适合中小流量、快速迭代场景） #

• 每个用户单独维护一个Redis List结构作为专属消息缓冲队列，key规则设为mq:buffer:user:{user_id}
• 每写入1条消息到对应用户List后，调用LLEN命令获取当前队列长度：
  • 如果长度刚好等于20，直接弹出全部20条消息，执行聚合处理后删除该缓冲key
  • 如果长度不足20，调用EXPIRE命令的NX参数给该key设置剩余存活时间为x秒，避免重复覆盖已经设置的过期时间
• 额外配套一个间隔小于x秒的后台定时扫描任务：
  • 用SCAN命令扫描所有mq:buffer:user:前缀的key，筛选出TTL小于等于0的过期key
  • 弹出这些key对应List的所有消息，执行聚合处理后删除key

该方案实现成本极低，Redis本身的持久化特性可以保证缓冲消息不丢失，适合大部分业务场景。

方案2：基于Kafka的实现（适合高流量、高可靠要求场景） #

• 先给Kafka配置按用户id分区的策略，保证同一个用户的所有消息都落到同一个分区，被同一个消费者实例处理
• 消费者侧本地维护每个用户的缓冲数组，同时给每个用户维护两个状态：当前消息计数、窗口起始时间
• 每消费1条消息放到对应用户的缓冲数组后做两个判断：
  • 如果数组长度达到20，立即执行聚合处理，清空该用户的缓冲数组和相关状态
  • 如果数组长度不足20，启动一个本地异步延时任务（可以用对应语言的DelayQueue类实现），延时时间为x秒，延时任务触发时如果数组还有未处理的消息，直接执行聚合处理后清空缓冲

该方案完全基于现有Kafka组件，不需要额外引入其他依赖，配合Kafka手动提交offset机制，只有缓冲内的消息处理完成后才提交对应offset，可以保证消息零丢失。

内容的提问来源于stack exchange，提问作者Haniel Bitton