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

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

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對請求訪問進(jìn)行限流，基于nginx擴(kuò)展模塊，

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

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

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

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

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

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

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

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

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

為什么要限流

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

計(jì)數(shù)器 Counter

基于時(shí)間滑動窗口Timed Sliding Window

漏桶Leaky bucket

令牌桶Token bucket

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

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

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

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

限制指定時(shí)間內(nèi)的請求數(shù)量，比如1秒內(nèi)最大的請求量為2個(gè)

demo如下：

public class PerTimeUnitCounterFlowControl {
    private static final long INTERVAL = 5 * 1000;//時(shí)間間隔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;
    }
}

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

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

如下：

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

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

demo如下：

public class TimedSlidingWindowFlowControl {
    private static final int LIMIT = 5;
    private long duration;//每個(gè)格子的時(shí)長
    private int bucketSize;//總格子數(shù)
    private final long windowTime;
    private final ScheduledExecutorService scheduledExecutor;
    private long startedTimestamp;
    private volatile int head;//指向第一個(gè)格子
    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;//指向第一個(gè)格子
        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("當(dāng)前count:" + count);
        if(count <= LIMIT){
            return true;
        }
        return false;
    }

    /**
     * 查找當(dāng)前的位置
     * @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;
            }
        }
    }

    /**
     * 獲取當(dāng)前計(jì)數(shù)
     * @return  int
     */
    private int getCurrentCount(){
        return Arrays.stream(buckets).mapToInt(buckets -> buckets.get()).sum();
    }

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

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

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

漏桶算法，即Leaky bucket，是一種很常用的限流算法，可以用來實(shí)現(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.

翻譯過來的意思為：

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

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

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

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

這個(gè)可以使用基于生產(chǎn)者-消費(fèi)者共享阻塞隊(duì)列實(shí)現(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("當(dāng)前漏桶的容量：" + 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();
                }
            }
        }
    }
}

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

關(guān)于令牌token的使用場景比較多了，比如Auth的access token，計(jì)算機(jī)網(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個(gè)。

桶中最多存放b個(gè)令牌，當(dāng)桶滿時(shí)，新添加的令牌被丟棄或拒絕。

當(dāng)一個(gè)n個(gè)字節(jié)大小的數(shù)據(jù)包到達(dá)，將從桶中刪除n個(gè)令牌，接著數(shù)據(jù)包被發(fā)送到網(wǎng)絡(luò)上。

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

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

public class RateLimiterTester {

    public static void main(String[] args) {
        RateLimiter limiter = RateLimiter.create(2);//發(fā)令牌的間隔時(shí)間約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ū)別和實(shí)現(xiàn)可以自行查看下文檔或網(wǎng)上找下相關(guān)分析。

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