保護高并發(fā)系統的三大利器：緩存、降級和限流。那什么是限流呢？用我沒讀過太多書的話來講，限流就是限制流量。我們都知道服務器的處理能力是有上限的，如果超過了上限繼續(xù)放任請求進來的話，可能會發(fā)生不可控的后果。而通過限流，在請求數量超出閾值的時候就排隊等待甚至拒絕服務，就可以使系統在扛不住過高并發(fā)的情況下做到有損服務而不是不服務。

舉個例子，如各地都出現口罩緊缺的情況，廣州政府為了緩解市民買不到口罩的狀況，上線了預約服務，只有預約到的市民才能到指定的藥店購買少量口罩。這就是生活中限流的情況，說這個也是希望大家這段時間保護好自己，注意防護 ：）

接下來就跟大家分享下接口限流的常見玩法吧，部分算法用python+redis粗略實現了一下，關鍵是圖解啊！你品，你細品~

固定窗口法

固定窗口法是限流算法里面最簡單的，比如我想限制1分鐘以內請求為100個，從現在算起的一分鐘內，請求就最多就是100個，這分鐘過完的那一刻把計數器歸零，重新計算，周而復始。

偽代碼實現

def can_pass_fixed_window(user, action, time_zone=60, times=30): ''' :param user: 用戶唯一標識 :param action: 用戶訪問的接口標識(即用戶在客戶端進行的動作) :param time_zone: 接口限制的時間段 :param time_zone: 限制的時間段內允許多少請求通過 ''' key = ’{}:{}’.format(user, action) # redis_conn 表示redis連接對象 count = redis_conn.get(key) if not count: count = 1 redis_conn.setex(key, time_zone, count) if count < times: redis_conn.incr(key) return True return False

這個方法雖然簡單，但有個大問題是無法應對兩個時間邊界內的突發(fā)流量。如上圖所示，如果在計數器清零的前1秒以及清零的后1秒都進來了100個請求，那么在短時間內服務器就接收到了兩倍的(200個)請求，這樣就有可能壓垮系統。會導致上面的問題是因為我們的統計精度還不夠，為了將臨界問題的影響降低，我們可以使用滑動窗口法。

滑動窗口法

滑動窗口法，簡單來說就是隨著時間的推移，時間窗口也會持續(xù)移動，有一個計數器不斷維護著窗口內的請求數量，這樣就可以保證任意時間段內，都不會超過最大允許的請求數。例如當前時間窗口是0s~60s，請求數是40，10s后時間窗口就變成了10s~70s，請求數是60。

時間窗口的滑動和計數器可以使用redis的有序集合(sorted set)來實現。score的值用毫秒時間戳來表示，可以利用當前時間戳-時間窗口的大小來計算出窗口的邊界，然后根據score的值做一個范圍篩選就可以圈出一個窗口；value的值僅作為用戶行為的唯一標識，也用毫秒時間戳就好。最后統計一下窗口內的請求數再做判斷即可。

偽代碼實現

def can_pass_slide_window(user, action, time_zone=60, times=30): ''' :param user: 用戶唯一標識 :param action: 用戶訪問的接口標識(即用戶在客戶端進行的動作) :param time_zone: 接口限制的時間段 :param time_zone: 限制的時間段內允許多少請求通過 ''' key = ’{}:{}’.format(user, action) now_ts = time.time() * 1000 # value是什么在這里并不重要，只要保證value的唯一性即可，這里使用毫秒時間戳作為唯一值 value = now_ts # 時間窗口左邊界 old_ts = now_ts - (time_zone * 1000) # 記錄行為 redis_conn.zadd(key, value, now_ts) # 刪除時間窗口之前的數據 redis_conn.zremrangebyscore(key, 0, old_ts) # 獲取窗口內的行為數量 count = redis_conn.zcard(key) # 設置一個過期時間免得占空間 redis_conn.expire(key, time_zone + 1) if not count or count < times: return True return False

雖然滑動窗口法避免了時間界限的問題，但是依然無法很好解決細時間粒度上面請求過于集中的問題，就例如限制了1分鐘請求不能超過60次，請求都集中在59s時發(fā)送過來，這樣滑動窗口的效果就大打折扣。 為了使流量更加平滑，我們可以使用更加高級的令牌桶算法和漏桶算法。

令牌桶法

令牌桶算法的思路不復雜，它先以固定的速率生成令牌，把令牌放到固定容量的桶里，超過桶容量的令牌則丟棄，每來一個請求則獲取一次令牌，規(guī)定只有獲得令牌的請求才能放行，沒有獲得令牌的請求則丟棄。

偽代碼實現

