流控算法

漏桶算法

令牌桶算法

一年一度的「雙 11」又要到了，阿里的碼農們進入了一年中最辛苦的時光。各種容量評估、壓測、擴容讓我們忙得不可開交。洛陽親友如相問，就說我搞雙十一。

如何讓系統在洶涌澎湃的流量面前談笑風生？我們的策略是不要讓系統超負荷工作。如果現有的系統扛不住業務目標怎么辦？加機器！機器不夠怎么辦？業務降級，系統限流！

正所謂「他強任他強，清風拂山崗；他橫任他橫，明月照大江」，降級和限流是大促保障中必不可少的神兵利器，丟卒保車，以暫停邊緣業務為代價保障核心業務的資源，以系統不被突發流量壓掛為第一要務。

集團的中間件有一個不錯的單機限流框架，支持兩種限流模式：控制速率和控制并發。限流這種東西，應該是來源于網絡里面的「流量整型」，通過控制數據包的傳輸速率和時機，來實現一些性能、服務質量方面的東西。令牌桶是一種常見的流控算法，屬于控制速率類型的。控制并發則相對要常見的多，比如操作系統里的「信號量」就是一種控制并發的方式。

在 Wikipedia 上，令牌桶算法是這么描述的：

1. 每秒會有 r 個令牌放入桶中，或者說，每過 1/r 秒桶中增加一個令牌

2. 桶中最多存放 b 個令牌，如果桶滿了，新放入的令牌會被丟棄

3. 當一個 n 字節的數據包到達時，消耗 n 個令牌，然后發送該數據包

4. 如果桶中可用令牌小于 n，則該數據包將被緩存或丟棄

令牌桶控制的是一個時間窗口內的通過的數據量，在 API 層面我們常說的 QPS、TPS，正好是一個時間窗口內的請求量或者事務量，只不過時間窗口限定在 1s 罷了。

現實世界的網絡工程中使用的令牌桶，比概念圖中的自然是復雜了許多，「令牌桶」的數量也不是一個而是兩個，簡單的算法描述可用參考中興的期刊[1]或者 RFC。

假如項目使用 Java 語言，我們可以輕松地借助 Guava 的 RateLimiter 來實現基于令牌桶的流控。RateLimiter 令牌桶算法的單桶實現，也許是因為在 Web 應用層面單桶實現就夠用了，雙筒實現就屬于過度設計。

RateLimiter 對簡單的令牌桶算法做了一些工程上的優化，具體的實現是 SmoothBursty。需要注意的是，RateLimiter 的另一個實現 SmoothWarmingUp，就不是令牌桶了，而是漏桶算法。也許是出于簡單起見，RateLimiter 中的時間窗口能且僅能為 1s，如果想搞其他時間單位的限流，只能另外造輪子。

SmoothBursty 積極響應×××總理的號召，上個月的流量沒用完，可以挪到下個月用。其實就是 SmoothBursty 有一個可以放 N 個時間窗口產生的令牌的桶，系統空閑的時候令牌就一直攢著，最好情況下可以扛 N 倍于限流值的高峰而不影響后續請求。如果不想像三峽大壩一樣能扛千年一遇的洪水，可以把 N 設置為 1，這樣就只屯一個時間窗口的令牌。

RateLimiter 有一個有趣的特性是「前人挖坑后人跳」，也就是說 RateLimiter 允許某次請求拿走超出剩余令牌數的令牌，但是下一次請求將為此付出代價，一直等到令牌虧空補上，并且桶中有足夠本次請求使用的令牌為止[2]。這里面就涉及到一個權衡，是讓前一次請求干等到令牌夠用才走掉呢，還是讓它先走掉后面的請求等一等呢？Guava 的設計者選擇的是后者，先把眼前的活干了，后面的事后面再說。

當我們要實現一個基于速率的單機流控框架的時候，RateLimiter 是一個完善的核心組件，就仿佛 Linux 內核對 GNU 操作系統那樣重要。但是我們還需要其他的一些東西才能把一個流控框架跑起來，比如一個通用的 API，一個攔截器，一個在線配置流控閾值的后臺等等。

下面隨便寫了一個簡單的流控框架 API，至于攔截器和后臺就懶得寫了，有時間再自己造一套中間件的輪子吧~

public class TrafficShaper {    public static class RateLimitException extends Exception {        private static final long serialVersionUID = 1L;        private String resource;        public String getResource() {            return resource;
        }        public RateLimitException(String resource) {            super(resource + " should not be visited so frequently");            this.resource = resource;
        }        @Override        public synchronized Throwable fillInStackTrace() {            return this;
        }
    }    private static final ConcurrentMap<String, RateLimiter>
            resourceLimiterMap = Maps.newConcurrentMap();    public static void updateResourceQps(String resource, double qps) {
        RateLimiter limiter = resourceLimiterMap.get(resource);        if (limiter == null) {
            limiter = RateLimiter.create(qps);
            RateLimiter putByOtherThread
                    = resourceLimiterMap.putIfAbsent(resource, limiter);            if (putByOtherThread != null) {
                limiter = putByOtherThread;
            }
        }
        limiter.setRate(qps);
    }    public static void removeResource(String resource) {
        resourceLimiterMap.remove(resource);
    }    public static void enter(String resource) throws RateLimitException {
        RateLimiter limiter = resourceLimiterMap.get(resource);        if (limiter == null) {            return;
        }        if (!limiter.tryAcquire()) {            throw new RateLimitException(resource);
        }
    }    public static void exit(String resource) {        //do nothing when use RateLimiter
    }
}