Java如何實作去重？這是在炫技嗎？

大家好，我3y啊，由于去重邏輯重構了幾次，好多股東直呼看不懂，于是我今天再安排一波對代碼的決議吧，austin支持兩種去重的型別：N分鐘相同內容達到N次去重和一天內N次相同渠道頻次去重，

Java開源專案訊息推送平臺??推送下發【郵件】【短信】【微信服務號】【微信小程式】【企業微信】【釘釘】等訊息型別，

• https://gitee.com/zhongfucheng/austin/
• https://github.com/ZhongFuCheng3y/austin

在最開始，我的第一版實作是這樣的：

public void duplication(TaskInfo taskInfo) {
    // 配置示例:{"contentDeduplication":{"num":1,"time":300},"frequencyDeduplication":{"num":5}}
    JSONObject property = JSON.parseObject(config.getProperty(DEDUPLICATION_RULE_KEY, AustinConstant.APOLLO_DEFAULT_VALUE_JSON_OBJECT));
    JSONObject contentDeduplication = property.getJSONObject(CONTENT_DEDUPLICATION);
    JSONObject frequencyDeduplication = property.getJSONObject(FREQUENCY_DEDUPLICATION);
?
    // 文案去重
    DeduplicationParam contentParams = DeduplicationParam.builder()
        .deduplicationTime(contentDeduplication.getLong(TIME))
        .countNum(contentDeduplication.getInteger(NUM)).taskInfo(taskInfo)
        .anchorState(AnchorState.CONTENT_DEDUPLICATION)
        .build();
    contentDeduplicationService.deduplication(contentParams);
?
?
    // 運營總規則去重(一天內用戶收到最多同一個渠道的訊息次數)
    Long seconds = (DateUtil.endOfDay(new Date()).getTime() - DateUtil.current()) / 1000;
    DeduplicationParam businessParams = DeduplicationParam.builder()
        .deduplicationTime(seconds)
        .countNum(frequencyDeduplication.getInteger(NUM)).taskInfo(taskInfo)
        .anchorState(AnchorState.RULE_DEDUPLICATION)
        .build();
    frequencyDeduplicationService.deduplication(businessParams);
}

那時候很簡單，基本主體邏輯都寫在這個入口上了，應該都能看得懂，后來，群里滴滴哥表示這種代碼不行，不能一眼看出來它干了什么，于是怒提了一波pull request重構了一版，入口是這樣的：

public void duplication(TaskInfo taskInfo) {
    
    // 配置樣例：{"contentDeduplication":{"num":1,"time":300},"frequencyDeduplication":{"num":5}}
    String deduplication = config.getProperty(DeduplicationConstants.DEDUPLICATION_RULE_KEY, AustinConstant.APOLLO_DEFAULT_VALUE_JSON_OBJECT);
    
    //去重
    DEDUPLICATION_LIST.forEach(
        key -> {
            DeduplicationParam deduplicationParam = builderFactory.select(key).build(deduplication, key);
            if (deduplicationParam != null) {
                deduplicationParam.setTaskInfo(taskInfo);
                DeduplicationService deduplicationService = findService(key + SERVICE);
                deduplicationService.deduplication(deduplicationParam);
            }
        }
    );
}

我猜想他的思路就是把構建去重引數和選擇具體的去重服務給封裝起來了，在最外層的代碼看起來就很簡潔了，后來又跟他聊了下，他的設計思路是這樣的：考慮到以后會有其他規則的去重就把去重邏輯單獨封裝起來了，之后用策略模版的設計模式進行了重構，重構后的代碼 模版不變，支持各種不同策略的去重，擴展性更高更強更簡潔

確實牛逼，

我基于上面的思路微改了下入口，代碼最終演變成這樣：

public void duplication(TaskInfo taskInfo) {
    // 配置樣例：{"deduplication_10":{"num":1,"time":300},"deduplication_20":{"num":5}}
    String deduplicationConfig = config.getProperty(DEDUPLICATION_RULE_KEY, CommonConstant.EMPTY_JSON_OBJECT);
?
    // 去重
    List<Integer> deduplicationList = DeduplicationType.getDeduplicationList();
    for (Integer deduplicationType : deduplicationList) {
        DeduplicationParam deduplicationParam = deduplicationHolder.selectBuilder(deduplicationType).build(deduplicationConfig, taskInfo);
        if (Objects.nonNull(deduplicationParam)) {
            deduplicationHolder.selectService(deduplicationType).deduplication(deduplicationParam);
        }
    }
}

