令牌桶算法限流

時間 2019-11-09

標籤算法限流欄目應用數學简体版

原文原文鏈接

限流

限流是對某一時間窗口內的請求數進行限制，保持系統的可用性和穩定性，防止因流量暴增而致使的系統運行緩慢或宕機。經常使用的限流算法有令牌桶和和漏桶，而Google開源項目Guava中的RateLimiter使用的就是令牌桶控制算法。html

在開發高併發系統時有三把利器用來保護系統：緩存、降級和限流java

緩存：緩存的目的是提高系統訪問速度和增大系統處理容量
降級：降級是當服務器壓力劇增的狀況下，根據當前業務狀況及流量對一些服務和頁面有策略的降級，以此釋放服務器資源以保證核心任務的正常運行
限流：限流的目的是經過對併發訪問/請求進行限速，或者對一個時間窗口內的請求進行限速來保護系統，一旦達到限制速率則能夠拒絕服務、排隊或等待、降級等處理

咱們常常在調別人的接口的時候會發現有限制，好比微信公衆平臺接口、百度API Store、聚合API等等這樣的，對方會限制天天最多調多少次或者每分鐘最多調多少次git

咱們本身在開發系統的時候也須要考慮到這些，好比咱們公司在上傳商品的時候就作了限流，由於用戶每一次上傳商品，咱們須要將商品數據同到到美團、餓了麼、京東、百度、自營等第三方平臺，這個工做量是巨大，頻繁操做會拖慢系統，故作限流。github

以上都是題外話，接下來咱們重點看一下令牌桶算法web

令牌桶算法

下面是從網上找的兩張圖來描述令牌桶算法：算法

RateLimiter

https://github.com/google/guavaspring

RateLimiter的代碼不長，註釋加代碼432行，看一下RateLimiter怎麼用segmentfault

 1 package com.cjs.example;
 2 
 3 import com.google.common.util.concurrent.RateLimiter;
 4 import org.springframework.web.bind.annotation.RequestMapping;
 5 import org.springframework.web.bind.annotation.RestController;
 6 
 7 import java.text.SimpleDateFormat;
 8 import java.util.Date;
 9 
10 @RestController
11 public class HelloController {
12 
13     private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
14 
15     private static final RateLimiter rateLimiter = RateLimiter.create(2);
16 
17     /**
18      * tryAcquire嘗試獲取permit，默認超時時間是0，意思是拿不到就當即返回false
19      */
20     @RequestMapping("/sayHello")
21     public String sayHello() {
22         if (rateLimiter.tryAcquire()) { //  一次拿1個
23             System.out.println(sdf.format(new Date()));
24             try {
25                 Thread.sleep(500);
26             } catch (InterruptedException e) {
27                 e.printStackTrace();
28             }
29         }else {
30             System.out.println("limit");
31         }
32         return "hello";
33     }
34 
35     /**
36      * acquire拿不到就等待，拿到爲止
37      */
38     @RequestMapping("/sayHi")
39     public String sayHi() {
40         rateLimiter.acquire(5); //  一次拿5個
41         System.out.println(sdf.format(new Date()));
42         return "hi";
43     }
44 
45 }

關於RateLimiter：

A rate limiter。每一個acquire()方法若是必要的話會阻塞直到一個permit可用，而後消費它。得到permit之後不須要釋放。
RateLimiter在併發環境下使用是安全的：它將限制全部線程調用的總速率。注意，它不保證公平調用。
RateLimiter在併發環境下使用是安全的：它將限制全部線程調用的總速率。注意，它不保證公平調用。Rate limiter（直譯爲：速度限制器）常常被用來限制一些物理或者邏輯資源的訪問速率。這和java.util.concurrent.Semaphore正好造成對照。
一個RateLimiter主要定義了發放permits的速率。若是沒有額外的配置，permits將以固定的速度分配，單位是每秒多少permits。默認狀況下，Permits將會被穩定的平緩的發放。
能夠配置一個RateLimiter有一個預熱期，在此期間permits的發放速度每秒穩步增加直到到達穩定的速率

