摘要：限流，是對流量控制?；跁r間的滑動窗口，參照于滑動窗口，將單位時間看做是一個窗口，將窗口中的每個格子設(shè)定為指定時間間隔，為格子總數(shù)，那么單位時間就是。很明顯格子劃分的越多，滑動窗口的滑動就越平滑，限流統(tǒng)計就越精確。

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

1）在Tcp協(xié)議中，F(xiàn)low Control，

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

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

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

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

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

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

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

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

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

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

Rate Limiting，在計算機網(wǎng)絡(luò)中，網(wǎng)絡(luò)接口控制器用限流來控制發(fā)送和接收的流量速率，以防止Dos攻擊。

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

為什么要限流

限流，為了保護系統(tǒng)在應(yīng)對流量高峰時，系統(tǒng)能夠依然以可控的處理速率對外提供服務(wù)，而不至于奔潰或變?yōu)椴豢煞?wù)的。
這也從側(cè)面體現(xiàn)了系統(tǒng)服務(wù)的穩(wěn)定性，如果是SOA服務(wù)的話，也體現(xiàn)了服務(wù)設(shè)計原則的自治性。

計數(shù)器 Counter

基于時間滑動窗口Timed Sliding Window

漏桶Leaky bucket

令牌桶Token bucket

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

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

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

基于單位時間的計數(shù)器

限制指定時間內(nèi)的請求數(shù)量，比如1秒內(nèi)最大的請求量為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;
    }
}

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

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

如下：

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

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

demo如下：

public class TimedSlidingWindowFlowControl {
    private static final int LIMIT = 5;
    private long duration;//每個格子的時長
    private int bucketSize;//總格子數(shù)
    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;
            }
        }
    }

    /**
     * 獲取當前計數(shù)
     * @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
    }

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

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

漏桶算法，即Leaky bucket，是一種很常用的限流算法，可以用來實現(xiàn)流量整形（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.

翻譯過來的意思為：

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

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

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

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

這個可以使用基于生產(chǎn)者-消費者共享阻塞隊列實現(xiàn)。

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());
    }

    /**
     * 內(nèi)部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();
                }
            }
        }
    }
}

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

關(guān)于令牌token的使用場景比較多了，比如Auth的access token，計算機網(wǎng)絡(luò)中輪轉(zhuǎn)訪問MAC協(xié)議中的Token passing等。

現(xiàn)在是關(guān)于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個字節(jié)大小的數(shù)據(jù)包到達，將從桶中刪除n個令牌，接著數(shù)據(jù)包被發(fā)送到網(wǎng)絡(luò)上。

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

算法實現(xiàn)直接使用Guava的RateLimiter

public class RateLimiterTester {

    public static void main(String[] args) {
        RateLimiter limiter = RateLimiter.create(2);//發(fā)令牌的間隔時間約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

令牌桶算法可以應(yīng)對突發(fā)流量，RateLimiter提供了SmoothBursty和SmoothWarmingUp兩種需求。具體區(qū)別和實現(xiàn)可以自行查看下文檔或網(wǎng)上找下相關(guān)分析。

推薦關(guān)于Hystrix一篇好文。