限流實踐

高併發系統三把利器：快取、降級和限流
限流大家多少都有所瞭解，就是限制當前請求流量、數量，避免系統被蜂擁而至的流量瞬間擊垮，剛好最近手上業務系統做了限流，順便整理出來。

我們的業務場景

• 雲變數和雲列表，使用者線上使用的程式變數和陣列，一些作品執行人數過多或者作者濫用（for迴圈中修改變數或陣列），大量請求訊息落到kafka，而我們消費機器能力有限，造成訊息堵塞。直接的影響就是業務不能正常使用，頻繁fullgc。
• 無論什麼系統，業務上的優化往往更能立竿見影，且成本較低。這裡不做詳述，比較偏業務，主要思路是儘量減少落入kafka中的無效訊息，合併重複訊息。

限流方式

• 限制總併發數（比如資料庫連線池、執行緒池）
• 限制瞬時併發數（如nginx的limit_conn模組，用來限制瞬時併發連線數）
• 限制時間視窗內的平均速率（如Guava的RateLimiter、nginx的limit_req模組，限制每秒的平均速率）
• 限制遠端介面呼叫速率
• 限制MQ的消費速率。
• 可以根據網路連線數、網路流量、CPU或記憶體負載等來限流

限流演算法

Google大法了相關資料，總結常用的限流演算法如下：

• 計數器
  • 通過一個計數器來記錄一定時間內某個介面的訪問數量，超過閾值則不允許繼續訪問，或者後續的請求放入佇列，計數器到下一個時間段清零，這裡缺點是，假設設定每3秒100，第3秒99個請求數，然後到了下一個時間段，計數被清零，第4秒又來了99個請求，實際上3-4兩秒時間超出了限制的100個請求，但這種情況較少。所有有了下面的時間視窗讓它更平滑
• 時間視窗
  • 建立時間視窗，100個格子，每個格子為100毫秒，每100毫秒滑動一格，判斷當前執行次數和前一次執行次數差是否超出閾值，沒有超過記錄當前次數，超過證明超出了限流次數，視窗粒度越細則越平滑
• 漏桶演算法
  • 把請求固定丟入‘漏桶’（佇列），執行緒以固定頻率進行消費處理，這樣對應用來說消費是固定的壓力，超出漏桶容量的請求將被丟棄
• 令牌桶限流
  • 基本思路為每秒生成固定數量令牌置入桶內，真正業務操作需要從桶中獲取令牌才能執行，沒有令牌拋棄請求，相對漏桶方式，更靈活，可以設定令牌生成的速度，獲取令牌的速度和個數。允許一些突發的流量（如果當前機器可以支撐的流量（令牌數）較多則可以更多的獲取令牌進行執行）

可選方案

• Guava 提供的令牌桶實現RateLimiter，但只支援單機的限流
• nginx 連線數限流模組ngx_http_limit_conn_module和漏桶演算法實現的請求限流模組ngx_http_limit_req_module
• redis+lua 指令碼實現
• ratelimitj開源框架（提供了基於redis、hazelcast、inmemory版本的實現方案）
• 阿里巴巴開源的 Sentinel（限流、熔斷降級、塑形、系統負載保護）

落地方案

• 一般我們這裡都是叢集環境，現在最早已經使用了guava RateLimiter，這裡系統也已經使用了redis，所以選擇了redis進行了實現

