常用限流算法与Guava RateLimiter源码解析

计数器算法

计数器算法是限流算法中最简单的一种算法，限制在一个时间窗口内，至多处理多少个请求。比如每分钟最多处理10个请求，则从第一个请求进来的时间为起点，60s的时间窗口内只允许最多处理10个请求。下一个时间窗口又以前一时间窗口过后第一个请求进来的时间为起点。常见的比如一分钟内只能获取一次短信验证码的功能可以通过计数器算法来实现。

Guava RateLimiter解析

Guava是Google开源的一个工具包，其中的RateLimiter是实现了令牌桶算法的一个限流工具类。在pom.xml中添加guava依赖，即可使用RateLimiter

<groupId>com.google.guava</groupId>

<artifactId>guava</artifactId>

</dependency>

如下测试代码示例了RateLimiter的用法,

public static void main(String[] args) {

RateLimiter rateLimiter = RateLimiter.create(1); //创建一个每秒产生一个令牌的令牌桶

for(int i=1;i<=5;i++) {

double waitTime = rateLimiter.acquire(i); //一次获取i个令牌

System.out.println("acquire:" + i + " waitTime:" + waitTime);

}

运行后，输出如下，

acquire:1 waitTime:0.0

acquire:2 waitTime:0.997729

acquire:3 waitTime:1.998076

acquire:4 waitTime:3.000303

acquire:5 waitTime:4.000223

第一次获取一个令牌时，等待0s立即可获取到（这里之所以不需要等待是因为令牌桶的预消费特性），第二次获取两个令牌，等待时间1s，这个1s就是前面获取一个令牌时因为预消费没有等待延到这次来等待的时间，这次获取两个又是预消费，所以下一次获取（取3个时）就要等待这次预消费需要的2s了，依此类推。可见预消费不需要等待的时间都由下一次来买单，以保障一定的平均处理速率（上例为1s一次）。

RateLimiter有两种实现：

SmoothBursty：令牌的生成速度恒定。使用 RateLimiter.create(double permitsPerSecond) 创建的是 SmoothBursty 实例。

SmoothWarmingUp：令牌的生成速度持续提升，直到达到一个稳定的值。WarmingUp，顾名思义就是有一个热身的过程。使用 RateLimiter.create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) 时创建就是 SmoothWarmingUp 实例，其中 warmupPeriod 就是热身达到稳定速度的时间。

类结构如下

ratelimiter-struct

关键属性及方法解析（以 SmoothBursty 为例）

1.关键属性

/** 桶中当前拥有的令牌数. */

double storedPermits;

/** 桶中最多可以保存多少秒存入的令牌数 */

double maxBurstSeconds;

/** 桶中能存储的最大令牌数，等于storedPermits*maxBurstSeconds. */

double maxPermits;

/** 放入令牌的时间间隔*/

double stableIntervalMicros;

/** 下次可获取令牌的时间点，可以是过去也可以是将来的时间点*/

private long nextFreeTicketMicros = 0L;

2.关键方法

调用 RateLimiter.create(double permitsPerSecond) 方法时，创建的是 SmoothBursty 实例，默认设置 maxBurstSeconds 为1s。SleepingStopwatch 是guava中的一个时钟类实现。

@VisibleForTesting

static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {

RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);

rateLimiter.setRate(permitsPerSecond);

return rateLimiter;

}

SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {

super(stopwatch);

this.maxBurstSeconds = maxBurstSeconds;

}

并通过调用 SmoothBursty.doSetRate(double, long) 方法进行初始化，该方法中:

调用 resync(nowMicros) 对 storedPermits 与 nextFreeTicketMicros 进行了调整——如果当前时间晚于 nextFreeTicketMicros，则计算这段时间内产生的令牌数，累加到 storedPermits 上，并更新下次可获取令牌时间 nextFreeTicketMicros 为当前时间。

计算 stableIntervalMicros 的值，1/permitsPerSecond。

调用 doSetRate(double, double) 方法计算 maxPermits 值（maxBurstSeconds*permitsPerSecond），并根据旧的 maxPermits 值对 storedPermits 进行调整。

源码如下所示

@Override

