這段時(shí)間因?yàn)闃I(yè)務(wù)需要，去學(xué)習(xí)使用了RateLimiter類來控制訪問皆警，發(fā)現(xiàn)十分方便好用爽彤，因此打算學(xué)習(xí)一下是如何實(shí)現(xiàn)的擒权。

RateLimiter經(jīng)常用于限制對(duì)一些物理資源或者邏輯資源的訪問速率。
重點(diǎn)在于控制的速率祥国。

基礎(chǔ)算法

是如何實(shí)現(xiàn)這種對(duì)速率的控制的呢昵观？這就跟RateLimiter的設(shè)計(jì)的算法——令牌桶算法 and 漏桶算法

令牌桶算法

令牌桶算法的原理是系統(tǒng)會(huì)以一個(gè)恒定的速度往桶里放入令牌，當(dāng)一個(gè)請(qǐng)求進(jìn)來的時(shí)候舌稀，則需要先從桶里獲取一個(gè)令牌啊犬，當(dāng)桶里沒有令牌可取時(shí)，則拒絕服務(wù)壁查。

1.每秒會(huì)有 r 個(gè)令牌放入桶中觉至，或者說，每過 1/r 秒桶中增加一個(gè)令牌（不是某個(gè)時(shí)間點(diǎn)一下子給出該秒所有的令牌）
2.桶中最多存放 b 個(gè)令牌睡腿，如果桶滿了语御，新放入的令牌會(huì)被丟棄
3.當(dāng)一個(gè) n 字節(jié)的數(shù)據(jù)包到達(dá)時(shí)峻贮，消耗 n 個(gè)令牌，然后發(fā)送該數(shù)據(jù)包
4.如果桶中可用令牌小于 n应闯，則該數(shù)據(jù)包將被緩存或丟棄

漏桶算法

漏桶算法思路很簡單纤控，水（請(qǐng)求）先進(jìn)入到漏桶里，漏桶以一定的速度出水碉纺，當(dāng)水流入速度過大會(huì)直接溢出船万，可以看出漏桶算法能強(qiáng)行限制數(shù)據(jù)的傳輸速率

區(qū)別：
漏桶算法能夠強(qiáng)行限制數(shù)據(jù)的傳輸速率。
令牌桶算法能夠在限制數(shù)據(jù)的平均傳輸速率的同時(shí)還允許某種程度的突發(fā)傳輸骨田。

RateLimiter的使用：

基本使用手冊(cè)里都有詳細(xì)描述 耿导，這里主要說一下之前遇到的問題
使用RateLimiter的時(shí)候會(huì)在最開始的時(shí)候create一個(gè)規(guī)定的速率。
這就是對(duì)速率的控制盛撑。

//每秒5個(gè)令牌碎节。以每0.2/個(gè)令牌的速度放進(jìn)去
RateLimiter rateLimiter = RateLimiter.create(5.0);

關(guān)于acquire 和 tryAcquire
double a = rateLimiter.acquire();
boolean b= rateLimiter.tryAcquire();

最開始，在代碼里使用了acquire方法抵卫，然后發(fā)現(xiàn)有些請(qǐng)求十分緩慢狮荔，后來發(fā)現(xiàn)是多余的請(qǐng)求會(huì)被阻塞，返回被阻塞的時(shí)間介粘。但是這樣會(huì)影響正常的訪問殖氏。之后改成了tryAcquire，tryAcquire是非阻塞的姻采，如果不能即時(shí)獲取到令牌雅采，會(huì)直接失敗，然后直接將返回false的請(qǐng)求給throw 出來慨亲。這樣去控制一些訪問的請(qǐng)求數(shù)婚瓜。

比如 我們按照我們上述規(guī)定的5個(gè)/s ，一秒內(nèi)5個(gè)請(qǐng)求進(jìn)來,每隔0.2s就可以允許一個(gè)請(qǐng)求刑棵，如果超過5個(gè)巴刻，那么就需要等待，如果用acquire蛉签，那么第六個(gè)請(qǐng)求需要等1.2秒之后才能獲取到令牌胡陪，這1.2s里 這個(gè)請(qǐng)求是在等待的。但是用tryAcquire的話就會(huì)立即返回false;

RateLimiter的實(shí)驗(yàn)：

       RateLimiter rateLimiter = RateLimiter.create(5);
            double a=rateLimiter.acquire(1);
            System.out.println("0 = "+a);

            double b=rateLimiter.acquire(3);
            System.out.println("1 = "+b);

            double c=rateLimiter.acquire(1);
            System.out.println("3 = "+c);

            double d=rateLimiter.acquire(18);
            System.out.println("4 = "+d);

            double e=rateLimiter.acquire(1);
            System.out.println("5 = "+e);

