来谈谈限流-RateLimiter源码分析

前一篇文章提到了限流的几种常见算法，本文将分析guava限流类RateLimiter的实现。

RateLimiter有两个实现类：SmoothBursty和SmoothWarmingUp，其都是令牌桶算法的变种实现，区别在于SmoothBursty加令牌的速度是恒定的，而SmoothWarmingUp会有个预热期，在预热期内加令牌的速度是慢慢增长的，直到达到固定速度为止。其适用场景是，对于有的系统而言刚启动时能承受的QPS较小，须要预热一段时间后才能达到最佳状态。

更多文章见我的博客：github.com/farmerjohng…

基本使用

RateLimiter的使用很简单：

//create方法传入的是每秒生成令牌的个数
RateLimiter rateLimiter= RateLimiter.create(1);
for (int i = 0; i < 5; i++) {
    //acquire方法传入的是须要的令牌个数，当令牌不足时会进行等待，该方法返回的是等待的时间
	double waitTime=rateLimiter.acquire(1);
	System.out.println(System.currentTimeMillis()/1000+" , "+waitTime);
}
复制代码

输出以下：

1548070953 , 0.0
1548070954 , 0.998356
1548070955 , 0.998136
1548070956 , 0.99982
复制代码

须要注意的是，当令牌不足时，acquire方法并不会阻塞本次调用，而是会算在下次调用的头上。好比第一次调用时，令牌桶中并无令牌，可是第一次调用也没有阻塞，而是在第二次调用的时候阻塞了1秒。也就是说，每次调用欠的令牌（若是桶中令牌不足）都是让下一次调用买单。

RateLimiter rateLimiter= RateLimiter.create(1);
double waitTime=rateLimiter.acquire(1000);
System.out.println(System.currentTimeMillis()/1000+" , "+waitTime);
waitTime=rateLimiter.acquire(1);
System.out.println(System.currentTimeMillis()/1000+" , "+waitTime);
复制代码

输出以下：

1548072250 , 0.0
1548073250 , 999.998773
复制代码

这样设计的目的是：

Last, but not least: consider a RateLimiter with rate of 1 permit per second, currently completely unused, and an expensive acquire(100) request comes. It would be nonsensical to just wait for 100 seconds, and /then/ start the actual task. Why wait without doing anything? A much better approach is to /allow/ the request right away (as if it was an acquire(1) request instead), and postpone /subsequent/ requests as needed. In this version, we allow starting the task immediately, and postpone by 100 seconds future requests, thus we allow for work to get done in the meantime instead of waiting idly.
复制代码

简单的说就是，若是每次请求都为本次买单会有没必要要的等待。好比说令牌增长的速度为每秒1个，初始时桶中没有令牌，这时来了个请求须要100个令牌，那须要等待100s后才能开始这个任务。因此更好的办法是先放行这个请求，而后延迟以后的请求。

另外，RateLimiter还有个tryAcquire方法，若是令牌够会当即返回true，不然当即返回false。

源码分析

本文主要分析SmoothBursty的实现。

首先看SmoothBursty中的几个关键字段:

// 桶中最多存放多少秒的令牌数
final double maxBurstSeconds;
//桶中的令牌个数
double storedPermits;
//桶中最多能存放多少个令牌，=maxBurstSeconds*每秒生成令牌个数
double maxPermits;
//加入令牌的平均间隔，单位为微秒，若是加入令牌速度为每秒5个，则该值为1000*1000/5
double stableIntervalMicros;
//下一个请求须要等待的时间
private long nextFreeTicketMicros = 0L; 
复制代码

RateLimiter的建立

先看建立RateLimiter的create方法。

// permitsPerSecond为每秒生成的令牌数
public static RateLimiter create(double permitsPerSecond) {
    return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
}

//SleepingStopwatch主要用于计时和休眠
static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
    //建立一个SmoothBursty
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
}
复制代码

create方法主要就是建立了一个SmoothBursty实例，并调用了其setRate方法。注意这里的maxBurstSeconds写死为1.0。

@Override
final void doSetRate(double permitsPerSecond, long nowMicros) {
    resync(nowMicros);
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);
}

