一文聊聊Redis中的限流策略

這篇文章帶大家了解Redis中的限流，介紹一下簡單的限流策略和漏斗限流，希望對大家有幫助！

一、簡單的限流

基本原則

當系統處理能力有限，如何組織規劃外的請求對系統施壓。首先我們先來看看一些簡單的限流策略，防止暴力攻擊。例如要對IP訪問，沒5s只能訪問10次，超過進行攔截。 【相關建議：Redis影片教學】

#如上圖，一般使用滑動視窗來統計區間時間內的造訪次數。 使用zset 記錄IP 存取次數，每個IP 透過key 儲存下來，score 儲存目前時間戳，value 唯一用時間戳記或UUID來實作

程式碼實作

public class RedisLimiterTest {
    private Jedis jedis;

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

    /**
     * @param ipAddress Ip地址
     * @param period    特定的时间内，单位秒
     * @param maxCount  最大允许的次数
     * @return
     */
    public boolean isIpLimit(String ipAddress, int period, int maxCount) {
        String key = String.format("ip:%s", ipAddress);
        // 毫秒时间戳
        long currentTimeMillis = System.currentTimeMillis();
        Pipeline pipe = jedis.pipelined();
        // redis事务，保证原子性
        pipe.multi();
        // 存放数据，value 和 score 都使用毫秒时间戳
        pipe.zadd(key, currentTimeMillis, "" + UUID.randomUUID());
        // 移除窗口区间所有的元素
        pipe.zremrangeByScore(key, 0, currentTimeMillis - period * 1000);
        // 获取时间窗口内的行为数量
        Response<Long> count = pipe.zcard(key);
        // 设置 zset 过期时间，避免冷用户持续占用内存，这里宽限1s
        pipe.expire(key, period + 1);
        // 提交事务
        pipe.exec();
        pipe.close();
        // 比较数量是否超标
        return count.get() > maxCount;
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);
        RedisLimiterTest limiter = new RedisLimiterTest(jedis);
        for (int i = 1; i <= 20; i++) {
            // 验证IP  10秒钟之内只能访问5次
            boolean isLimit = limiter.isIpLimit("222.73.55.22", 10, 5);
            System.out.println("访问第" + i + "次, 结果：" + (isLimit ? "限制访问" : "允许访问"));
        }
    }
}

執行結果##

访问第1次, 结果：允许访问
访问第2次, 结果：允许访问
访问第3次, 结果：允许访问
访问第4次, 结果：允许访问
访问第5次, 结果：允许访问
访问第6次, 结果：限制访问
访问第7次, 结果：限制访问
... ...

缺點：要記錄時間窗口所有的行為記錄，量很大，比如，限定60s內不能超過100萬次這種場景，不太適合這樣限流，因為會消耗大量的儲存空間。

二、漏斗限流

基本原理

#漏斗的容量是限定的，如果滿了，就裝不進去了。
如果將漏嘴放開，水就會往下流，流走一部分之後，就又可以繼續往裡面灌水。
如果漏嘴流水的速率大於灌水的速率，那麼漏斗永遠都裝不滿。
如果漏嘴流水速率小於灌水的速率，那麼一旦漏斗滿了，灌水就需要暫停並等待漏斗騰空。

範例程式碼

public class FunnelLimiterTest {

    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<>();

    /**
     * @param capacity    漏斗容量
     * @param leakingRate 漏嘴流水速率 quota/s
     */
    public boolean isIpLimit(String ipAddress, int capacity, float leakingRate) {
        String key = String.format("ip:%s", ipAddress);
        Funnel funnel = funnels.get(key);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate);
            funnels.put(key, funnel);
        }
        return !funnel.watering(1); // 需要1个quota
    }

    public static void main(String[] args) throws Exception{
        FunnelLimiterTest limiter = new FunnelLimiterTest();
        for (int i = 1; i <= 50; i++) {
            // 每1s执行一次
            Thread.sleep(1000);
            // 漏斗容量是2 ，漏嘴流水速率是0.5每秒，
            boolean isLimit = limiter.isIpLimit("222.73.55.22", 2, (float)0.5/1000);
            System.out.println("访问第" + i + "次, 结果：" + (isLimit ? "限制访问" : "允许访问"));
        }
    }
}

執行結果

访问第1次, 结果：允许访问    # 第1次，容量剩余2，执行后1
访问第2次, 结果：允许访问    # 第2次，容量剩余1，执行后0
访问第3次, 结果：允许访问    # 第3次，由于过了2s, 漏斗流水剩余1个空间，所以容量剩余1，执行后0
访问第4次, 结果：限制访问    # 第4次，过了1s, 剩余空间小于1， 容量剩余0
访问第5次, 结果：允许访问    # 第5次，由于过了2s, 漏斗流水剩余1个空间，所以容量剩余1，执行后0
访问第6次, 结果：限制访问    # 以此类推...
访问第7次, 结果：允许访问
访问第8次, 结果：限制访问
访问第9次, 结果：允许访问
访问第10次, 结果：限制访问

我們觀察
• Funnel 物件的幾個字段，我們發現可以將Funnel 物件的內容按字段儲存到一個hash 結構中，灌水的時候將hash 結構的字段取出來進行邏輯運算後，再將新值回填到hash 結構中就完成了一次行為頻度的偵測。
但是有個問題，我們無法保證整個過程的原子性。從
• hash 結構中取值，然後在記憶體中運算，再回填到 hash 結構，這三個過程無法原子化，這意味著需要適當的加鎖控制。而一旦加鎖，就代表會有加鎖失敗，加鎖失敗就需要選擇重試或放棄。
如果重試的話，就會導致效能下降。如果放棄的話，就會影響使用者體驗。同時，程式碼的複雜度也跟著升高很多。這真是個艱難的選擇，我們該如何解決這個問題呢？
• Redis-Cell 救星來了！

Redis-Cell

Redis 4.0 提供了一個限流 Redis 模組，它叫做

redis-cell。該模組也使用了漏斗演算法，並提供了原子的限流指令。 模組只有1條指令cl.throttle，它的參數和回傳值都略顯複雜，接下來讓我們來看看這個指令具體該如何使用。

> cl.throttle key:xxx 15 30 60 1

• 15 : 15 capacity 這是漏斗容量
• 30 60 : 30 operations / 60 seconds 這是漏水速率
• 1 : need 1 quota (可選參數，預設值也是1)

> cl.throttle laoqian:reply 15 30 60
1) (integer) 0   # 0 表示允许，1表示拒绝
2) (integer) 15  # 漏斗容量capacity
3) (integer) 14  # 漏斗剩余空间left_quota
4) (integer) -1  # 如果拒绝了，需要多长时间后再试(漏斗有空间了，单位秒)
5) (integer) 2   # 多长时间后，漏斗完全空出来(left_quota==capacity，单位秒)

在執行限流指令時，如果被拒絕了，就需要丟棄或重試。

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

更多程式相關知識，請造訪：

程式設計影片！ ！

以上是一文聊聊Redis中的限流策略的詳細內容。更多資訊請關注PHP中文網其他相關文章！