到這，應該大多數人還能跟上吧？在講具體的代碼之前，我們先來簡單看看去重功能的代碼結構（這會對后面看代碼有幫助）

去重的邏輯可以統一抽象為：在X時間段內達到了Y閾值，還記得我曾經說過：「去重」的本質：「業務Key」+「存盤」，那么去重實作的步驟可以簡單分為（我這邊存盤就用的Redis）：

• 通過Key從Redis獲取記錄
• 判斷該Key在Redis的記錄是否符合條件
• 符合條件的則去重，不符合條件的則重新塞進Redis更新記錄

為了方便調整去重的引數，我把X時間段和Y閾值都放到了配置里{"deduplication_10":{"num":1,"time":300},"deduplication_20":{"num":5}}，目前有兩種去重的具體實作：

1、5分鐘內相同用戶如果收到相同的內容，則應該被過濾掉

2、一天內相同的用戶如果已經收到某渠道內容5次，則應該被過濾掉

從配置中心拿到配置資訊了以后，Builder就是根據這兩種型別去構建出DeduplicationParam，就是以下代碼：

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

Builder和DeduplicationService都用了類似的寫法（在子類初始化的時候指定型別，在父類統一接收，放到Map里管理）

而統一管理著這些服務有個中心的地方，我把這取名為DeduplicationHolder

/**
 * @author huskey
 * @date 2022/1/18
 */
@Service
public class DeduplicationHolder {
?
    private final Map<Integer, Builder> builderHolder = new HashMap<>(4);
    private final Map<Integer, DeduplicationService> serviceHolder = new HashMap<>(4);
?
    public Builder selectBuilder(Integer key) {
        return builderHolder.get(key);
    }
?
    public DeduplicationService selectService(Integer key) {
        return serviceHolder.get(key);
    }
?
    public void putBuilder(Integer key, Builder builder) {
        builderHolder.put(key, builder);
    }
?
    public void putService(Integer key, DeduplicationService service) {
        serviceHolder.put(key, service);
    }
}

前面提到的業務Key，是在AbstractDeduplicationService的子類下構建的：

而具體的去重邏輯實作則都在LimitService下，{一天內相同的用戶如果已經收到某渠道內容5次}是在SimpleLimitService中處理使用mget和pipelineSetEX就完成了實作，而{5分鐘內相同用戶如果收到相同的內容}是在SlideWindowLimitService中處理，使用了lua腳本完成了實作，

LimitService的代碼都來源于@caolongxiu的pull request，建議大家可以對比commit再學習一番：https://gitee.com/zhongfucheng/austin/pulls/19

1、頻次去重采用普通的計數去重方法，限制的是每天發送的條數， 2、內容去重采用的是新開發的基于redis中zset的滑動視窗去重，可以做到嚴格控制單位時間內的頻次， 3、redis使用lua腳本來保證原子性和減少網路io的損耗 4、redis的key增加前綴做到資料隔離（后期可能有動態更換去重方法的需求） 5、把具體限流去重方法從DeduplicationService抽取出來，DeduplicationService只需設定構造器注入時注入的AbstractLimitService（具體限流去重服務）型別即可動態更換去重的方法 6、使用雪花演算法生成zset的唯一value,score使用的是當前的時間戳

針對滑動視窗去重，有會引申出新的問題：limit.lua的邏輯？為什么要移除時間視窗的之前的資料？為什么ARGV[4]引數要唯一？為什么要expire？

A: 使用滑動視窗可以保證N分鐘達到N次進行去重，滑動視窗可以回顧下TCP的，也可以回顧下刷LeetCode時的一些題，那這為什么要移除，就不陌生了，

為什么ARGV[4]要唯一，具體可以看看zadd這條命令，我們只需要保證每次add進視窗內的成員是唯一的，那么就不會觸發有更新的操作（我認為這樣設計會更加簡單些），而唯一Key用雪花演算法比較方便，

為什么expire？，如果這個key只被呼叫一次，那就很有可能在redis記憶體常駐了，expire能避免這種情況，

如果想學Java專案的，強烈推薦我的專案訊息推送平臺Austin（8K stars），可以用作畢業設計**，可以用作校招，可以看看生產環境是怎么推送訊息的，訊息推送平臺??推送下發【郵件】【短信】【微信服務號】【微信小程式】【企業微信】【釘釘】等訊息型別**，

• https://gitee.com/zhongfucheng/austin/
• https://github.com/ZhongFuCheng3y/austin

標籤：其他

上一篇：Python生成器深度決議：構建強大的資料處理管道

下一篇：返回列表