void resync(long nowMicros) {
    // 若是当前时间比nextFreeTicketMicros大，说明上一个请求欠的令牌已经补充好了，本次请求不用等待
    if (nowMicros > nextFreeTicketMicros) {
      // 计算这段时间内须要补充的令牌，coolDownIntervalMicros返回的是stableIntervalMicros
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
     // 更新桶中的令牌，不能超过maxPermits
      storedPermits = min(maxPermits, storedPermits + newPermits);
      // 这里先设置为nowMicros
      nextFreeTicketMicros = nowMicros;
    }
}

@Override
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
    double oldMaxPermits = this.maxPermits;
    maxPermits = maxBurstSeconds * permitsPerSecond;
    if (oldMaxPermits == Double.POSITIVE_INFINITY) {
        // if we don't special-case this, we would get storedPermits == NaN, below
        storedPermits = maxPermits;
    } else {
        //第一次调用oldMaxPermits为0，因此storedPermits（桶中令牌个数）也为0
        storedPermits =
                (oldMaxPermits == 0.0)
                        ? 0.0 // initial state
                        : storedPermits * maxPermits / oldMaxPermits;
    }
}
复制代码

setRate方法中设置了maxPermits=maxBurstSeconds * permitsPerSecond;而maxBurstSeconds 为1，因此maxBurstSeconds只会保存1秒中的令牌数。

须要注意的是SmoothBursty是非public的类，也就是说只能经过RateLimiter.create方法建立，而该方法中的maxBurstSeconds 是写死1.0的，也就是说咱们只能建立桶大小为permitsPerSecond*1的SmoothBursty对象（固然反射的方式不在讨论范围），在guava的github仓库里有好几条issue（issue1,issue2,issue3,issue4）但愿能由外部设置maxBurstSeconds，可是并无看到官方人员的回复。而在惟品会的开源项目vjtools中，有人提出了这个问题，惟品会的同窗对guava的RateLimiter进行了拓展。

对于guava的这样设计我很不理解，有清楚的朋友能够说下~

到此为止一个SmoothBursty对象就建立好了，接下来咱们分析其acquire方法。

acquire方法

public double acquire(int permits) {
    // 计算本次请求须要休眠多久（受上次请求影响）
    long microsToWait = reserve(permits);
    // 开始休眠
    stopwatch.sleepMicrosUninterruptibly(microsToWait);
    return 1.0 * microsToWait / SECONDS.toMicros(1L);
}
 
final long reserve(int permits) {
    checkPermits(permits);
    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方法，可能会更新桶中的令牌值和nextFreeTicketMicros
    resync(nowMicros);
    // 若是上次请求花费的令牌尚未补齐，这里returnValue为上一次请求后须要等待的时间，不然为nowMicros
    long returnValue = nextFreeTicketMicros;
    double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
    // 缺乏的令牌数
    double freshPermits = requiredPermits - storedPermitsToSpend;
    // waitMicros为下一次请求须要等待的时间；SmoothBursty的storedPermitsToWaitTime返回0
    long waitMicros =
        storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
            + (long) (freshPermits * stableIntervalMicros);
    // 更新nextFreeTicketMicros
    this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
    // 减小令牌
    this.storedPermits -= storedPermitsToSpend;
    return returnValue;
}
复制代码

acquire中会调用reserve方法得到当前请求须要等待的时间，而后进行休眠。reserve方法最终会调用到reserveEarliestAvailable，在该方法中会先调用上文提到的resync方法对桶中的令牌进行补充（若是须要的话），而后减小桶中的令牌，以及计算此次请求欠的令牌数及须要等待的时间（由下次请求负责等待）。

若是上一次请求没有欠令牌或欠的令牌已经还清则返回值为nowMicros，不然返回值为上一次请求缺乏的令牌个数*生成一个令牌所须要的时间。

End

本文讲解了RateLimiter子类SmoothBursty的源码，对于另外一个子类SmoothWarmingUp的原理你们能够自行分析。相对于传统意义上的令牌桶，RateLimiter的实现仍是略有不一样，主要体如今一次请求的花费由下一次请求来承担这一点上。