一篇帶給你 Sentinel 和常用流控算法

作者：開發者老鄭 2021-05-25 08:01:55

本文主要講述常見的幾種限流算法：計數器算法、漏桶算法、令牌桶算法。然后結合我對 Sentinel 1.8.0 的理解，給大家分享 Sentinel 在源碼中如何使用這些算法進行流控判斷。

計數器限流算法

我們可以直接通過一個計數器，限制每一秒鐘能夠接收的請求數。比如說 qps定為 1000，那么實現思路就是從第一個請求進來開始計時，在接下去的 1s 內，每來一個請求，就把計數加 1，如果累加的數字達到了 1000，那么后續的請求就會被全部拒絕。等到 1s 結束后，把計數恢復成 0 ，重新開始計數。

優點：實現簡單

缺點：如果1s 內的前半秒，已經通過了 1000 個請求，那后面的半秒只能請求拒絕，我們把這種現象稱為“突刺現象”。

實現代碼案例：

public class Counter { 
    public long timeStamp = getNowTime(); 
    public int reqCount = 0; 
    public final int limit = 100; // 時間窗口內最大請求數 
    public final long interval = 1000; // 時間窗口ms 
 
    public boolean limit() { 
        long now = getNowTime(); 
        if (now < timeStamp + interval) { 
            // 在時間窗口內 
            reqCount++; 
            // 判斷當前時間窗口內是否超過最大請求控制數 
            return reqCount <= limit; 
        } else { 
            timeStamp = now; 
            // 超時后重置 
            reqCount = 1; 
            return true; 
        } 
    } 
 
    public long getNowTime() { 
        return System.currentTimeMillis(); 
    } 
}

滑動時間窗算法

滑動窗口，又稱 Rolling Window。為了解決計數器算法的缺陷，我們引入了滑動窗口算法。下面這張圖，很好地解釋了滑動窗口算法：

在上圖中，整個紅色的矩形框表示一個時間窗口，在我們的例子中，一個時間窗口就是一分鐘。然后我們將時間窗口進行劃分，比如圖中，我們就將滑動窗口劃成了6格，所以每格代表的是10秒鐘。每過10秒鐘，我們的時間窗口就會往右滑動一格。每一個格子都有自己獨立的計數器counter，比如當一個請求在0:35秒的時候到達，那么0:30~0:39對應的counter就會加1。

那么滑動窗口怎么解決剛才的臨界問題的呢?我們可以看上圖，0:59到達的100個請求會落在灰色的格子中，而1:00到達的請求會落在橘黃色的格子中。當時間到達1:00時，我們的窗口會往右移動一格，那么此時時間窗口內的總請求數量一共是200個，超過了限定的100個，所以此時能夠檢測出來觸發了限流。

我再來回顧一下剛才的計數器算法，我們可以發現，計數器算法其實就是滑動窗口算法。只是它沒有對時間窗口做進一步地劃分，所以只有1格。

由此可見，當滑動窗口的格子劃分的越多，那么滑動窗口的滾動就越平滑，限流的統計就會越精確。

實現代碼案例：

public class SlideWindow { 
 
    /** 隊列id和隊列的映射關系，隊列里面存儲的是每一次通過時候的時間戳，這樣可以使得程序里有多個限流隊列 */ 
    private volatile static Map<String, List<Long>> MAP = new ConcurrentHashMap<>(); 
 
    private SlideWindow() {} 
 
    public static void main(String[] args) throws InterruptedException { 
        while (true) { 
            // 任意10秒內，只允許2次通過 
            System.out.println(LocalTime.now().toString() + SlideWindow.isGo("ListId", 2, 10000L)); 
            // 睡眠0-10秒 
            Thread.sleep(1000 * new Random().nextInt(10)); 
        } 
    } 
 
    /** 
     * 滑動時間窗口限流算法 
     * 在指定時間窗口，指定限制次數內，是否允許通過 
     * 
     * @param listId     隊列id 
     * @param count      限制次數 
     * @param timeWindow 時間窗口大小 
     * @return 是否允許通過 
     */ 
    public static synchronized boolean isGo(String listId, int count, long timeWindow) { 
        // 獲取當前時間 
        long nowTime = System.currentTimeMillis(); 
        // 根據隊列id，取出對應的限流隊列，若沒有則創建 
        List<Long> list = MAP.computeIfAbsent(listId, k -> new LinkedList<>()); 
        // 如果隊列還沒滿，則允許通過，并添加當前時間戳到隊列開始位置 
        if (list.size() < count) { 
            list.add(0, nowTime); 
            return true; 
        } 
 
        // 隊列已滿（達到限制次數），則獲取隊列中最早添加的時間戳 
        Long farTime = list.get(count - 1); 
        // 用當前時間戳 減去 最早添加的時間戳 
        if (nowTime - farTime <= timeWindow) { 
            // 若結果小于等于timeWindow，則說明在timeWindow內，通過的次數大于count 
            // 不允許通過 
            return false; 
        } else { 
            // 若結果大于timeWindow，則說明在timeWindow內，通過的次數小于等于count 
            // 允許通過，并刪除最早添加的時間戳，將當前時間添加到隊列開始位置 
            list.remove(count - 1); 
            list.add(0, nowTime); 
            return true; 
        } 
    } 
 
}

在 Sentinel 中通過 LeapArray 結構來實現時間窗算法, 它的核心代碼如下(只列舉獲取時間窗方法)：

/** 
     * 獲取當前的時間窗 
     * 
     * Get bucket item at provided timestamp. 
     * 
     * @param timeMillis a valid timestamp in milliseconds 
     * @return current bucket item at provided timestamp if the time is valid; null if time is invalid 
     */ 
public WindowWrap<T> currentWindow(long timeMillis) { 
  if (timeMillis < 0) { 
    return null; 
  } 
 
  int idx = calculateTimeIdx(timeMillis); 
  // Calculate current bucket start time. 
  // 計算窗口的開始時間，計算每個格子的開始時間 
  long windowStart = calculateWindowStart(timeMillis); 
 
  /* 
         * Get bucket item at given time from the array. 
         * 
         * (1) Bucket is absent, then just create a new bucket and CAS update to circular array. 
         * (2) Bucket is up-to-date, then just return the bucket. 
         * (3) Bucket is deprecated, then reset current bucket and clean all deprecated buckets. 
         */ 
  while (true) { 
    WindowWrap<T> old = array.get(idx); 
    // 如果沒有窗格，創建窗格 
    if (old == null) { 
      /* 
                 *     B0       B1      B2    NULL      B4 
                 * ||_______|_______|_______|_______|_______||___ 
                 * 200     400     600     800     1000    1200  timestamp 
                 *                             ^ 
                 *                          time=888 
                 *            bucket is empty, so create new and update 
                 * 
                 * If the old bucket is absent, then we create a new bucket at {@code windowStart}, 
                 * then try to update circular array via a CAS operation. Only one thread can 
                 * succeed to update, while other threads yield its time slice. 
                 */ 
      WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis)); 
      if (array.compareAndSet(idx, null, window)) { 
        // Successfully updated, return the created bucket. 
        return window; 
      } else { 
        // Contention failed, the thread will yield its time slice to wait for bucket available. 
        Thread.yield(); 
      } 
      // 當前窗格存在，返回歷史窗格 
    } else if (windowStart == old.windowStart()) { 
      /* 
                 *     B0       B1      B2     B3      B4 
                 * ||_______|_______|_______|_______|_______||___ 
                 * 200     400     600     800     1000    1200  timestamp 
                 *                             ^ 
                 *                          time=888 
                 *            startTime of Bucket 3: 800, so it's up-to-date 
                 * 
                 * If current {@code windowStart} is equal to the start timestamp of old bucket, 
                 * that means the time is within the bucket, so directly return the bucket. 
                 */ 
      return old; 
      // 
    } else if (windowStart > old.windowStart()) { 
      /* 
                 *   (old) 
                 *             B0       B1      B2    NULL      B4 
                 * |_______||_______|_______|_______|_______|_______||___ 
                 * ...    1200     1400    1600    1800    2000    2200  timestamp 
                 *                              ^ 
                 *                           time=1676 
                 *          startTime of Bucket 2: 400, deprecated, should be reset 
                 * 
                 * If the start timestamp of old bucket is behind provided time, that means 
                 * the bucket is deprecated. We have to reset the bucket to current {@code windowStart}. 
                 * Note that the reset and clean-up operations are hard to be atomic, 
                 * so we need a update lock to guarantee the correctness of bucket update. 
                 * 
                 * The update lock is conditional (tiny scope) and will take effect only when 
                 * bucket is deprecated, so in most cases it won't lead to performance loss. 
                 */ 
      if (updateLock.tryLock()) { 
        try { 
          // Successfully get the update lock, now we reset the bucket. 
          // 清空所有的窗格數據 
          return resetWindowTo(old, windowStart); 
        } finally { 
          updateLock.unlock(); 
        } 
      } else { 
        // Contention failed, the thread will yield its time slice to wait for bucket available. 
        Thread.yield(); 
      } 
      // 如果時鐘回撥，重新創建時間格 
    } else if (windowStart < old.windowStart()) { 
      // Should not go through here, as the provided time is already behind. 
      return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis)); 
    } 
  } 
}