結(jié)果：

0 = 0.0 //第一個(gè)是預(yù)支的 
1 = 0.194598 //因?yàn)榈谝粋€(gè)預(yù)支了碍舍，所以第二次要的等1/5s
3 = 0.596369 //同理 第二次取了3個(gè) 就需要給1/5 * 3 s
4 = 0.198327
5 = 3.597525

和Semaphore 的區(qū)別

這個(gè)是JDK 里 concurrent下的一個(gè)類柠座，也是可以指定一個(gè)令牌總數(shù)。多線程情況下片橡，每個(gè)線程要繼續(xù)執(zhí)行妈经，需要獲取這個(gè)令牌。每個(gè)線程用完了放回去。主要是控制并發(fā)總數(shù)吹泡。
這里RateLimiter 是控制一個(gè)速率

RateLimiter的源碼：

SmoothBursty：恒定的速度放入令牌
SmoothWarmingUp：緩慢提升到一定值放入令牌
區(qū)別：可以控制速率
RateLimiter 中的時(shí)間窗口能且僅能為 1s录煤，可以見下面源碼分析。

 @VisibleForTesting
  static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
  //可以看這里是寫死的1s荞胡；
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
//這里才是主要設(shè)置令牌數(shù)的地方
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
  }


 static final class SmoothBursty extends SmoothRateLimiter {
    /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */
    final double maxBurstSeconds;

    SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {
      super(stopwatch);
      this.maxBurstSeconds = maxBurstSeconds;
    }

    @Override
    void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
      double oldMaxPermits = this.maxPermits;
    //允許保存的最大的permit 
      maxPermits = maxBurstSeconds * permitsPerSecond;
      if (oldMaxPermits == Double.POSITIVE_INFINITY) {
        // if we don't special-case this, we would get storedPermits == NaN, below
        storedPermits = maxPermits;
      } else {
        storedPermits =
            (oldMaxPermits == 0.0)
                ? 0.0 // initial state
                : storedPermits * maxPermits / oldMaxPermits;
      }
    }



 public double acquire(int permits) {
    long microsToWait = reserve(permits); 先計(jì)算獲取這些請(qǐng)求需要讓線程等待多長時(shí)間
    stopwatch.sleepMicrosUninterruptibly(microsToWait); //讓線程阻塞microTowait微秒長的時(shí)間
    return 1.0 * microsToWait / SECONDS.toMicros(1L);//返回阻塞的時(shí)間
  }


 final long reserve(int permits) {
    checkPermits(permits);//主要判斷permits為正
    synchronized (mutex()) { 
//stopwatch 這個(gè)是到當(dāng)前為止 等待睡眠的時(shí)間
      return reserveAndGetWaitLength(permits, stopwatch.readMicros());
    }
  }


//返回需要等待的時(shí)間（正數(shù)）
  final long reserveAndGetWaitLength(int permits, long nowMicros) {
//獲取持續(xù)的時(shí)間
    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
    return max(momentAvailable - nowMicros, 0);
  }



//返回
  final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
    resync(nowMicros); //補(bǔ)充令牌
    long returnValue = nextFreeTicketMicros;
    double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
//需要重新生成多少個(gè)permit
    double freshPermits = requiredPermits - storedPermitsToSpend;
//生成freshPermits個(gè)permit需要的時(shí)間 這個(gè)時(shí)間就是下次我們需要減去的時(shí)間  
  long waitMicros =
        storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
            + (long) (freshPermits * stableIntervalMicros);

//下次獲取的時(shí)候需要減去的時(shí)間=當(dāng)前下次獲取的時(shí)候需要減去的時(shí)間+這次需要等待的時(shí)間
    this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
    this.storedPermits -= storedPermitsToSpend; //減去消耗的令牌
    return returnValue;
  }



//補(bǔ)充令牌
 void resync(long nowMicros) {
//如果需要的時(shí)間大于這次獲取的時(shí)候需要減去的時(shí)間
    if (nowMicros > nextFreeTicketMicros) {
    // nextFreeTicketMicros的意思是：下次獲取的時(shí)候需要減去的時(shí)間
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
//storedPermits能保存的permit
//就是過去一段時(shí)間的利用不足的storedPermits 最多不超過create的permit
      storedPermits = min(maxPermits, storedPermits + newPermits);
      nextFreeTicketMicros = nowMicros;
    }
  }