摘要：限流，是對流量控制。基于時間的滑動窗口，參照于滑動窗口，將單位時間看做是一個窗口，將窗口中的每個格子設定為指定時間間隔，為格子總數，那么單位時間就是。很明顯格子劃分的越多，滑動窗口的滑動就越平滑，限流統計就越精確。

限流，在一些我們已知的場景有：

1）在Tcp協議中，Flow Control，

流量控制以動態調整發送空間大小(滑動窗口)的形式來反映接收端接收消息的能力，反饋給發送端以調整發送速度，避免發送速度過快導致的丟包或者過慢降低整體性能。

在Tcp協議中，通過在首部設置window size的值來控制窗口大小。

2) 在Web sever中，使用nginx對請求訪問進行限流，基于nginx擴展模塊，

ngx_http_limit_conn_module，可以針對Ip或Domain來限制連接數。

ngx_http_limit_req_module，可以通過自定義鍵值來限制請求處理的頻率。

3) 在現在一些主流的開放平臺，都有針對API調用的限制，比如淘寶開放平臺針對商品查詢接口的限制次數，百度地圖針對開發者““地點檢索”API是有指定限額的。這些都是針對API的限流。

4) 秒殺系統中，由于庫存量一般是很少的，對應只有少部分的用戶才能秒殺成功，因此我們要限制絕大部分用戶流量。

5） 各種框架使用時的數量限制，如數據連接池最大連接數、線程池最大線程數、zk最大連接等等

以上是限流的常見場景。限流，是對流量控制。

Flow Control，是在數據通信中，流量控制是管理兩個節點之間的數據傳輸速率的過程，以防止快速發送者壓倒慢速接收者。
它為接收機提供了一種控制傳輸速度的機制，使接收節點不會被來自發送節點的數據淹沒。

Rate Limiting，在計算機網絡中，網絡接口控制器用限流來控制發送和接收的流量速率，以防止Dos攻擊。

可以看出，限流主要是控制發送和接受雙方的流量速率，保證工作正常進行。

為什么要限流

限流，為了保護系統在應對流量高峰時，系統能夠依然以可控的處理速率對外提供服務，而不至于奔潰或變為不可服務的。
這也從側面體現了系統服務的穩定性，如果是SOA服務的話，也體現了服務設計原則的自治性。

計數器 Counter

基于時間滑動窗口Timed Sliding Window

漏桶Leaky bucket

令牌桶Token bucket

注：以下算法只做算法演示，肯定有很多細節未考慮，包括在多線程下行為是否正確等

計數器是最簡單的一種，針對資源設置訪問最大總量（上限）Max，以及定義一個計數器Counter，每當需要對資源訪問時，Counter++，當Counter小于Max,訪問可以通過，否則不可用。

一般這個場景在項目中比較常見，比如我們使用Semphore的acquire、release來控制多線程對資源的許可，比如Jedis Pool的對象池borrow、return。

基于單位時間的計數器

限制指定時間內的請求數量，比如1秒內最大的請求量為2個

demo如下：

public class PerTimeUnitCounterFlowControl {
    private static final long INTERVAL = 5 * 1000;//時間間隔ms
    private long timestamp;
    private int counter;
    private int limit;
    private long interval;

    public PerTimeUnitCounterFlowControl(long interval,int limit) {
        this.interval = interval <= 0? INTERVAL:interval;
        this.timestamp = SystemClock.now();
        this.limit = limit;
    }

    /**
     *
     * @return
     */
    public boolean acquire(){
        long now = SystemClock.now();
        if (now < timestamp + interval){
            counter++;
            return counter <= limit;
        }
        timestamp = now;
        counter = 1;
        return true;
    }
}

該算法的缺陷是，在時間節點重置的時隙里可能被突發請求超限。

Timed Sliding Window，參照于Tcp滑動窗口，將單位時間T看做是一個窗口，將窗口中的每個格子設定為指定時間間隔Duration，Window Size為格子總數 buckets，那么單位時間就是buckets * Duration。每個格子有自己獨立的計數器。當時間每過去Duration時候，窗口就會向右滑動一個格子。

