一個簡單的時間窗口設計與實現

如何設計一個計數的時間窗口

時間窗口，一般對於一些實時信息展現中用得比較多，好比維持一個五分鐘的交易明細時間窗口，就須要記錄當前時間，到五分鐘以前的全部交易明細，而五分鐘以前的數據，則丟掉

一個簡單的實現就是用一個隊列來作，新的數據在對頭添加；同時起一個線程，不斷的詢問隊尾的數據是否過時，若是過時則丟掉

另一中場景須要利用到這個時間窗口內的數據進行計算，如計算着五分鐘交易中資金的流入流出總和，若是依然用上面的這種方式，會有什麼問題？

• 若是時間窗口大，須要存儲大量的明細數據
• 咱們主要關心的只有資金流入流出；存這些明細數據得不償失
• 每次新增or刪除過時數據時，實時計算流入流出消耗性能

針對這種特殊的場景，是否有什麼取巧的實現方式呢？

I. 方案設計

1. 基於隊列的輪詢刪除方式

將時間窗口分割成一個一個的時間片，每一個時間片中記錄資金的流入流出總數，而後總的流入流出就是全部時間片的流入流出的和

新增數據：

• 若未跨時間片，則更新隊頭的值
• 若跨時間片，新增一個隊列頭

刪除數據：

• 輪詢任務，判斷隊列尾是否過時
• 隊尾過時，則刪除隊尾，此時若隊頭數據未加入計算，也須要加入計算

2. 基於隊列的新增時刪除方式

相比較前面的輪詢方式，這個的應用場景爲另一種，只有在新增數據時，確保數據的準確性便可，不須要輪詢的任務去刪除過時的數據

簡單來講，某些場景下（好比能確保數據不會斷續的進來，即每一個時間片都至少有一個數據過來），此時但願個人時間窗口數據是由新增的數據來驅動並更新

新增數據：

• 未跨時間片，則更新隊頭值
• 跨時間片，新塞入一個，並刪除舊的數據

II. 基於數組的時間窗口實現

針對上面第二種，基於數組給出一個簡單的實現，本篇主要是給出一個基礎的時間窗口的設計與實現方式，固然也須要有進階的case，好比上面的資金流入流出中，我須要分別計算5min,10min,30min,1h,3h,6h,12h,24h的時間窗口，該怎麼來實現呢？可否用一個隊列就知足全部的時間窗口的計算呢？關於這些留待下一篇給出

1. 時間輪計算器

前面用隊列的方式比較好理解，這裏爲何用數組方式來實現？

• 固定長度，避免頻繁的新增和刪除對象
• 定位和更新數據方便

首先是須要實現一個時間輪計算器，根據傳入的時間，獲取須要刪除的過時數據

@Data
public class TimeWheelCalculate {
    private static final long START = 0;

    private int period;
    private int length;

    /** * 劃分的時間片個數 */
    private int cellNum;

    private void check() {
        if (length % period != 0) {
            throw new IllegalArgumentException(
                    "length % period should be zero but not! now length: " + length + " period: " + period);
        }
    }

    public TimeWheelCalculate(int period, int length) {
        this.period = period;
        this.length = length;

        check();

        this.cellNum = length / period;
    }

    public int calculateIndex(long time) {
        return (int) ((time - START) % length / period);
    }

    /** * 獲取全部過時的時間片索引 * * @param lastInsertTime 上次更新時間輪的時間戳 * @param nowInsertTime 本次更新時間輪的時間戳 * @return */
    public List<Integer> getExpireIndexes(long lastInsertTime, long nowInsertTime) {
        if (nowInsertTime - lastInsertTime >= length) {
            // 已通過了一輪，過去的數據所有丟掉
            return null;
        }

        List<Integer> removeIndexList = new ArrayList<>();
        int lastIndex = calculateIndex(lastInsertTime);
        int nowIndex = calculateIndex(nowInsertTime);
        if (lastIndex == nowIndex) {
            // 尚未跨過這個時間片，則不須要刪除過時數據
            return Collections.emptyList();
        } else if (lastIndex < nowIndex) {
            for (int tmp = lastIndex; tmp < nowIndex; tmp++) {
                removeIndexList.add(tmp);
            }
        } else {
            for (int tmp = lastIndex; tmp < cellNum; tmp++) {
                removeIndexList.add(tmp);
            }

            for (int tmp = 0; tmp < nowIndex; tmp++) {
                removeIndexList.add(tmp);
            }
        }
        return removeIndexList;
    }
}
複製代碼