final void doSetRate(double permitsPerSecond, long nowMicros) {

resync(nowMicros);

double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;

this.stableIntervalMicros = stableIntervalMicros;

doSetRate(permitsPerSecond, stableIntervalMicros);

}

/** Updates {@code storedPermits} and {@code nextFreeTicketMicros} based on the current time. */

void resync(long nowMicros) {

// if nextFreeTicket is in the past, resync to now

if (nowMicros > nextFreeTicketMicros) {

double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();

storedPermits = min(maxPermits, storedPermits + newPermits);

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 {

storedPermits =

(oldMaxPermits == 0.0)

? 0.0 // initial state

: storedPermits * maxPermits / oldMaxPermits;

}

调用 acquire(int) 方法获取指定数量的令牌时，

调用 reserve(int) 方法，该方法最终调用 reserveEarliestAvailable(int, long) 来更新下次可取令牌时间点与当前存储的令牌数，并返回本次可取令牌的时间点，根据该时间点计算需要等待的时间

阻塞等待1中返回的等待时间

返回等待的时间（秒）

源码如下所示

/** 获取指定数量（permits）的令牌，阻塞直到获取到令牌，返回等待的时间*/

@CanIgnoreReturnValue

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);

}

/** 针对此次需要获取的令牌数更新下次可取令牌时间点与存储的令牌数，返回本次可取令牌的时间点*/

@Override

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {

resync(nowMicros); // 更新当前数据

long returnValue = nextFreeTicketMicros;

double storedPermitsToSpend = min(requiredPermits, this.storedPermits); // 本次可消费的令牌数

double freshPermits = requiredPermits - storedPermitsToSpend; // 需要新增的令牌数

long waitMicros =

storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)

+ (long) (freshPermits * stableIntervalMicros); // 需要等待的时间

this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros); // 更新下次可取令牌的时间点

this.storedPermits -= storedPermitsToSpend; // 更新当前存储的令牌数

return returnValue;

}

acquire(int) 方法是获取不到令牌时一直阻塞，直到获取到令牌，tryAcquire(int,long,TimeUnit) 方法则是在指定超时时间内尝试获取令牌，如果获取到或超时时间到则返回是否获取成功

先判断是否能在指定超时时间内获取到令牌，通过 nextFreeTicketMicros <= timeoutMicros + nowMicros 是否为true来判断，即可取令牌时间早于当前时间加超时时间则可取（预消费的特性），否则不可获取。

如果不可获取，立即返回false。

如果可获取，则调用 reserveAndGetWaitLength(permits, nowMicros) 来更新下次可取令牌时间点与当前存储的令牌数，返回等待时间（逻辑与前面相同），并阻塞等待相应的时间，返回true。

源码如下所示

public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {

long timeoutMicros = max(unit.toMicros(timeout), 0);

checkPermits(permits);

long microsToWait;

synchronized (mutex()) {

long nowMicros = stopwatch.readMicros();

if (!canAcquire(nowMicros, timeoutMicros)) { //判断是否能在超时时间内获取指定数量的令牌

return false;

} else {

microsToWait = reserveAndGetWaitLength(permits, nowMicros);

}

stopwatch.sleepMicrosUninterruptibly(microsToWait);

return true;

}

private boolean canAcquire(long nowMicros, long timeoutMicros) {

return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros; //只要可取时间小于当前时间+超时时间，则可获取（可预消费的特性！）

}

@Override

final long queryEarliestAvailable(long nowMicros) {

return nextFreeTicketMicros;

}

以上就是 SmoothBursty 实现的基本处理流程。注意两点：

RateLimiter 通过限制后面请求的等待时间，来支持一定程度的突发请求——预消费的特性。

RateLimiter 令牌桶的实现并不是起一个线程不断往桶里放令牌，而是以一种延迟计算的方式（参考resync函数），在每次获取令牌之前计算该段时间内可以产生多少令牌，将产生的令牌加入令牌桶中并更新数据来实现，比起一个线程来不断往桶里放令牌高效得多。（想想如果需要针对每个用户限制某个接口的访问，则针对每个用户都得创建一个RateLimiter，并起一个线程来控制令牌存放的话，如果在线用户数有几十上百万，起线程来控制是一件多么恐怖的事情）

深圳网站建设www.sz886.com