如下：

每當有請求過來時，都會落在指定格子里，然后獲取當前窗口的所有計數器之和，以此來觸發是否限流。

很明顯格子劃分的越多，滑動窗口的滑動就越平滑，限流統計就越精確。

demo如下：

public class TimedSlidingWindowFlowControl {
    private static final int LIMIT = 5;
    private long duration;//每個格子的時長
    private int bucketSize;//總格子數
    private final long windowTime;
    private final ScheduledExecutorService scheduledExecutor;
    private long startedTimestamp;
    private volatile int head;//指向第一個格子
    private AtomicInteger[] buckets;

    public TimedSlidingWindowFlowControl(long duration, int bucketSize) {
        this.duration = duration;
        this.bucketSize = bucketSize;
        this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
        this.windowTime = duration * bucketSize;
        buckets = new AtomicInteger[bucketSize];
    }

    /**
     * 初始化
     */
    protected void init(){
        startedTimestamp = SystemClock.now();
        for(int i = 0; i < bucketSize;i++){
            buckets[i] = new AtomicInteger(0);
        }
        head = 0;//指向第一個格子
        scheduledExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                timeRolling();
            }
        },duration/2,duration, TimeUnit.MILLISECONDS);
    }

    /**
     * 獲取許可
     * @return
     */
    private boolean acquire(){
        long now = SystemClock.now();
        long timestampDiff = now - startedTimestamp;
        long mask = timestampDiff % (windowTime);
        //相對于head的位置
        int idx = getBucketIndex(mask);
        if(idx == -1){
            throw new IllegalStateException("illegalState");
        }
        buckets[idx].incrementAndGet();
        int count = getCurrentCount();
        System.out.println("當前count:" + count);
        if(count <= LIMIT){
            return true;
        }
        return false;
    }

    /**
     * 查找當前的位置
     * @param mask
     * @return
     */
    private int getBucketIndex(long mask){
        int cursor = head;
        int stopIndex = cursor;
        if(mask <= duration){
            return cursor;
        }
        long d = duration;
        while (true){
            cursor = (cursor + 1) % bucketSize;
            if(cursor == stopIndex){
                return -1;
            }
            d = d + duration;
            if(mask <= d){
                return cursor;
            }
        }
    }

    /**
     * 獲取當前計數
     * @return  int
     */
    private int getCurrentCount(){
        return Arrays.stream(buckets).mapToInt(buckets -> buckets.get()).sum();
    }

    /**
     * 時間滾動
     */
    private void timeRolling(){
        //每次格子移動都會更改head
        int last = head;
        head = (head + 1) % bucketSize;
        System.out.println("時間向前移動一格：" + head);
        buckets[last].set(0);//reset
    }

    /**
     * 關閉
     */
    protected void shutdown() throws InterruptedException {
        scheduledExecutor.shutdown();
        scheduledExecutor.awaitTermination(5,TimeUnit.SECONDS);
    }
}

上面的demo可能有些細節未考慮，基本思路是使用定時任務模擬時鐘滾動，循環復用計數器桶，使用head指針始終指向第一個桶，統計所有的桶計數的和 判斷是否觸發限流。
實際上考慮“使用定時任務模擬時鐘滾動”，這種方式有一些缺點：會浪費Cpu資源，而且還依賴時鐘。可以考慮采用類似Guava的RateLimiter的延遲計算機制。
另更多關于滑動窗口計數可以參考Storm的RollingCountBolt和Hystrix的Metrics實現。

漏桶算法，即Leaky bucket，是一種很常用的限流算法，可以用來實現流量整形（Traffic Shaping）和流量控制（Traffic Policing）。

以下是wikipedia對Leaky bucket的算法描述：

[A] fixed capacity bucket, associated with each virtual connection or user, leaks at a fixed rate.
If the bucket is empty, it stops leaking.
For a packet to conform, it has to be possible to add a specific amount of water to the bucket: The specific amount added by a conforming packet can be the same for all packets, or can be proportional to the length of the packet.
If this amount of water would cause the bucket to exceed its capacity then the packet does not conform and the water in the bucket is left unchanged.

翻譯過來的意思為：

一個固定容量的漏桶，按照常量固定速率流出水滴；

如果桶是空的，則不需流出水滴；

可以以任意速率流入水滴到漏桶；

如果流入水滴超出了桶的容量，則流入的水滴溢出了（被丟棄），而漏桶容量是不變的。

這個可以使用基于生產者-消費者共享阻塞隊列實現。

demo如下：

public class LeakyBucketFlowControl{
    private int capacity;
    private LinkedBlockingQueue bucket;
    private int flowOutNum;//以恒定的速率流出
    private int flowOutTimeUnit;//
    private static final int VALUE = 1;
    private Thread thread;
    private volatile boolean stop = false;

    public LeakyBucketFlowControl(int capacity, int flowOutNum, int flowOutTimeUnit) {
        this.capacity = capacity;
        this.flowOutNum = flowOutNum;
        this.flowOutTimeUnit = flowOutTimeUnit;
        this.bucket = new LinkedBlockingQueue<>(capacity);
        this.thread = new Thread(new Worker());
    }

    /**
     * init
     */
    public void init(){
        thread.start();
    }

    /**
     * 獲取許可
     * @return
     */
    protected boolean acquire(){
        boolean of = bucket.offer(VALUE);
        return of;
    }

    /**
     * shutdown
     */
    public void shutdown(){
        stop = true;
        System.out.println("當前漏桶的容量：" + bucket.size());
    }

    /**
     * 內部worker
     */
    class Worker implements Runnable{

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted() && !stop){
                try {
                    TimeUnit.MILLISECONDS.sleep(flowOutTimeUnit);
                    for(int i = 1;i <= flowOutNum;i++){
                        bucket.take();
                    }
                    System.out.println("漏桶流出容量為：" + bucket.size());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

一般情況下，漏桶對于流量整形有用， 無論什么樣的請求速率進來，漏桶總是以恒定的速率執行，但對于突發傳輸有一定限制，除非當前漏桶已經為空。

關于令牌token的使用場景比較多了，比如Auth的access token，計算機網絡中輪轉訪問MAC協議中的Token passing等。

現在是關于token在限流中的算法描述：

[A] token is added to the bucket every  1/r seconds.
The bucket can hold at the most  b tokens. If a token arrives when the bucket is full, it is discarded.
When a packet (network layer PDU) of n bytes arrives, n tokens are removed from the bucket, and the packet is sent to the network.
If fewer than n tokens are available, no tokens are removed from the bucket, and the packet is considered to be non-conformant.

令牌將按照固定的速率被放入令牌桶中。比如每秒放10個。

桶中最多存放b個令牌，當桶滿時，新添加的令牌被丟棄或拒絕。

當一個n個字節大小的數據包到達，將從桶中刪除n個令牌，接著數據包被發送到網絡上。

如果桶中的令牌不足n個，則不會刪除令牌，且該數據包將被限流（要么丟棄，要么緩沖區等待）。

算法實現直接使用Guava的RateLimiter

public class RateLimiterTester {

    public static void main(String[] args) {
        RateLimiter limiter = RateLimiter.create(2);//發令牌的間隔時間約500ms
        double x = limiter.acquire(5) * 1000;
        System.out.println(x + "....");
        for (int i = 1;i <= 5;i++){
            double y = limiter.acquire()  * 1000;
            System.out.println(y);
        }
    }
}

輸出
0.0....
2497.7299999999996
491.842
495.838
497.392
498.442

令牌桶算法可以應對突發流量，RateLimiter提供了SmoothBursty和SmoothWarmingUp兩種需求。具體區別和實現可以自行查看下文檔或網上找下相關分析。

推薦關于Hystrix一篇好文。