快捷導航

RateLimiter 源碼分析

更新時間：2017年09月28日 09:13:52 作者：foolishAndStupid

本文主要對ratelimiter的常用方法以及源碼進行了分析解讀，具有一定參考價值，需要的朋友可以了解下。

俗話說得好，緩存，限流和降級是系統(tǒng)的三把利劍。剛好項目中每天早上導出數(shù)據(jù)時因調(diào)訂單接口頻率過高，訂單系統(tǒng)擔心會對用戶側(cè)的使用造成影響，讓我們對調(diào)用限速一下，所以就正好用上了。

常用的限流算法有2種：漏桶算法和令牌桶算法。

漏桶算法

漏桶算法：請求先進入“桶”中，然后桶以一定的速率處理請求。如果請求的速率過快會導致桶溢出。根據(jù)描述可以知道，漏桶算法會強制限制請求處理的速度。任你請求的再快還是再慢，我都是以這種速率來處理。

但是對于很多情況下，除了要求能夠限制平均處理速度外，還要求能允許一定程度的的突發(fā)情況。這樣的話，漏桶算法就不合適了，用令牌桶算法更合適。

令牌桶算法

令牌桶算法的原理是：系統(tǒng)以恒定的速率往桶里丟一定數(shù)量的令牌，請求只有拿到了令牌才能處理。當桶里沒有令牌時便可拒絕服務(wù)。

Guava中的Ratelimiter便是實現(xiàn)的令牌桶算法，同時能支持一定程度的突發(fā)請求。

private static RateLimiter one=RateLimiter.create(2);//每秒2個
  private static RateLimiter two=RateLimiter.create(2);//每秒2個
  private RateLimitUtil(){};
  public static void acquire(RateLimiter r,int num){
    double time =r.acquire(num);
    System.out.println("wait time="+time);
  }
  public static void main(String[] args) throws InterruptedException {
    acquire(one,1);
    acquire(one,1);
    acquire(one,1);
    System.out.println("-----");
    acquire(two,10);
    acquire(two,1);
  }

輸出結(jié)果：

wait time=0.0
wait time=0.499163
wait time=0.489308
-----
wait time=0.0
wait time=4.497819

可以看到，我們以每秒2個請求的速度生成令牌。對one來說，當?shù)?次和第3次獲取請求的時候，等待的時間加起來就差不多剛好是1秒。對two來說，當?shù)谝淮潍@取了10個令牌之后，第二次獲取1個請求，就差不多等待5S（10/2=5）?？梢钥吹剑琯uava通過限制后面請求的等待時間，來支持一定程度的突發(fā)請求。

接下來，就是通過源碼來解析它！

當我第一次看到令牌桶的算法描述的時候，我還以為真是有一個線程每隔X秒往一個類似計數(shù)器的地方加數(shù)字呢….

guava的限流算法有2種模式，一種是穩(wěn)定速度，還有一種是生成令牌的速度慢慢提升直到維持在一個穩(wěn)定的速度。2種模式原理類似，只是在具體等待多久的時間計算上有區(qū)別。以下就專門指穩(wěn)定速度的模式。

先來看看它的acquire()方法：

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

主要分3步：

1. 根據(jù)limiter創(chuàng)建時傳入的參數(shù)，計算出生成這些數(shù)量的令牌需要多長的時間。

2. 讓線程阻塞microTowait這么長的時間（單位：微秒）

3. 再返回阻塞了多久，單位：秒

具體它是怎么計算需要多長時間的呢？讓我們來看看reserve(permits)方法。

final long reserve(int permits) {
  checkPermits(permits);//檢查參數(shù)是否合法
  synchronized (mutex()) {
   return reserveAndGetWaitLength(permits, stopwatch.readMicros());
  }
 }
    ↓
    ↓
    ↓
 final long reserveAndGetWaitLength(int permits, long nowMicros) {
  long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
  return max(momentAvailable - nowMicros, 0);
 }
    ↓
    ↓
    ↓
 final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  resync(nowMicros);//here
  long returnValue = nextFreeTicketMicros;
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
  double freshPermits = requiredPermits - storedPermitsToSpend;
  long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
    + (long) (freshPermits * stableIntervalMicros);
  this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;
  this.storedPermits -= storedPermitsToSpend;
  return returnValue;
 }

最終調(diào)用的是reserveEarliestAvailable方法。先看看resync(nowMicros)方法。