漏桶算法

漏桶算法(Leaky Bucket)是網絡世界中流量整形(Traffic Shaping)或速率限制(Rate Limiting)時經常使用的一種算法，它的主要目的是控制數據注入到網絡的速率，平滑網絡上的突發流量。漏桶算法提供了一種機制，通過它，突發流量可以被整形以便為網絡提供一個穩定的流量, 執行過程如下圖所示。

實現代碼案例：

public class LeakyBucket { 
  public long timeStamp = System.currentTimeMillis();  // 當前時間 
  public long capacity; // 桶的容量 
  public long rate; // 水漏出的速度 
  public long water; // 當前水量(當前累積請求數) 
 
  public boolean grant() { 
    long now = System.currentTimeMillis(); 
    // 先執行漏水，計算剩余水量 
    water = Math.max(0, water - (now - timeStamp) * rate);  
 
    timeStamp = now; 
    if ((water + 1) < capacity) { 
      // 嘗試加水,并且水還未滿 
      water += 1; 
      return true; 
    } else { 
      // 水滿，拒絕加水 
      return false; 
    } 
  } 
}

說明：

(1)未滿加水：通過代碼 water +=1進行不停加水的動作。

(2)漏水：通過時間差來計算漏水量。

(3)剩余水量：總水量-漏水量。

在 Sentine 中RateLimiterController 實現了了漏桶算法 , 核心代碼如下

@Override 
public boolean canPass(Node node, int acquireCount, boolean prioritized) { 
  // Pass when acquire count is less or equal than 0. 
  if (acquireCount <= 0) { 
    return true; 
  } 
  // Reject when count is less or equal than 0. 
  // Otherwise,the costTime will be max of long and waitTime will overflow in some cases. 
  if (count <= 0) { 
    return false; 
  } 
 
  long currentTime = TimeUtil.currentTimeMillis(); 
  // Calculate the interval between every two requests. 
  // 計算時間間隔 
  long costTime = Math.round(1.0 * (acquireCount) / count * 1000); 
 
  // Expected pass time of this request. 
  // 期望的執行時間 
  long expectedTime = costTime + latestPassedTime.get(); 
 
  // 當前時間 > 期望時間 
  if (expectedTime <= currentTime) { 
    // Contention may exist here, but it's okay. 
    // 可以通過，并且設置最后通過時間 
    latestPassedTime.set(currentTime); 
    return true; 
  } else { 
    // Calculate the time to wait. 
    // 等待時間 = 期望時間 - 最后時間 - 當前時間 
    long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis(); 
    // 等待時間 > 最大排隊時間 
    if (waitTime > maxQueueingTimeMs) { 
      return false; 
    } else { 
      // 上次時間 + 間隔時間 
      long oldTime = latestPassedTime.addAndGet(costTime); 
      try { 
        // 等待時間 
        waitTime = oldTime - TimeUtil.currentTimeMillis(); 
        // 等待時間 > 最大排隊時間 
        if (waitTime > maxQueueingTimeMs) { 
          latestPassedTime.addAndGet(-costTime); 
          return false; 
        } 
        // in race condition waitTime may <= 0 
        // 休眠等待 
        if (waitTime > 0) { 
          Thread.sleep(waitTime); 
        } 
        // 等待完了，就放行 
        return true; 
      } catch (InterruptedException e) { 
      } 
    } 
  } 
  return false; 
}

令牌桶算法

令牌桶算法是網絡流量整形(Traffic Shaping)和速率限制(Rate Limiting)中最常使用的一種算法。典型情況下，令牌桶算法用來控制發送到網絡上的數據的數目，并允許突發數據的發送。如下圖所示：

簡單的說就是，一邊請求時會消耗桶內的令牌，另一邊會以固定速率往桶內放令牌。當消耗的請求大于放入的速率時，進行相應的措施，比如等待，或者拒絕等。

實現代碼案例：

public class TokenBucket { 
  public long timeStamp = System.currentTimeMillis();  // 當前時間 
  public long capacity; // 桶的容量 
  public long rate; // 令牌放入速度 
  public long tokens; // 當前令牌數量 
 
  public boolean grant() { 
    long now = System.currentTimeMillis(); 
    // 先添加令牌 
    tokens = Math.min(capacity, tokens + (now - timeStamp) * rate); 
    timeStamp = now; 
    if (tokens < 1) { 
      // 若不到1個令牌,則拒絕 
      return false; 
    } else { 
      // 還有令牌，領取令牌 
      tokens -= 1; 
      return true; 
    } 
  } 
}

Sentinel 在 WarmUpController 中運用到了令牌桶算法，在這里可以實現對系統的預熱，設定預熱時間和水位線，對于預熱期間多余的請求直接拒絕掉。

public boolean canPass(Node node, int acquireCount, boolean prioritized) { 
  long passQps = (long) node.passQps(); 
 
  long previousQps = (long) node.previousPassQps(); 
  syncToken(previousQps); 
 
  // 開始計算它的斜率 
  // 如果進入了警戒線，開始調整他的qps 
  long restToken = storedTokens.get(); 
  if (restToken >= warningToken) { 
    long aboveToken = restToken - warningToken; 
    // 消耗的速度要比warning快，但是要比慢 
    // current interval = restToken*slope+1/count 
    double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count)); 
    if (passQps + acquireCount <= warningQps) { 
      return true; 
    } 
  } else { 
    if (passQps + acquireCount <= count) { 
      return true; 
    } 
  } 
 
  return false; 
}

限流算法總結

計數器 VS 時間窗

時間窗算法的本質也是通過計數器算法實現的。

時間窗算法格子劃分的越多，那么滑動窗口的滾動就越平滑，限流的統計就會越精確，但是也會占用更多的內存存儲。

漏桶 VS 令牌桶

漏桶算法和令牌桶算法本質上是為了做流量整形或速率限制，避免系統因為大流量而被打崩，但是兩者的核心差異在于限流的方向是相反的

漏桶：限制的是流量的流出速率，是相對固定的。

令牌桶：限制的是流量的平均流入速率，并且允許一定程度的突然性流量，最大速率為桶的容量和生成token的速率。

在某些場景中，漏桶算法并不能有效的使用網絡資源，因為漏桶的漏出速率是相對固定的，所以在網絡情況比較好并且沒有擁塞的狀態下，漏桶依然是會有限制的，并不能放開量，因此并不能有效的利用網絡資源。而令牌桶算法則不同，其在限制平均速率的同時，支持一定程度的突發流量。

參考文檔

https://www.cnblogs.com/linjiqin/p/9707713.html

https://www.cnblogs.com/dijia478/p/13807826.html

責任編輯：姜華來源：運維開發故事