解剖屎山，寻觅黄金之第二弹 天天通讯

大【dà】家好，我3y啊。由于去重逻辑重【chóng】构了几次，好多股东直【zhí】呼看不懂，于是【shì】我今天再安排一【yī】波对代码的【de】解析【xī】吧。austin支持两种去重的【de】类【lèi】型：N分【fèn】钟相同内容达【dá】到【dào】N次【cì】去重和一天内N次【cì】相同渠道频次去重。

在最开始，我的第一版实现是这样的：

(资料图片仅供参考)

publicvoidduplication(TaskInfotaskInfo){//配置示【shì】例【lì】:{"contentDeduplication":{"num":1,"time":300},"frequencyDeduplication":{"num":5}}JSONObjectproperty=JSON.parseObject(config.getProperty(DEDUPLICATION_RULE_KEY,AustinConstant.APOLLO_DEFAULT_VALUE_JSON_OBJECT));JSONObjectcontentDeduplication=property.getJSONObject(CONTENT_DEDUPLICATION);JSONObjectfrequencyDeduplication=property.getJSONObject(FREQUENCY_DEDUPLICATION);//文案【àn】去重【chóng】DeduplicationParamcontentParams=DeduplicationParam.builder().deduplicationTime(contentDeduplication.getLong(TIME)).countNum(contentDeduplication.getInteger(NUM)).taskInfo(taskInfo).anchorState(AnchorState.CONTENT_DEDUPLICATION).build();contentDeduplicationService.deduplication(contentParams);//运营总规【guī】则去重(一【yī】天内用户收到最多同一个渠【qú】道的消息次数)Longseconds=(DateUtil.endOfDay(newDate()).getTime()-DateUtil.current())/1000;DeduplicationParambusinessParams=DeduplicationParam.builder().deduplicationTime(seconds).countNum(frequencyDeduplication.getInteger(NUM)).taskInfo(taskInfo).anchorState(AnchorState.RULE_DEDUPLICATION).build();frequencyDeduplicationService.deduplication(businessParams);}

那时候很简单，基【jī】本主体逻辑【jí】都【dōu】写在【zài】这个【gè】入口上了，应【yīng】该【gāi】都能看得懂。后来，群里滴滴哥表【biǎo】示这种【zhǒng】代码不行，不能一眼看【kàn】出来它干了什么。于是怒提了一【yī】波【bō】pull request重构了一版，入【rù】口是【shì】这样的：

publicvoidduplication(TaskInfotaskInfo){//配置【zhì】样例【lì】：{"contentDeduplication":{"num":1,"time":300},"frequencyDeduplication":{"num":5}}Stringdeduplication=config.getProperty(DeduplicationConstants.DEDUPLICATION_RULE_KEY,AustinConstant.APOLLO_DEFAULT_VALUE_JSON_OBJECT);//去重DEDUPLICATION_LIST.forEach(key->{DeduplicationParamdeduplicationParam=builderFactory.select(key).build(deduplication,key);if(deduplicationParam!=null){deduplicationParam.setTaskInfo(taskInfo);DeduplicationServicededuplicationService=findService(key+SERVICE);deduplicationService.deduplication(deduplicationParam);}});}

我猜想他的思路就是把构建去重参数和选择具体【tǐ】的去【qù】重服务【wù】给封装【zhuāng】起来了，在最外层的【de】代码看起来【lái】就很简洁了。后来又【yòu】跟他【tā】聊了【le】下，他的设计思路是这样的：考【kǎo】虑到【dào】以后会有其他规则的去重就把去重逻【luó】辑【jí】单独封【fēng】装起【qǐ】来了，之后用策略模【mó】版【bǎn】的设计模【mó】式【shì】进【jìn】行【háng】了【le】重构，重构后的代【dài】码 模版不【bú】变，支持【chí】各种不同策略的去重，扩展性更高更强更简【jiǎn】洁

确实牛逼。

我基于上面的思路微改了下入口，代码最终演变成这样：

publicvoidduplication(TaskInfotaskInfo){//配【pèi】置样【yàng】例：{"deduplication_10":{"num":1,"time":300},"deduplication_20":{"num":5}}StringdeduplicationConfig=config.getProperty(DEDUPLICATION_RULE_KEY,CommonConstant.EMPTY_JSON_OBJECT);//去重ListdeduplicationList=DeduplicationType.getDeduplicationList();for(IntegerdeduplicationType:deduplicationList){DeduplicationParamdeduplicationParam=deduplicationHolder.selectBuilder(deduplicationType).build(deduplicationConfig,taskInfo);if(Objects.nonNull(deduplicationParam)){deduplicationHolder.selectService(deduplicationType).deduplication(deduplicationParam);}}}

到这【zhè】，应该【gāi】大多数人【rén】还能跟上吧？在讲【jiǎng】具体【tǐ】的代码之前，我们先来简单看【kàn】看去重功能的代码【mǎ】结构（这会【huì】对【duì】后面看代码有帮助）

去重的逻辑可【kě】以【yǐ】统一【yī】抽象为：在X时间段内达到了Y阈值，还【hái】记得我曾经说过：「去重」的本质：「业务Key」+「存储」。那么去重实现【xiàn】的步【bù】骤【zhòu】可以【yǐ】简单分为【wéi】（我【wǒ】这边存【cún】储就用【yòng】的Redis）：

通【tōng】过Key从Redis获【huò】取记录判断【duàn】该Key在Redis的记录【lù】是否符合条件【jiàn】符合条件的【de】则去【qù】重，不符合【hé】条件的则重新塞进Redis更新记录

为了【le】方便调整【zhěng】去重的参数【shù】，我把X时【shí】间段和Y阈值都放【fàng】到了配置里{"deduplication_10":{"num":1,"time":300},"deduplication_20":{"num":5}}。目前有两种去【qù】重的具体实现：

1、5分钟内相同用户如果收到相同的内容，则应该被过滤掉

2、一天内【nèi】相同的用【yòng】户【hù】如果已【yǐ】经收到某渠道内容【róng】5次，则应该被过滤【lǜ】掉

从配置中心拿到配置【zhì】信息了以后，Builder就是根据这两种【zhǒng】类【lèi】型【xíng】去构建出DeduplicationParam，就是以下【xià】代码：

DeduplicationParamdeduplicationParam=deduplicationHolder.selectBuilder(deduplicationType).build(deduplicationConfig,taskInfo);

Builder和DeduplicationService都用了类【lèi】似的写法（在子类初始化【huà】的时候指定类型，在【zài】父类【lèi】统一接收，放到Map里管理）

而统【tǒng】一管理着这【zhè】些服务有个中心的【de】地方，我把【bǎ】这取名【míng】为DeduplicationHolder

/***@authorhuskey*@date2022/1/18*/@ServicepublicclassDeduplicationHolder{privatefinalMapbuilderHolder=newHashMap>(4);privatefinalMapserviceHolder=newHashMap>(4);publicBuilderselectBuilder(Integerkey){returnbuilderHolder.get(key);}publicDeduplicationServiceselectService(Integerkey){returnserviceHolder.get(key);}publicvoidputBuilder(Integerkey,Builderbuilder){builderHolder.put(key,builder);}publicvoidputService(Integerkey,DeduplicationServiceservice){serviceHolder.put(key,service);}}

前【qián】面提到的业【yè】务Key，是在AbstractDeduplicationService的【de】子类下构【gòu】建的：

而【ér】具体的去重【chóng】逻辑实现【xiàn】则都在LimitService下，{一天【tiān】内相同的用户【hù】如【rú】果已经收到某渠道内容5次}是在【zài】SimpleLimitService中处理使用mget和pipelineSetEX就完成了实现【xiàn】。而【ér】{5分钟内相同用户【hù】如果收到相【xiàng】同的【de】内【nèi】容}是【shì】在SlideWindowLimitService中处理，使用了lua脚本完成了【le】实【shí】现。

LimitService的代码都【dōu】来源于@caolongxiu的【de】pull request，建议大家【jiā】可以对比commit再学习一番【fān】：https://gitee.com/zhongfucheng/austin/pulls/19

1、频次去重采用普通的计数去重方法，限制的是每天发送的条数。

2、内容去【qù】重采用的是新开发的基【jī】于redis中zset的滑【huá】动【dòng】窗口去【qù】重，可以做【zuò】到严格控制单【dān】位时间内的频次。

3、redis使用lua脚本来保证原子性和减少网络io的损耗

4、redis的key增加前缀做到【dào】数据隔【gé】离【lí】（后期可能有动态更换去重方【fāng】法的需求【qiú】）

5、把具体【tǐ】限【xiàn】流【liú】去重方【fāng】法从DeduplicationService抽取出来，DeduplicationService只需设置构造器注入时注入的AbstractLimitService（具【jù】体【tǐ】限流【liú】去重【chóng】服【fú】务）类型【xíng】即可动态更【gèng】换去重的方法 6、使用雪花算法生成zset的【de】唯一【yī】value,score使用的是当前的时间戳

针对【duì】滑动窗口去重，有会【huì】引申出新的【de】问题【tí】：limit.lua的逻辑？为什【shí】么要【yào】移除时间窗口的之前的数据？为什【shí】么ARGV[4]参数要唯一？为什【shí】么要expire？

A: 使用【yòng】滑动窗口可【kě】以保证N分钟达【dá】到N次进行去【qù】重【chóng】。滑动窗口可以【yǐ】回顾下TCP的，也可以【yǐ】回顾下刷LeetCode时的一【yī】些题【tí】，那这为什【shí】么要【yào】移除，就不陌生了。

为什【shí】么ARGV[4]要唯一【yī】，具体可以看看zadd这【zhè】条【tiáo】命令，我们【men】只需【xū】要保证每次【cì】add进窗口内的成员是唯【wéi】一的，那么【me】就不会触发有更新【xīn】的操作（我认【rèn】为这样【yàng】设计【jì】会更加简单些），而【ér】唯一Key用雪【xuě】花算法比较方便。

为什么expire？，如果这个key只【zhī】被调用【yòng】一次【cì】。那就【jiù】很有【yǒu】可能在redis内存常驻了【le】，expire能避免这种情况。

推荐项目

最后再叨叨吧，很多人可能会【huì】发一段截图【tú】，跑【pǎo】来问【wèn】我为什么要【yào】这样【yàng】写，为什么要以【yǐ】这【zhè】种方式实现，能不能以【yǐ】这【zhè】种方式实现。这时候【hòu】，我更想看【kàn】到的是【shì】：你【nǐ】已经实【shí】现了第【dì】二种【zhǒng】方式了，然后探讨你写的这种方【fāng】案好不好，现有的代码差在哪里。

毕竟问问题很【hěn】简【jiǎn】单【dān】，我又不是客【kè】服，总不能没诚意的问【wèn】题我都得【dé】一一回答吧。

如果【guǒ】想学Java项目的，我还是强烈推荐我的开源项目消【xiāo】息【xī】推送平台Austin，可以用作【zuò】毕【bì】业设【shè】计，可【kě】以用作【zuò】校招【zhāo】，可以看【kàn】看生产环境是怎么【me】推送消息的。

仓【cāng】库地址【zhǐ】（可【kě】点击阅读原文跳转【zhuǎn】）：https://gitee.com/zhongfucheng/austin

我开通了股东【dōng】服务【wù】内容，感兴趣可以点击下【xià】方看看，主要【yào】针【zhēn】对的【de】是项目哟

VIP服务