private void resync(long nowMicros) {
  // if nextFreeTicket is in the past, resync to now
  if (nowMicros > nextFreeTicketMicros) {
   storedPermits = min(maxPermits,
     storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
   nextFreeTicketMicros = nowMicros;
  }
 }

nextFreeTicketMicros的意思是：下次獲取的時候需要減去的時間。如果是第一次調(diào)用accquire()方法，那nowMicros - nextFreeTicketMicros 就是從初始化（初始化的時候會給nextFreeTicketMicros 賦值一次,具體可以看RateLimiter的構(gòu)造器）到第一次請求，這中間發(fā)生的時間。

這個方法的意思，如果當前時間比上一輪設(shè)置的下次獲取的時間大（因為存在提前獲取的情況，比如上次直接獲取了10個，那上輪設(shè)置的nextFreeTicketMicros就是上一輪的時間+5s。后面會提到），那就計算這個中間理論上能生成多少的令牌。比如這中間隔了1秒鐘，然后stableIntervalMicros=5000（穩(wěn)定生成速度的情況下）,那么，就這中間就可以生成2個令牌。再加上它原先存儲的storedPermits個，如果比maxPermits大，那最大也只能存maxPermits這么多。如果比maxPermits小，那就是storedPermits=原先存的+這中間生成的數(shù)量。同時記錄下下次獲取的時候需要減去的時間，也就是當前時間（nextFreeTicketMicros ）。

接下來繼續(xù)看reserveEarliestAvailable方法：

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) { //1
  resync(nowMicros);   //2
  long returnValue = nextFreeTicketMicros;//3
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits);//4
  double freshPermits = requiredPermits - storedPermitsToSpend;//5
  long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
    + (long) (freshPermits * stableIntervalMicros);//6
  this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;//7
  this.storedPermits -= storedPermitsToSpend;//8
  return returnValue;//9
 }

我們一行一行來看：

第二行設(shè)置好之后。第3行中將下次獲取的時候需要減去的時間作為返回值（這點很重要）。

這2句是什么意思呢？

其實這2句就是使得RateLimiter能一定程度的突發(fā)請求的原因。假設(shè)requiredPermits=10，而我們能存的storedPermits=2，那么freshPermits=8，也就是多取了8個。而第6行就是計算這多取的8個需要多長時間才能生成？需要3秒。那么，就將這3秒鐘加到我們前面賦值的“下次獲取的時候需要減去的時間 ”。

比如在05秒的時候一次性獲取了10個，那么，第7行的意思就是nextFreeTicketMicros=13S對應(yīng)的系統(tǒng)的毫秒數(shù)。然后storedPermits就是-8。當過了1秒鐘，下一次請求來調(diào)用acquire(1)的時候，resync方法中由于nowMicros

final long reserveAndGetWaitLength(int permits, long nowMicros) {
  long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
  return max(momentAvailable - nowMicros, 0);//取較大的值
 }

也就是說，reserveAndGetWaitLength會返回max(13-6,0)，也就是7。而該方法的返回值又是用于sleep線程的，也就是我們在一開始看到的：

public double acquire(int permits) {
  long microsToWait = reserve(permits);
  stopwatch.sleepMicrosUninterruptibly(microsToWait);
  return 1.0 * microsToWait / SECONDS.toMicros(1L);
 }

總結(jié)起來，最主要的是nowMicros,nextFreeTicketMicros這2個值。nextFreeTicketMicros在一開始構(gòu)造器執(zhí)行的時候會賦值一次為構(gòu)造器執(zhí)行的時間。當?shù)谝淮握{(diào)用accquire()的時候，resync會被執(zhí)行，然后在accquire()中將nextFreeTicketMicros設(shè)置為當前時間。但是，需要注意的是，在reserveEarliestAvailable中會根據(jù)請求的令牌數(shù)和當前存儲的令牌數(shù)進行比較。如果請求的令牌數(shù)很大，則會計算出生成這些多余的令牌需要的時間，并加在nextFreeTicketMicros上，從而保證下次調(diào)用accquire()的時候，根據(jù)nextFreeTicketMicros和當時的nowMicros相減，若>0，則需要等到對應(yīng)的時間。也就能應(yīng)對流量的突增情況了。

所以最重要的是nextFreeTicketMicros，它記錄了你這次獲取的時候，能夠開始生成令牌的時間。比如當前是05S，那若nextFreeTicketMicros=10，表示它要到10S才能開始生成令牌，誰叫前面的多拿了這么多呢。至于它這次是多拿了還是只是拿一個令牌，等待時間都是這么多。如果這次又多拿了，那下次就等待更久！

private static RateLimiter too=RateLimiter.create(2);//每秒2個
  private RateLimitUtil(){};
  public static void acquire(RateLimiter r,int num){
    double time =r.acquire(num);
    System.out.println("wait time="+time);
  }
  public static void main(String[] args) throws InterruptedException {
    acquire(too,1);
    acquire(too,10);//只等待了0.5秒就獲取了10個
    acquire(too,10);//等待了5秒就獲取了10個
    acquire(too,1);//雖然只獲取1個，也是等待5秒
  }

總結(jié)

以上就是本文關(guān)于RateLimiter 常用方法以及源碼分析的全部內(nèi)容，希望對大家有所幫助。感興趣的朋友可以參閱：關(guān)于Openfire集群源碼的分析、 Spring SpringMVC在啟動完成后執(zhí)行方法源碼解析、 Java查看本機端口是否被占用源碼等。感謝大家對腳本之家網(wǎng)站的支持！

您可能感興趣的文章: