012-redis應用-05-限流【簡單限流、漏斗限流】

一、概述

　　限流主要目的控制流量、用於控制用戶行為，避免垃圾請求

1.1、簡單限流

　　限流需求中存在一個滑動時間窗口，適用 zset 數據結構的 score 值，可以通過 score 來圈出這個時間窗口。而且我們只需要保留這個時間窗口，窗口之外的數據都 可以刪除。

　　 zset 的 value 只需要保證唯一性即可，用 uuid 會比較浪費空間，可以使用毫秒時間戳。

　　　　

　　如圖所示，用一個 zset 結構記錄用戶的行為歷史，每一個行為都會作為 zset 中的一個 key 保存下來。同一個用戶同一種行為用一個 zset 記錄。

　　為節省內存，我們只需要保留時間窗口內的行為記錄，同時如果用戶是冷用戶，滑動時 間窗口內的行為是空記錄，那么這個 zset 就可以從內存中移除，不再占用空間。

　　通過統計滑動窗口內的行為數量與閾值 max_count 進行比較就可以得出當前的行為是否 允許。用代碼表示如下: 

public class SimpleRateLimiter {
    private Jedis jedis;

    public SimpleRateLimiter(Jedis jedis) {
        this.jedis = jedis;
    }

    public boolean isActionAllowed(String userId, String actionKey, int period, int maxCount) {
        String key = String.format("hist:%s:%s", userId, actionKey);
        long nowTs = System.currentTimeMillis();
        Pipeline pipe = jedis.pipelined();
        pipe.multi();
        pipe.zadd(key, nowTs, "" + nowTs);
        pipe.zremrangeByScore(key, 0, nowTs - period * 1000);
        Response<Long> count = pipe.zcard(key);
        pipe.expire(key, period + 1);
        pipe.exec();
        pipe.close();
        return count.get() <= maxCount;
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        SimpleRateLimiter limiter = new SimpleRateLimiter(jedis);
        for (int i = 0; i < 20; i++) {
            System.out.println(limiter.isActionAllowed("laoqian", "reply", 60, 5));
        }
    }
}

 

　　它的整體思路就是:每一個行為到來時，都維護一次時間窗口。將時間窗口外的記錄全部清理掉，只保留窗口內的記錄。zset 集合中只有 score 值非常重要，value 值沒有特別的意義，只需要保證它是唯一的就可 以了。

　　因為這幾個連續的 Redis 操作都是針對同一個 key 的，使用 pipeline 可以顯著提升 Redis 存取效率。但這種方案也有缺點，因為它要記錄時間窗口內所有的行為記錄，如果這 個量很大，比如限定 60s 內操作不得超過 100w 次這樣的參數，它是不適合做這樣的限流 的，因為會消耗大量的存儲空間。 

1.2、漏斗限流

　　漏斗（funnel ）限流是最常用的限流方法之一 

　　漏洞的容量是有限的，如果將漏嘴堵住，然后一直往里面灌水，它就會變滿，直至再也裝不進去。如果將漏嘴放開，水就會往下流，流走一部分之后，就又可以繼續往里面灌水。如果漏嘴流水的速率大於灌水的速率，那么漏斗永遠都裝不滿。如果漏嘴流水速率小於灌水的速率，那么一旦漏斗滿了，灌水就需要暫停並等待漏斗騰空。

　　所以，漏斗的剩余空間就代表着當前行為可以持續進行的數量，漏嘴的流水速率代表着系統允許該行為的最大頻率。

1.2.1、java版本【自行實現】

public class FunnelRateLimiter {
    static class Funnel {
        int capacity;
        float leakingRate;
        int leftQuota;
        long leakingTs;

        public Funnel(int capacity, float leakingRate) {
            this.capacity = capacity;
            this.leakingRate = leakingRate;
            this.leftQuota = capacity;
            this.leakingTs = System.currentTimeMillis();
        }

        void makeSpace() {
            long nowTs = System.currentTimeMillis();
            long deltaTs = nowTs - leakingTs;
            int deltaQuota = (int) (deltaTs * leakingRate);
            if (deltaQuota < 0) { // 間隔時間太長，整數數字過大溢出
                this.leftQuota = capacity;
                this.leakingTs = nowTs;
                return;
            }
            if (deltaQuota < 1) { // 騰出空間太小，最小單位是 1
                return;
            }
            this.leftQuota += deltaQuota;
            this.leakingTs = nowTs;
            if (this.leftQuota > this.capacity) {
                this.leftQuota = this.capacity;
            }
        }

        boolean watering(int quota) {
            makeSpace();
            if (this.leftQuota >= quota) {
                this.leftQuota -= quota;
                return true;
            }
            return false;
        }
    }

    private Map<String, Funnel> funnels = new HashMap<>();

    public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
        String key = String.format("%s:%s", userId, actionKey);
        Funnel funnel = funnels.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnels.put(key, funnel);
        }
        return funnel.watering(1); // 需要 1 個 quota
    }
}

 

　　Funnel 對象的 make_space 方法是漏斗算法的核心，其在每次灌水前都會被調用以觸發 漏水，給漏斗騰出空間來。能騰出多少空間取決於過去了多久以及流水的速率。Funnel 對象 占據的空間大小不再和行為的頻率成正比，它的空間占用是一個常量。 

1.2.2、Redis-Cell實現

　　Redis 4.0 提供了一個限流 Redis 模塊，它叫 redis-cell。該模塊也使用了漏斗算法，並 提供了原子的限流指令。 該模塊只有 1 條指令 cl.throttle 

1.2.2.1、安裝方式

官方提供了安裝包和源碼編譯兩種方式，源碼編譯要安裝rust環境，比較復雜，這里介紹安裝包方式安裝：

1. 根據操作系統下載安裝包;
2. 將文件解壓到redis能訪問到的路徑下；
3. 進入 redis-cli，執行命令臨時啟動： module load /path/to/libredis_cell.so;永久：redis-server --loadmodule /aaa/bbb/libredis_cell.so

1.2.2.2、參數說明

cl.throttle test 100 400 60 3

test： redis key

100： 官方叫max_burst，其值為令牌桶的容量 - 1， 首次執行時令牌桶會默認填滿，漏斗容量

400： 與下一個參數一起，表示在指定時間窗口內允許訪問的次數

60： 指定的時間窗口，單位：秒

3： 表示本次要申請的令牌數，不寫則默認為 1

以上命令表示從一個初始值為100的令牌桶中取3個令牌，該令牌桶的速率限制為400次/60秒。

127.0.0.1:6379> cl.throttle test 100 400 60 3
1) (integer) 0
2) (integer) 101
3) (integer) 98
4) (integer) -1
5) (integer) 0

1： 是否成功，0：成功，1：拒絕

2： 令牌桶的容量，大小為初始值+1

3： 當前令牌桶中可用的令牌

4： 若請求被拒絕，這個值表示多久后才令牌桶中會重新添加令牌，單位：秒，可以作為重試時間

5： 表示多久后令牌桶中的令牌會存滿

1.2.2.3、測試

以一個速率稍慢一點的令牌桶來演示一下，連續快速執行以下命令，每次從桶中取出3個令牌，當桶中令牌不足時，請求被拒絕。

127.0.0.1:6379> CL.THROTTLE test2 10 5 60 3
1) (integer) 0
2) (integer) 11
3) (integer) 8
4) (integer) -1
5) (integer) 36
127.0.0.1:6379> CL.THROTTLE test2 10 5 60 3
1) (integer) 0
2) (integer) 11
3) (integer) 5
4) (integer) -1
5) (integer) 71
127.0.0.1:6379> CL.THROTTLE test2 10 5 60 3
1) (integer) 0
2) (integer) 11
3) (integer) 2
4) (integer) -1
5) (integer) 107
127.0.0.1:6379> CL.THROTTLE test2 10 5 60 3
1) (integer) 1
2) (integer) 11
3) (integer) 2
4) (integer) 10
5) (integer) 106

　　在執行限流指令時，如果被拒絕了，就需要丟棄或重試。cl.throttle 指令考慮的非常周 到，連重試時間都幫你算好了，直接取返回結果數組的第四個值進行 sleep 即可，如果不想 阻塞線程，也可以異步定時任務來重試。