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

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

接下來就跟大家分享下介面限流的常見玩法吧，部分演算法用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

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

滑動視窗法

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

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

虛擬碼實現

def can_pass_slide_window(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,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時傳送過來，這樣滑動視窗的效果就大打折扣。 為了使流量更加平滑，我們可以使用更加高階的令牌桶演算法和漏桶演算法。

令牌桶法

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

虛擬碼實現

def can_pass_token_bucket(user,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

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

漏桶演算法

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

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

關於漏桶演算法的實現方式有一點值得注意，我在瀏覽相關內容時發現網上大多數對於漏桶演算法的虛擬碼實現，都只是實現了

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

基於meter的漏桶

基於 meter 的實現相對來說比較簡單，其實它就有一個計數器，然後有訊息要傳送的時候，就看計數器夠不夠，如果計數器沒有滿的話，那麼這個訊息就可以被處理，如果計數器不足以傳送訊息的話，那麼這個訊息將會被丟棄。

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

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

基於queue的漏桶

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

在訊息被放進 queue 之後，還需要維護一個定時器，這個定時器的週期就是我們設定的頻率週期，例如我們設定得頻率是 5條/s，那麼定時器的週期就是 200ms，定時器每 200ms 去 queue 裡獲取一次訊息，如果有訊息，那麼就傳送出去，如果沒有就輪空。

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

如果覺得上面的都太難，不好實現，那麼我牆裂建議你嘗試一下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” 使用者行為

執行以上命令之後，redis會返回如下資訊：

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

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

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。