当前位置：首页 > 问答 > 正文

用Redis做消息订阅去重，避免重复消息那点事儿讲讲

根据常见的Redis使用场景和技术社区讨论,如CSDN、博客园、Stack Overflow等平台上的相关话题，以及《Redis实战》等书籍中的思想，结合个人理解进行阐述）

今天咱们来聊一个挺实际的问题：用Redis做消息订阅的时候，怎么避免处理重复的消息，这事儿听起来好像挺简单的，不就是别让同一条消息被处理两次嘛，但真做起来，里头还是有些门道值得琢磨的。

为啥会有重复消息呢？

咱们想象一个常见的场景：你有一个消息生产者，比如一个下单系统，每当用户成功下单，它就往一个Redis的频道（Channel）里发布一条消息，说“订单123创建成功了”，另一边，有一个或多个消息消费者，比如库存扣减服务、发货服务，它们订阅了这个频道，一看到这条消息，就各自去干自己的活儿。

那重复消息可能从哪儿来呢？主要有这么几个地方：

生产者这边手抖了：比如下单系统可能因为网络问题，以为第一次发布没成功（其实可能已经成功了），于是又重试了一次，发布了完全一样的消息，这在微服务架构里挺常见的。
消费者这边出状况了：消费者服务确实收到了消息，也处理完了（比如库存都扣了），但是在它还没来得及告诉消息系统“我搞定啦”（比如确认ACK机制）的时候，消费者服务自己突然崩溃重启了，消息系统（比如Redis的Pub/Sub本身不持久化，但一些基于Redis的消息队列模式可能会有）一看，“诶，这条消息没收到确认啊”，等消费者服务重新连上来，可能就又把那条老消息推给它了。
网络抽风了：网络问题可能导致消息确认丢失，从而引发重发。

重复消息有时候是避免不了的,我们得在消费者这一端想想办法，给它加上一个“防护网”。

Redis怎么帮我们给消息去重呢？

核心思想就一句话：让每条消息都有一个唯一的“身份证”（ID），消费者在处理消息前，先拿这个“身份证”去Redis里查查户口，看是不是已经处理过了，如果查过了，就直接忽略这条消息；如果没查过，就处理它，并且在处理成功后，把这个“身份证”在Redis里标记为“已处理”。

下面说说几种常见的具体做法：

用SETNX命令和过期时间

这是比较经典和简单的一种方法,SETNX是“SET if Not eXists”的缩写，意思是只有当这个键不存在的时候，才设置它。

消息的唯一ID：我们需要给每条消息一个唯一的标识，可以用“订单ID”加上一个操作类型（order:123:created"），或者用一个全局唯一的消息ID（比如UUID）。
操作步骤：
1. 消费者收到一条消息,先提取出它的唯一ID。
2. 执行一个Redis命令：SETNX key value，这里的key就是消息的唯一ID，value可以随便设个值，比如1或者"processed"，这不重要，重要的是这个key是否存在。
3. 如果SETNX返回1,说明这个key之前不存在，我们成功设置了它，意味着这条消息是第一次见到，那么消费者就可以放心地去处理业务逻辑了。
4. 处理完业务逻辑后,最好给这个key设置一个过期时间（TTL），比如5分钟或者半小时，这是因为，如果消息量很大，这些已处理的ID会一直留在Redis里，占用内存，设置过期时间能让它们自动清理，命令是 EXPIRE key seconds，也可以直接用 SET key value EX seconds NX 一条命令同时完成设置和过期时间。
5. 如果SETNX返回0,说明这个key已经存在了，意味着这条消息之前已经被处理过（或者正在被处理），那么消费者就直接跳过，不去做任何业务操作。

这种方法简单有效,非常适合消息量不是特别巨大、去重精度要求高的场景。

用布隆过滤器（Bloom Filter）

如果消息量非常大,比如每秒几万甚至几十万条，用SETNX给每条消息都存一个key，即使有过期时间，对内存的消耗也可能是个问题，这时候可以考虑布隆过滤器。

布隆过滤器是啥：你可以把它理解成一个很节省空间的“集合”，它用多个哈希函数把元素映射到一个很长的二进制向量（bit数组）上，它能告诉你“某个元素肯定不在集合里”或者“可能在集合里”。
为啥省空间：因为它不存储元素本身，只存储几个比特位的标记。
怎么用来去重：
1. 消费者收到消息,提取唯一ID。
2. 用布隆过滤器检查这个ID：BF.EXISTS key item（需要Redis 4.0以上版本支持布隆过滤器模块）。
3. 如果返回0,说明这个ID肯定没出现过，是新车，那就处理消息，并且用 BF.ADD key item 把这个ID加到过滤器里。
4. 如果返回1,说明这个ID可能出现过，这时候，为了保险起见，我们通常就把它当作重复消息处理，直接跳过，是的，布隆过滤器有极小的误判率（把新车误认为是旧车），但它绝不会把旧车漏掉（不会把重复消息当成新的），对于消息去重这个场景，我们宁可错杀一千（偶尔浪费一条新消息），不可放过一个（坚决不处理重复消息），所以这个特性是合适的。

布隆过滤器适合海量数据去重,且可以接受极小概率误判的场景。

用Sorted Set（有序集合）

这个方法稍微绕一点,但也能实现，特别适合需要按时间顺序去重，或者想顺便清理很老的去重记录的场合。

思路：把消息的唯一ID作为Sorted Set的member，把当前的时间戳（比如毫秒时间戳）作为它的score。
操作步骤：
1. 收到消息,提取ID和当前时间戳current_time。
2. 先使用 ZADD key NX score member 命令尝试添加，NX选项表示只有当这个member不存在时才添加，如果添加成功，说明是新车，就去处理业务。
3. 可以定期（比如每次处理消息后，或者用一个定时任务）用 ZREMRANGEBYSCORE key -inf (current_time - interval) 来删除interval时间（比如1小时）之前的所有去重记录，防止集合无限膨胀。