• redis+lua

  • redis可以呼叫lua指令碼，lua程式碼將會被當成一個命令去執行，lua指令碼執行完成redis才會繼續執行其他命令，這樣保證了併發環境下可以正確的進行執行限流判斷操作。

  • lua指令碼（基本為記錄對應請求對應key計數是否超出限制數量）

    -- ratelimit 物件
    RateLimit = { acquire = false, key = nil, limit = nil, count = 0 }
    function RateLimit:new(o, acquire, key, limit, count)
        o = o or {}
        setmetatable(o, self)
        self.__index = self
        self.acquire = acquire or false
        self.key = key or nil
        self.limit = limit or 0
        self.count = count or 0;
        return o
    end
    
    -- 獲取執行許可權，超出流量控制則拒絕
    local key = "rate.limit:" .. KEYS[1]
    local limit = tonumber(KEYS[2])
    local expire_time = KEYS[3]
    -- 判斷key是否存在
    local is_exists = redis.call("EXTSTS", key);
    -- 如果key存在則呼叫inscr操作key
    -- 如果值大於limit數量返回0，否則返回1
    if is_exists == 1 then
        local count = redis.call("INSCR", key)
        if redis.call("INSCR", key) > limit then
            return Rectangle:new(nil, false, key, limit, count)
        else
            return Rectangle:new(nil, true, key, limit, count)
        end
        -- 如果不存在key，則呼叫set命令設定key的值為1
        -- 設定key過期時間
    else
        redis.call("SET", key, 1)
        redis.call("EXPIRE", key, expire_time)
        return Rectangle:new(nil, true, key, limit, 1)
    end
    

  • 封裝java介面（這裡加了log，可以簡單監控下當前限流情況，對異常請求可以根據業務需要做一些人工介入）

    package com.netease.edu.scratch.cloud.limit;
    
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 限流
     *
     * @author zhangchanglu<[email protected]  corp.netease.com>
     * @since 2018/08/08 22:08.
     */
    public interface RateLimit {
        /**
         * 獲取執行許可權
         *
         * @param key    業務key
         * @param limit  限流大小
         * @param expire 限流重置時間
         * @return 流量情況
         */
        RateLimitVO acquire(String key, int limit, long expire, TimeUnit timeUnit);
    
        /**
         * 監控當前key流量情況
         *
         * @param key 業務key
         * @param limit 限制流量數
         * @return 流量情況
         */
        RateLimitVO log(String key, int limit);
        /**
         * 監控當前所有key流量情況
         * @param limit 限制流量數
         * @return 流量情況
         */
        Map<String,RateLimitVO> logAll(int limit);
    
        /**
         * 獲取儲存redis key
         * @param key 業務key
         * @return 業務key
         */
        String getKey(String key);
    
    }
    
    

  • 主要實現方法（這裡用的spring的redisTemplate，呼叫傳遞value沒辦法正常獲取，所以所有引數都是用了key進行了傳遞

    /**
         * 獲取執行許可權
         *
         * @param key    業務key
         * @param limit  限流大小
         * @param expire 限流重置時間
         * @return 是否在允許執行
         */
        public RateLimitVO acquire(String key, int limit, long expire, TimeUnit timeUnit) {
            addLog(key);
            key = getKey(key);
            RedisScript<Long> redisScript = new DefaultRedisScript<>(scriptStr, Long.class);
            Long count = redisTemplate.execute(redisScript, Lists.newArrayList(key, String.valueOf(limit), String.valueOf(timeUnit.toSeconds(expire))));
            RateLimitVO rateLimitVO = rateLimitHelper.getRateLimitVO(key, limit, count);
            log.info("當前流量情況：" + rateLimitVO);
            return rateLimitVO;
        }
    

• 這裡我們用的是我們的rds，但跟dba溝通後，為了安全，限制了eval命令的呼叫，而redis呼叫lua指令碼是依賴eval命令的，所以這裡用使用程式碼進行了實現，思路一樣的只是藉助了分散式鎖。

存在的問題

• 我們業務上也是粗略進行限流，並沒有精確進行限制流量，基本上避免大量無效和惡意請求即可
• 這裡的限流其實對於整個系統來講，系統已經接受了大量的連線，依然會有壓力存在，只是減少了kafka、資料庫的壓力 ，所以對限流來說應該越靠前越好，使用nginx來進行限流應該能更好的保證系統的穩定性，這也是我們後續要考慮做的。