[[428295]]

為什么要限流

日常生活中，有哪些需要限流的地方？

像我旁邊有一個國家景區，平時可能根本沒什么人前往，但是一到五一或者春節就人滿為患，這時候景區管理人員就會實行一系列的政策來限制進入人流量。

為什么要限流呢？假如景區能容納一萬人，現在進去了三萬人，勢必摩肩接踵，整不好還會有事故發生，這樣的結果就是所有人的體驗都不好，如果發生了事故景區可能還要關閉，導致對外不可用，這樣的后果就是所有人都覺得體驗糟糕透了。

限流的思想就是在保證可用的情況下盡可能多增加進入的人數，其余的人在外面排隊等待，保證里面的一萬人可以正常游玩。

回到網絡上，同樣也是這個道理，例如某某明星公布了戀情，訪問從平時的50萬增加到了500萬，系統最多可以支撐200萬訪問，那么就要執行限流規則，保證是一個可用的狀態，不至于服務器崩潰導致所有請求不可用。

限流思路

對系統服務進行限流，一般有如下幾個模式：

熔斷

系統在設計之初就把熔斷措施考慮進去。當系統出現問題時，如果短時間內無法修復，系統要自動做出判斷，開啟熔斷開關，拒絕流量訪問，避免大流量對后端的過載請求。

系統也應該能夠動態監測后端程序的修復情況，當程序已恢復穩定時，可以關閉熔斷開關，恢復正常服務。常見的熔斷組件有Hystrix以及阿里的Sentinel，兩種互有優缺點，可以根據業務的實際情況進行選擇。

服務降級

將系統的所有功能服務進行一個分級，當系統出現問題需要緊急限流時，可將不是那么重要的功能進行降級處理，停止服務，這樣可以釋放出更多的資源供給核心功能的去用。

例如在電商平臺中，如果突發流量激增，可臨時將商品評論、積分等非核心功能進行降級，停止這些服務，釋放出機器和CPU等資源來保障用戶正常下單，而這些降級的功能服務可以等整個系統恢復正常后，再來啟動，進行補單/補償處理。除了功能降級以外，還可以采用不直接操作數據庫，而全部讀緩存、寫緩存的方式作為臨時降級方案。

延遲處理

這個模式需要在系統的前端設置一個流量緩沖池，將所有的請求全部緩沖進這個池子，不立即處理。然后后端真正的業務處理程序從這個池子中取出請求依次處理，常見的可以用隊列模式來實現。這就相當于用異步的方式去減少了后端的處理壓力，但是當流量較大時，后端的處理能力有限，緩沖池里的請求可能處理不及時，會有一定程度延遲。后面具體的漏桶算法以及令牌桶算法就是這個思路。

特權處理

這個模式需要將用戶進行分類，通過預設的分類，讓系統優先處理需要高保障的用戶群體，其它用戶群的請求就會延遲處理或者直接不處理。

緩存、降級、限流區別

• 緩存是用來增加系統吞吐量，提升訪問速度提供高并發。
• 降級是在系統某些服務組件不可用的時候、流量暴增、資源耗盡等情況下，暫時屏蔽掉出問題的服務，繼續提供降級服務，給用戶盡可能的友好提示，返回兜底數據，不會影響整體業務流程，待問題解決再重新上線服務
• 限流是指在使用緩存和降級無效的場景。比如當達到閾值后限制接口調用頻率，訪問次數，庫存個數等，在出現服務不可用之前，提前把服務降級。只服務好一部分用戶。

 限流的算法

限流算法很多，常見的有三類，分別是計數器算法、漏桶算法、令牌桶算法，下面逐一講解。

計數器算法

簡單粗暴，比如指定線程池大小，指定數據庫連接池大小、Nginx連接數等，這都屬于計數器算法。

計數器算法是限流算法里最簡單也是最容易實現的一種算法。舉個例子，比如我們規定對于A接口，我們1分鐘的訪問次數不能超過100個。那么我們可以這么做：在一開始的時候，我們可以設置一個計數器counter，每當一個請求過來的時候，counter就加1，如果counter的值大于100并且該請求與第一個請求的間隔時間還在1分鐘之內，那么說明請求數過多，拒絕訪問；如果該請求與第一個請求的間隔時間大于1分鐘，且counter的值還在限流范圍內，那么就重置counter，就是這么簡單粗暴。

漏桶算法

漏桶算法思路很簡單，水（請求）先進入到漏桶里，漏桶以一定的速度出水，當水流入速度過大會超過桶可接納的容量時直接溢出，可以看出漏桶算法能強行限制數據的傳輸速率。

這樣做的好處是：

• 削峰：有大量流量進入時，會發生溢出，從而限流保護服務可用
• 緩沖：不至于直接請求到服務器，緩沖壓力

令牌桶算法

令牌桶與漏桶相似，不同的是令牌桶桶中放了一些令牌，服務請求到達后，要獲取令牌之后才會得到服務，舉個例子，我們平時去食堂吃飯，都是在食堂內窗口前排隊的，這就好比是漏桶算法，大量的人員聚集在食堂內窗口外，以一定的速度享受服務，如果涌進來的人太多，食堂裝不下了，可能就有一部分人站到食堂外了，這就沒有享受到食堂的服務，稱之為溢出，溢出可以繼續請求，也就是繼續排隊，那么這樣有什么問題呢？

如果這時候有特殊情況，比如有些趕時間的志愿者啦、或者高三要高考啦，這種情況就是突發情況，如果也用漏桶算法那也得慢慢排隊，這也就沒有解決我們的需求，對于很多應用場景來說，除了要求能夠限制數據的平均傳輸速率外，還要求允許某種程度的突發傳輸。這時候漏桶算法可能就不合適了，令牌桶算法更為適合。如圖所示，令牌桶算法的原理是系統會以一個恒定的速度往桶里放入令牌，而如果請求需要被處理，則需要先從桶里獲取一個令牌，當桶里沒有令牌可取時，則拒絕服務。

令牌桶好處就是，如果某一瞬間訪問量劇增或者有突發情況，可以通過改變桶中令牌數量來改變連接數，就好比那個食堂排隊吃飯的問題，如果現在不是直接去窗口排隊，而是先來樓外拿飯票然后再去排隊，那么有高三的學生時可以將增加飯票數量或者優先將令牌給高三的學生，這樣比漏桶算法更加靈活。

并發限流

簡單來說就是設置系統閾值總的QPS個數，這些也挺常見的，就拿Tomcat來說，很多參數就是出于這個考慮，例如配置的acceptCount設置響應連接數，maxConnections設置瞬時最大連接數，maxThreads設置最大線程數，在各個框架或者組件中，并發限流體現在下面幾個方面：

• 限制總并發數（如數據庫連接池、線程池）
• 限制瞬時并發數（Nginx的limit_conn模塊，用來限制瞬時并發連接數）
• 限制時間窗口內的平均速率（如Guava的RateLimiter、Nginx的limit_req模塊，限制每秒的平均速率）
• 其他的還有限制遠程接口調用速率、限制MQ的消費速率。
• 另外還可以根據網絡連接數、網絡流量、CPU或內存負載等來限流。

有了并發限流，就意味著在處理高并發的時候多了一種保護機制，不用擔心瞬間流量導致系統掛掉或雪崩，最終做到有損服務而不是不服務；但是限流需要評估好，不能亂用，否則一些正常流量出現一些奇怪的問題而導致用戶體驗很差造成用戶流失。

接口限流

接口限流分為兩個部分，一是限制一段時間內接口調用次數，參照前面限流算法的計數器算法，二是設置滑動時間窗口算法。

接口總數

控制一段時間內接口被調用的總數量，可以參考前面的計數器算法，不再贅述。

接口時間窗口

固定時間窗口算法（也就是前面提到的計數器算法）的問題是統計區間太大，限流不夠精確，而且在第二個統計區間時沒有考慮與前一個統計區間的關系與影響（第一個區間后半段 + 第二個區間前半段也是一分鐘）。為了解決上面我們提到的臨界問題，我們試圖把每個統計區間分為更小的統計區間，更精確的統計計數。