def can_pass_token_bucket(user, action, time_zone=60, times=30): ''' :param user: 用戶唯一標識 :param action: 用戶訪問的接口標識(即用戶在客戶端進行的動作) :param time_zone: 接口限制的時間段 :param time_zone: 限制的時間段內允許多少請求通過 ''' # 請求來了就倒水，倒水速率有限制 key = ’{}:{}’.format(user, action) rate = times / time_zone # 令牌生成速度 capacity = times # 桶容量 tokens = redis_conn.hget(key, ’tokens’) # 看桶中有多少令牌 last_time = redis_conn.hget(key, ’last_time’) # 上次令牌生成時間 now = time.time() tokens = int(tokens) if tokens else capacity last_time = int(last_time) if last_time else now delta_tokens = (now - last_time) * rate # 經過一段時間后生成的令牌 if delta_tokens > 1: tokens = tokens + tokens # 增加令牌 if tokens > tokens: tokens = capacity last_time = time.time() # 記錄令牌生成時間 redis_conn.hset(key, ’last_time’, last_time) if tokens >= 1: tokens -= 1 # 請求進來了，令牌就減少1 redis_conn.hset(key, ’tokens’, tokens) return True return False

令牌桶法限制的是請求的平均流入速率，優(yōu)點是能應對一定程度上的突發(fā)請求，也能在一定程度上保持流量的來源特征，實現難度不高，適用于大多數應用場景。

漏桶算法

漏桶算法的思路與令牌桶算法有點相反。大家可以將請求想象成是水流，水流可以任意速率流入漏桶中，同時漏桶以固定的速率將水流出。如果流入速度太大會導致水滿溢出，溢出的請求被丟棄。

通過上圖可以看出漏桶法的特點是：不限制請求流入的速率，但是限制了請求流出的速率。這樣突發(fā)流量可以被整形成一個穩(wěn)定的流量，不會發(fā)生超頻。

關于漏桶算法的實現方式有一點值得注意，我在瀏覽相關內容時發(fā)現網上大多數對于漏桶算法的偽代碼實現，都只是實現了

根據維基百科，漏桶算法的實現理論有兩種，分別是基于 meter 的和基于 queue 的，他們實現的具體思路不同，我大概介紹一下。

基于meter的漏桶

基于 meter 的實現相對來說比較簡單，其實它就有一個計數器，然后有消息要發(fā)送的時候，就看計數器夠不夠，如果計數器沒有滿的話，那么這個消息就可以被處理，如果計數器不足以發(fā)送消息的話，那么這個消息將會被丟棄。

那么這個計數器是怎么來的呢，基于 meter 的形式的計數器就是發(fā)送的頻率，例如你設置得頻率是不超過 5條/s ，那么計數器就是 5，在一秒內你每發(fā)送一條消息就減少一個，當你發(fā)第 6 條的時候計時器就不夠了，那么這條消息就被丟棄了。

這種實現有點類似最開始介紹的固定窗口法，只不過時間粒度再小一些，偽代碼就不上了。

基于queue的漏桶

基于 queue 的實現起來比較復雜，但是原理卻比較簡單，它也存在一個計數器，這個計數器卻不表示速率限制，而是表示 queue 的大小，這里就是當有消息要發(fā)送的時候看 queue 中是否還有位置，如果有，那么就將消息放進 queue 中，這個 queue 以 FIFO 的形式提供服務；如果 queue 沒有位置了，消息將被拋棄。

在消息被放進 queue 之后，還需要維護一個定時器，這個定時器的周期就是我們設置的頻率周期，例如我們設置得頻率是 5條/s，那么定時器的周期就是 200ms，定時器每 200ms 去 queue 里獲取一次消息，如果有消息，那么就發(fā)送出去，如果沒有就輪空。

注意，網上很多關于漏桶法的偽代碼實現只實現了水流入桶的部分，沒有實現關鍵的水從桶中漏出的部分。如果只實現了前半部分，其實跟令牌桶沒有大的區(qū)別噢😯

如果覺得上面的都太難，不好實現，那么我墻裂建議你嘗試一下redis-cell這個模塊！

redis-cell

Redis 4.0 提供了一個限流 Redis 模塊，它叫 redis-cell。該模塊也使用了漏斗算法，并提供了原子的限流指令。有了這個模塊，限流問題就非常簡單了。 這個模塊需要單獨安裝，安裝教程網上很多，它只有一個指令：

CL.THROTTLE

CL.THROTTLE user123 15 30 60 1▲ ▲ ▲ ▲ ▲| | | | └───── apply 1 operation (default if omitted) 每次請求消耗的水滴| | └──┴─────── 30 operations / 60 seconds 漏水的速率| └───────────── 15 max_burst 漏桶的容量└─────────────────── key “user123” 用戶行為

執(zhí)行以上命令之后，redis會返回如下信息：

> cl.throttle laoqian:reply 15 30 601) (integer) 0 # 0 表示允許，1表示拒絕2) (integer) 16 # 漏桶容量3) (integer) 15 # 漏桶剩余空間left_quota4) (integer) -1 # 如果拒絕了，需要多長時間后再試(漏桶有空間了，單位秒)5) (integer) 2 # 多長時間后，漏桶完全空出來(單位秒)

有了上面的redis模塊，就可以輕松對付大多數的限流場景了。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持好吧啦網。