這個計算器的實現比較簡單，首先是指定時間窗口的長度(length),時間片(period)，其主要提供兩個方法

• calculateIndex 根據當前時間，肯定過時的數據在數組的索引
• getExpireIndexes 根據上次插入的時間，和當前插入的時間，計算兩次插入時間之間，全部的過時數據索引

2. 時間輪容器

容器內保存的時間窗口下的數據，包括實時數據，和過去n個時間片的數組，其主要的核心就是在新增數據時，須要判斷

• 若跨時間片，則刪除過時數據，更新實時數據，更新總數
• 若未跨時間片，則直接更新實時數據便可

@Data
public class TimeWheelContainer {
    private TimeWheelCalculate calculate;

    /** * 歷史時間片計數，每一個時間片對應其中的一個元素 */
    private int[] counts;

    /** * 實時的時間片計數 */
    private int realTimeCount;

    /** * 整個時間輪計數 */
    private int timeWheelCount;

    private Long lastInsertTime;


    public TimeWheelContainer(TimeWheelCalculate calculate) {
        this.counts = new int[calculate.getCellNum()];
        this.calculate = calculate;
        this.realTimeCount = 0;
        this.timeWheelCount = 0;
        this.lastInsertTime = null;
    }

    public void add(long now, int amount) {
        if (lastInsertTime == null) {
            realTimeCount = amount;
            lastInsertTime = now;
            return;
        }


        List<Integer> removeIndex = calculate.getExpireIndexes(lastInsertTime, now);
        if (removeIndex == null) {
            // 二者時間間隔超過一輪，則清空計數
            realTimeCount = amount;
            lastInsertTime = now;
            timeWheelCount = 0;
            clear();
            return;
        }

        if (removeIndex.isEmpty()) {
            // 沒有跨過期間片，則只更新實時計數
            realTimeCount += amount;
            lastInsertTime = now;
            return;
        }

        // 跨過了時間片，則須要在總數中刪除過時的數據，並追加新的數據
        for (int index : removeIndex) {
            timeWheelCount -= counts[index];
            counts[index] = 0;
        }
        timeWheelCount += realTimeCount;
        counts[calculate.calculateIndex(lastInsertTime)] = realTimeCount;
        lastInsertTime = now;
        realTimeCount = amount;
    }

    private void clear() {
        for (int i = 0; i < counts.length; i++) {
            counts[i] = 0;
        }
    }
}
複製代碼

3. 測試

主要就是驗證上面的實現有沒有明顯的問題，爲何是明顯的問題？

• 深層次的bug在實際的使用中，更容易暴露

public class CountTimeWindow {

    public static void main(String[] args) {
        TimeWheelContainer timeWheelContainer = new TimeWheelContainer(new TimeWheelCalculate(2, 20));

        timeWheelContainer.add(0, 1);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 0, "first");

        timeWheelContainer.add(1, 1);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 0, "first");

        timeWheelContainer.add(2, 1);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 2, "second");
        Assert.isTrue(timeWheelContainer.getCounts()[0] == 2, "second");

        for (int i = 3; i < 20; i++) {
            timeWheelContainer.add(i, 1);
            System.out.println("add index: " + i + " count: " + timeWheelContainer.getTimeWheelCount());
        }

        // 恰好一輪

        timeWheelContainer.add(20, 3);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 20, "third");
        timeWheelContainer.add(21, 3);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 20, "third");


        // 減去過時的那個數據
        timeWheelContainer.add(22, 3);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 26 - 2, "fourth");
        Assert.isTrue(timeWheelContainer.getCounts()[0] == 6, "fourth");


        timeWheelContainer.add(26, 3);
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 24 - 2 - 2 + 3, "fifth");
        System.out.println(Arrays.toString(timeWheelContainer.getCounts()));


        timeWheelContainer.add(43, 3);
        System.out.println(Arrays.toString(timeWheelContainer.getCounts()));
        Assert.isTrue(timeWheelContainer.getTimeWheelCount() == 6, "six");
    }
}
複製代碼

III. 其餘

1. 一灰灰Blog： https://liuyueyi.github.io/hexblog

一灰灰的我的博客，記錄全部學習和工做中的博文，歡迎你們前去逛逛

2. 聲明

盡信書則不如，已上內容，純屬一家之言，因我的能力有限，不免有疏漏和錯誤之處，如發現bug或者有更好的建議，歡迎批評指正，不吝感激

• 微博地址: 小灰灰Blog
• QQ： 一灰灰/3302797840

3. 掃描關注