在上面的例子中，假設QPS可以接受100次查詢/秒，前一分鐘前40秒訪問很低，后20秒突增，并且這個持續了一段時間，直到第二分鐘的第40秒才開始降下來，根據前面的計數方法，前一秒的QPS為94，后一秒的QPS為92，那么沒有超過設定參數，但是！但是在中間區域，QPS達到了142，這明顯超過了我們的允許的服務請求數目，所以固定窗口計數器不太可靠，需要滑動窗口計數器。

計數器算法其實就是固定窗口算法，只是它沒有對時間窗口做進一步地劃分，所以只有1格；由此可見，當滑動窗口的格子劃分的越多，也就是將秒精確到毫秒或者納秒，那么滑動窗口的滾動就越平滑，限流的統計就會越精確。需要注意的是，消耗的空間就越多。

限流實現

這一部分是限流的具體實現，簡單說說，畢竟長篇代碼沒人愿意看。

guava實現

引入包：

<!-- https://mvnrepository.com/artifact/com.google.guava/guava --> 
<dependency> 
<groupId>com.google.guava</groupId> 
<artifactId>guava</artifactId> 
<version>28.1-jre</version> 
</dependency> 

核心代碼：

LoadingCache<Long, AtomicLong> counter = CacheBuilder.newBuilder(). 
            expireAfterWrite(2, TimeUnit.SECONDS) 
            .build(new CacheLoader<Long, AtomicLong>() { 
 
                @Override 
                public AtomicLong load(Long secend) throws Exception { 
                    // TODO Auto-generated method stub 
                    return new AtomicLong(0); 
                } 
            }); 
    counter.get(1l).incrementAndGet(); 

 令牌桶實現

穩定模式（SmoothBursty：令牌生成速度恒定）：

public static void main(String[] args) { 
    // RateLimiter.create(2)每秒產生的令牌數 
    RateLimiter limiter = RateLimiter.create(2); 
    // limiter.acquire() 阻塞的方式獲取令牌 
    System.out.println(limiter.acquire());; 
    try { 
        Thread.sleep(2000); 
    } catch (InterruptedException e) { 
        // TODO Auto-generated catch block 
        e.printStackTrace(); 
    } 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
}  

RateLimiter.create(2)容量和突發量，令牌桶算法允許將一段時間內沒有消費的令牌暫存到令牌桶中，用來突發消費。

漸進模式（SmoothWarmingUp：令牌生成速度緩慢提升直到維持在一個穩定值）：

// 平滑限流，從冷啟動速率（滿的）到平均消費速率的時間間隔 
    RateLimiter limiter = RateLimiter.create(2,1000l,TimeUnit.MILLISECONDS); 
    System.out.println(limiter.acquire());; 
    try { 
        Thread.sleep(2000); 
    } catch (InterruptedException e) { 
        // TODO Auto-generated catch block 
        e.printStackTrace(); 
    } 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 
 
    System.out.println(limiter.acquire());; 
    System.out.println(limiter.acquire());; 

超時：

boolean tryAcquire = limiter.tryAcquire(Duration.ofMillis(11)); 

在timeout時間內是否能夠獲得令牌，異步執行。

分布式系統限流

Nginx + Lua實現

可以使用resty.lock保持原子特性，請求之間不會產生鎖的重入。

https://github.com/openresty/lua-resty-lock

使用lua_shared_dict存儲數據：

local locks = require "resty.lock" 
 
local function acquire() 
local lock =locks:new("locks") 
local elapsed, err =lock:lock("limit_key") --互斥鎖 保證原子特性 
local limit_counter =ngx.shared.limit_counter --計數器 
 
local key = "ip:" ..os.time() 
local limit = 5 --限流大小 
local current =limit_counter:get(key) 
 
if current ~= nil and current + 1> limit then --如果超出限流大小 
   lock:unlock() 
   return 0 
end 
if current == nil then 
   limit_counter:set(key, 1, 1) --第一次需要設置過期時間，設置key的值為1， 
--過期時間為1秒 
else 
    limit_counter:incr(key, 1) --第二次開始加1即可 
end 
lock:unlock() 
return 1 
end 
ngx.print(acquire())