基本用法：

final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
void submitTasks(List<Runnable> tasks, Executor executor) {
    for (Runnable task : tasks) {
        rateLimiter.acquire(); // may wait
        executor.execute(task);
    }
}

實現

SmoothBursty以穩定的速度生成permit緩存

SmoothWarmingUp是漸進式的生成，最終達到最大值趨於穩定安全

源碼片斷解讀：

public abstract class RateLimiter {

    /**
     * 用給定的吞吐量（「permits per second」）建立一個RateLimiter。
     * 一般是QPS
     */
    public static RateLimiter create(double permitsPerSecond) {
        return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
    }
    
    static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
        RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
        rateLimiter.setRate(permitsPerSecond);
        return rateLimiter;
    }
    
    /**
     * 用給定的吞吐量(QPS)和一個預熱期建立一個RateLimiter
     */
    public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
        checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
        return create(permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
    }

    static RateLimiter create(
            double permitsPerSecond,
            long warmupPeriod,
            TimeUnit unit,
            double coldFactor,
            SleepingStopwatch stopwatch) {
        RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
        rateLimiter.setRate(permitsPerSecond);
        return rateLimiter;
    }

    private final SleepingStopwatch stopwatch;

    //    鎖
    private volatile Object mutexDoNotUseDirectly;

    private Object mutex() {
        Object mutex = mutexDoNotUseDirectly;
        if (mutex == null) {
            synchronized (this) {
                mutex = mutexDoNotUseDirectly;
                if (mutex == null) {
                    mutexDoNotUseDirectly = mutex = new Object();
                }
            }
        }
        return mutex;
    }
    
    /**
     * 從RateLimiter中獲取一個permit，阻塞直到請求能夠得到爲止
     * @return 休眠的時間，單位是秒，若是沒有被限制則是0.0
     */
    public double acquire() {
        return acquire(1);
    }
  
    /**
     * 從RateLimiter中獲取指定數量的permits，阻塞直到請求能夠得到爲止
     */
    public double acquire(int permits) {
        long microsToWait = reserve(permits);
        stopwatch.sleepMicrosUninterruptibly(microsToWait);
        return 1.0 * microsToWait / SECONDS.toMicros(1L);
    }
    
    /**
     * 預約給定數量的permits以備未來使用
     * 直到這些預約數量的permits能夠被消費則返回逝去的微秒數
     */
    final long reserve(int permits) {
        checkPermits(permits);
        synchronized (mutex()) {
            return reserveAndGetWaitLength(permits, stopwatch.readMicros());
        }
    }
    
    private static void checkPermits(int permits) {
        checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
    }
    
    final long reserveAndGetWaitLength(int permits, long nowMicros) {
        long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
        return max(momentAvailable - nowMicros, 0);
    }
}


abstract class SmoothRateLimiter extends RateLimiter {
    
    /** The currently stored permits. */
    double storedPermits;

    /** The maximum number of stored permits. */
    double maxPermits;

    /**
     * The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits
     * per second has a stable interval of 200ms.
     */
    double stableIntervalMicros;
    
    /**
     * The time when the next request (no matter its size) will be granted. After granting a request,
     * this is pushed further in the future. Large requests push this further than small requests.
     */
    private long nextFreeTicketMicros = 0L; // could be either in the past or future
    
    final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
        resync(nowMicros);
        long returnValue = nextFreeTicketMicros;
        double storedPermitsToSpend = min(requiredPermits, this.storedPermits);    //    本次能夠獲取到的permit數量
        double freshPermits = requiredPermits - storedPermitsToSpend;    //    差值，若是存儲的permit大於本次須要的permit數量則此處是0，不然是一個正數
        long waitMicros =
            storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
                + (long) (freshPermits * stableIntervalMicros);    //    計算須要等待的時間（微秒）

        this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
        this.storedPermits -= storedPermitsToSpend;    //    減去本次消費的permit數
        return returnValue;
    }
    
    void resync(long nowMicros) {
        // if nextFreeTicket is in the past, resync to now
        if (nowMicros > nextFreeTicketMicros) {    //    表示當前能夠得到permit
            double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();    //    計算這段時間能夠生成多少個permit
            storedPermits = min(maxPermits, storedPermits + newPermits);    //    若是超過maxPermit，則取maxPermit，不然取存儲的permit+新生成的permit
            nextFreeTicketMicros = nowMicros;    //    設置下一次能夠得到permit的時間點爲當前時間
        }
    }
}

RateLimiter實現的令牌桶算法，不只能夠應對正常流量的限速，並且能夠處理突發暴增的請求，實現平滑限流。

經過代碼，咱們能夠看到它能夠預消費，怎麼講呢

nextFreeTicketMicros表示下一次請求得到permits的最先時間。每次受權一個請求之後，這個值會向後推移（PS：想象一下時間軸）即向將來推移。所以，大的請求會比小的請求推得更。這裏的大小指的是獲取permit的數量。這個應該很好理解，由於上一次請求獲取的permit數越多，那麼下一次再獲取受權時更待的時候會更長，反之，若是上一次獲取的少，那麼時間向後推移的就少，下一次得到許可的時間更短。可見，都是有代價的。正所謂：要浪漫就要付出代價。

還要注意到一點，就是獲取令牌和處理請求是兩個動做，並且，並非每一次都獲取一個，也不要想固然的認爲一個請求獲取一個permit（或者叫令牌），能夠再看看前面那幅圖

Stopwatch

一個以納秒爲單位度量流逝時間的對象。它是一個相對時間，而不是絕對時間。

Stopwatch stopwatch = Stopwatch.createStarted();
System.out.println("hahah");
stopwatch.stop();
Duration duration = stopwatch.elapsed();
System.out.println(stopwatch);

Semaphore（信號量）

A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each acquire() blocks if necessary until a permit is available, and then takes it. Each release() adds a permit, potentially releasing a blocking acquirer. However, no actual permit objects are used; the Semaphore just keeps a count of the number available and acts accordingly.

一個信號量維護了一系列permits。

每次調用acquire()方法獲取permit，若是必要的話會阻塞直到有一個permit可用爲止。

調用release()方法則會釋放本身持有的permit，即用完了再還回去。

信號量限制的是併發訪問臨界資源的線程數。

令牌桶算法 VS 漏桶算法

漏桶

漏桶的出水速度是恆定的，那麼意味着若是瞬時大流量的話，將有大部分請求被丟棄掉（也就是所謂的溢出）。

令牌桶

生成令牌的速度是恆定的，而請求去拿令牌是沒有速度限制的。這意味，面對瞬時大流量，該算法能夠在短期內請求拿到大量令牌，並且拿令牌的過程並非消耗很大的事情。

最後，不管是對於令牌桶拿不到令牌被拒絕，仍是漏桶的水滿了溢出，都是爲了保證大部分流量的正常使用，而犧牲掉了少部分流量，這是合理的，若是由於極少部分流量須要保證的話，那麼就可能致使系統達到極限而掛掉，得不償失。

小定律：排隊理論

https://en.wikipedia.org/wiki/Little%27s_law

the long-term average number L of customers in a stationary system is equal to the long-term average effective arrival rate λ multiplied by the average time W that a customer spends in the system. Expressed algebraically the law is：

在一個固定系統中，顧客的長期平均數量L等於顧客的長期平均到達速率λ乘以顧客在系統中平均花費的時間W。用公式表示爲：

雖然這看起來很容易，但這是一個很是顯著的舉世矚目的結果，由於這種關係「不受到達過程的分佈，服務分佈，服務順序，或其餘任何因素的影響」。這個結果適用於任何系統，特別是適用於系統內的系統。惟一的要求是系統必須是穩定的非搶佔式的。