RateLimiter 源码分析(Guava 和 Sentinel 实现)

本文主要介绍关于流控的两部分内容。

第一部分介绍 Guava 中 RateLimiter 的源码,包括它的两种模式,目前网上大部分文章只分析简单的 SmoothBursty 模式,而没有分析带有预热的 SmoothWarmingUp。第二部分介绍 Sentinel 中流控的实现,本文不要求读者了解 Sentinel,这部分内容和 Sentinel 耦合很低,所以读者不需要有阅读压力。Sentinel 中流控设计是参考 Guava RateLimiter 的,所以阅读第二部分内容,需要有第一部分内容的背景。

Guava RateLimiter

RateLimiter 基于漏桶算法,但它参考了令牌桶算法,这里不讨论流控算法,请自行查找资料。本文使用 Guava 版本是 26.0-jre。

RateLimiter 使用介绍

RateLimiter 的接口非常简单,它有两个静态方法用来实例化,实例化以后,我们只需要关心 acquire 就行了,甚至都没有 release 操作。

// RateLimiter 接口列表:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 实例化的两种方式:
public static RateLimiter create(double permitsPerSecond){}
public static RateLimiter create(double permitsPerSecond,long warmupPeriod,TimeUnit unit) {}

public double acquire() {}
public double acquire(int permits) {}

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

public final double getRate() {}
public final void setRate(double permitsPerSecond) {}

RateLimiter 的作用是用来限流的,我们知道 java 并发包中提供了 Semaphore,它也能够提供对资源使用进行控制,我们看一下下面的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Semaphore
Semaphore semaphore = new Semaphore(10);
for (int i = 0; i < 100; i++) {
executor.submit(new Runnable() {
@Override
public void run() {
semaphore.acquireUninterruptibly(1);
try {
doSomething();
} finally {
semaphore.release();
}
}
});
}

Semaphore 用来控制同时访问某个资源的并发数量,如上面的代码,我们设置 100 个线程工作,但是我们能做到最多只有 10 个线程能同时到 doSomething() 方法中。它控制的是并发数量

而 RateLimiter 是用来控制访问资源的速率(rate)的,它强调的是控制速率。比如控制每秒只能有 100 个请求通过,比如允许每秒发送 1MB 的数据。

它的构造方法指定一个 permitsPerSecond 参数,代表每秒钟产生多少个 permits,这就是我们的速率。

RateLimiter 允许预占未来的令牌,比如,每秒产生 5 个 permits,我们可以单次请求 100 个 permits,这样,紧接着的下一个请求需要等待大概 20 秒才能获取到 permits。

SmoothRateLimiter 介绍

RateLimiter 目前只有一个子类,那就是抽象类 SmoothRateLimiter,SmoothRateLimiter 有两个实现类,也就是我们这边要介绍的两种模式,我们先简单介绍下中间的抽象类 SmoothRateLimiter,然后后面分两个小节分别介绍它的两个实现类。

1

RateLimiter 作为抽象类,只有两个属性:

1
2
3
private final SleepingStopwatch stopwatch;

private volatile Object mutexDoNotUseDirectly;

stopwatch 非常重要,它用来“计时”,RateLimiter 把实例化的时间设置为 0 值,后续都是取相对时间,用微秒表示。

mutexDoNotUseDirectly 用来做锁,RateLimiter 依赖于 synchronized 来控制并发,所以我们之后可以看到,各个属性甚至都没有用 volatile 修饰。

然后我们来看 SmoothRateLimiter 的属性,分别代表什么意思。

1
2
3
4
5
6
7
8
9
10
11
12
// 当前还有多少 permits 没有被使用,被存下来的 permits 数量
double storedPermits;

// 最大允许缓存的 permits 数量,也就是 storedPermits 能达到的最大值
double maxPermits;

// 每隔多少时间产生一个 permit,
// 比如我们构造方法中设置每秒 5 个,也就是每隔 200ms 一个,这里单位是微秒,也就是 200,000
double stableIntervalMicros;

// 下一次可以获取 permits 的时间,这个时间是相对 RateLimiter 的构造时间的,是一个相对时间,理解为时间戳吧
private long nextFreeTicketMicros = 0L;

其实,看到这几个属性,我们就可以大致猜一下它的内部实现了:

nextFreeTicketMicros 是一个很关键的属性。我们每次获取 permits 的时候,先拿 storedPermits 的值,因为它是存货,如果够,storedPermits 减去相应的值就可以了,如果不够,那么还需要将 nextFreeTicketMicros 往前推,表示我预占了接下来多少时间的量了。那么下一个请求来的时候,如果还没到 nextFreeTicketMicros 这个时间点,需要 sleep 到这个点再返回,当然也要将这个值再往前推。

大家在这里可能会有疑惑,因为时间是一直往前走的,应该要一直往池中添加 permits,所以 storedPermits 的值需要不断往上添加,难道需要另外开启一个线程来添加 permits?其实不是的,只需要在关键的操作中同步一下,重新计算就好了。

SmoothBursty 分析

我们先从比较简单的 SmoothBursty 出发,来分析 RateLimiter 的源码,之后我们再分析 SmoothWarmingUp。

Bursty 是突发的意思,它说的不是下面这个意思:我们设置了 1k 每秒,而我们可以一次性获取 5k 的 permits,这个场景表达的不是突发,而是在说预先占有了接下来几秒产生的 permits。希望大家不要被误导了。

突发说的是,RateLimiter 会缓存一定数量的 permits 在池中,这样对于突发请求,能及时得到满足。想象一下我们的某个接口,很久没有请求过来,突然同时来了好几个请求,如果我们没有缓存一些 permits 的话,很多线程就需要等待了。

SmoothBursty 默认缓存最多 1 秒钟的 permits,不可以修改。

RateLimiter 的静态构造方法:

1
2
3
public static RateLimiter create(double permitsPerSecond) {
return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
}

构造参数 permitsPerSecond 指定每秒钟可以产生多少个 permits。

1
2
3
4
5
static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
rateLimiter.setRate(permitsPerSecond);
return rateLimiter;
}

我们看到,这里实例化的是 SmoothBursty 的实例,它的构造方法很简单,而且它只有一个属性 maxBurstSeconds,这里就不贴代码了。

构造函数指定了 maxBurstSeconds 为 1.0,也就是说,最多会缓存 1 秒钟,也就是 (1.0 * permitsPerSecond) 这么多个 permits 到池中。

这个 1.0 秒,关系到 storedPermits 和 maxPermits:

0 <= storedPermits <= maxPermits = permitsPerSecond

我们继续往后看 setRate 方法:

1
2
3
4
5
6
7
public final void setRate(double permitsPerSecond) {
checkArgument(
permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
synchronized (mutex()) {
doSetRate(permitsPerSecond, stopwatch.readMicros());
}
}

setRate 这个方法是一个 public 方法,它可以用来调整速率。我们这边继续跟的是初始化过程,但是大家提前知道这个方法是用来调整速率用的,对理解源码有很大的帮助。注意看,这里用了 synchronized 控制并发。

1
2
3
4
5
6
7
8
9
@Override
final void doSetRate(double permitsPerSecond, long nowMicros) {
// 同步
resync(nowMicros);
// 计算属性 stableIntervalMicros
double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
this.stableIntervalMicros = stableIntervalMicros;
doSetRate(permitsPerSecond, stableIntervalMicros);
}

resync 方法很简单,它用来调整 storedPermits 和 nextFreeTicketMicros。这就是我们说的,在关键的节点,需要先更新一下 storedPermits 到正确的值。

1
2
3
4
5
6
7
8
9
void resync(long nowMicros) {
// 如果 nextFreeTicket 已经过掉了,想象一下很长时间都没有再次调用 limiter.acquire() 的场景
// 需要将 nextFreeTicket 设置为当前时间,重新计算 storedPermits
if (nowMicros > nextFreeTicketMicros) {
double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
storedPermits = min(maxPermits, storedPermits + newPermits);
nextFreeTicketMicros = nowMicros;
}
}

coolDownIntervalMicros() 这个方法大家先不用关注,可以看到,在 SmoothBursty 类中的实现是直接返回了 stableIntervalMicros 的值,也就是我们说的,每产生一个 permit 的时间长度。

当然了,细心的读者,可能会发现,此时的 stableIntervalMicros 其实没有设置,也就是说,上面发生了一次除以 0 值的操作,得到的 newPermits 其实是一个无穷大。而 maxPermits 此时还是 0 值,不过这里其实没有关系。

我们回到前面一个方法,resync 同步以后,会设置 stableIntervalMicros 为一个正确的值,然后进入下面的方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
double oldMaxPermits = this.maxPermits;
// 这里计算了,maxPermits 为 1 秒产生的 permits
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 的值域变化了,需要等比例缩放
storedPermits =
(oldMaxPermits == 0.0)
? 0.0 // initial state
: storedPermits * maxPermits / oldMaxPermits;
}
}

上面这个方法,我们要这么看,原来的 RateLimiter 是用某个 permitsPerSecond 值初始化的,现在我们要调整这个频率。对于 maxPermits 来说,是重新计算,而对于 storedPermits 来说,是做等比例的缩放。

到此,构造方法就完成了,我们得到了一个 RateLimiter 的实现类 SmoothBursty 的实例,可能上面的源码你还是会有一些疑惑,不过也没关系,继续往下看,可能你的很多疑惑就解开了。

接下来,我们来分析 acquire 方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@CanIgnoreReturnValue
public double acquire() {
return acquire(1);
}

@CanIgnoreReturnValue
public double acquire(int permits) {
// 预约,如果当前不能直接获取到 permits,需要等待
// 返回值代表需要 sleep 多久
long microsToWait = reserve(permits);
// sleep
stopwatch.sleepMicrosUninterruptibly(microsToWait);
// 返回 sleep 的时长
return 1.0 * microsToWait / SECONDS.toMicros(1L);
}

我们来看 reserve 方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
final long reserve(int permits) {
checkPermits(permits);
synchronized (mutex()) {
return reserveAndGetWaitLength(permits, stopwatch.readMicros());
}
}

final long reserveAndGetWaitLength(int permits, long nowMicros) {
// 返回 nextFreeTicketMicros
long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
// 计算时长
return max(momentAvailable - nowMicros, 0);
}

继续往里看:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Override
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
// 这里做一次同步,更新 storedPermits 和 nextFreeTicketMicros (如果需要)
resync(nowMicros);
// 返回值就是 nextFreeTicketMicros,注意刚刚已经做了 resync 了,此时它是最新的正确的值
long returnValue = nextFreeTicketMicros;
// storedPermits 中可以使用多少个 permits
double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
// storedPermits 中不够的部分
double freshPermits = requiredPermits - storedPermitsToSpend;
// 为了这个不够的部分,需要等待多久时间
long waitMicros =
storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) // 这部分固定返回 0
+ (long) (freshPermits * stableIntervalMicros);
// 将 nextFreeTicketMicros 往前推
this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
// storedPermits 减去被拿走的部分
this.storedPermits -= storedPermitsToSpend;
return returnValue;
}

我们可以看到,获取 permits 的时候,其实是获取了两部分,一部分来自于存量 storedPermits,存量不够的话,另一部分来自于预占未来的 freshPermits

这里提一个关键点吧,我们看到,返回值是 nextFreeTicketMicros 的旧值,因为只要到这个时间点,就说明当次 acquire 可以成功返回了,而不管 storedPermits 够不够。如果 storedPermits 不够,会将 nextFreeTicketMicros 往前推一定的时间,预占了一定的量。

到这里,acquire 方法就分析完了,大家看到这里,逆着往前看就是了。应该说,SmoothBursty 的源码还是非常简单的。

SmoothWarmingUp 分析

分析完了 SmoothBursty,我们再来分析 SmoothWarmingUp 会简单一些。我们说过,SmoothBursty 可以处理突发请求,因为它会缓存最多 1 秒的 permits,而待会我们会看到 SmoothWarmingUp 完全不同的设计。

SmoothWarmingUp 适用于资源需要预热的场景,比如我们的某个接口业务,需要使用到数据库连接,由于连接需要预热才能进入到最佳状态,如果我们的系统长时间处于低负载或零负载状态(当然,应用刚启动也是一样的),连接池中的连接慢慢释放掉了,此时我们认为连接池是冷的。

假设我们的业务在稳定状态下,正常可以提供最大 1000 QPS 的访问,但是如果连接池是冷的,我们就不能让系统达到 1000 个 QPS,要限制住突发流量,因为这会拖垮我们的系统,我们应该有个预热升温的过程。

对应到 SmoothWarmingUp 中,如果系统处于低负载状态,storedPermits 会一直增加,当请求来的时候,我们要从 storedPermits 中取 permits,最关键的点在于,从 storedPermits 中取 permits 的操作是比较耗时的,因为没有预热。

回顾一下前面介绍的 SmoothBursty,它从 storedPermits 中获取 permits 是不需要等待时间的,因为它是存货,而这边洽洽相反,从 storedPermits 获取需要更多的时间,这是最大的不同,先理解这一点,能帮助你更好地理解源码。

大家先有一些粗的概念,然后我们来看下面这个图:

2

这个图不容易看懂,X 轴代表 storedPermits 的数量,Y 轴代表获取一个 permits 需要的时间。简单粗暴地说就是:存货越多,代表系统越冷,获取令牌所需时间越多。

假设指定 permitsPerSecond 为 10,那么 stableInterval 为 100ms,而 coldInterval 是 3 倍,也就是 300ms(coldFactor,3 倍是写死的,用户不能修改)。也就是说,当达到 maxPermits 时,此时处于系统最冷的时候,获取一个 permit 需要 300ms,而如果 storedPermits 小于 thresholdPermits 的时候,只需要 100ms。

利用 “获取冷的 permits ” 需要等待更多时间,来限制突发请求通过,达到系统预热的目的。

想象有一条垂直线 x=k,它与 X 轴的交点 k 代表当前 storedPermits 的数量:

  • 当系统在非常繁忙的时候,这条线停留在 x=0 处,此时 storedPermits 为 0
  • 当 limiter 没有被使用的时候,这条线慢慢往右移动,直到 x=maxPermits 处;
  • 如果 limiter 被重新使用,那么这条线又慢慢往左移动,直到 x=0 处;

当 【thresholdPermits <= storedPermits <= maxPermits】 状态时,我们认为 limiter 中的 permits 是冷的,此时获取一个 permit 需要较多的时间,因为需要预热,有一个关键的分界点是 thresholdPermits

预热时间是我们在构造的时候指定的,图中梯形的面积就是预热时间,因为预热完成后,我们能进入到一个稳定的速率中(stableInterval),下面我们 根据构造参数计算出 thresholdPermits 和 maxPermits 的值

有一个关键点,从 thresholdPermits 到 0 的时间,是从 maxPermits 到 thresholdPermits 时间的一半,也就是梯形的面积是长方形面积的 2 倍,梯形的面积是 warmupPeriod。

3

之所以长方形的面积是 warmupPeriod/2,也就是梯形面积的一半,是因为 coldFactor 是硬编码的 3。具体的可以参考一下文章下面评论区的讨论。

梯形面积为 warmupPeriod,而长方形面积为 stableInterval * thresholdPermits,即:

1
warmupPeriod = 2 * stableInterval * thresholdPermits

由此,我们得出 thresholdPermits 的值:

1
thresholdPermits = 0.5 * warmupPeriod / stableInterval

然后我们根据梯形面积的计算公式:

1
warmupPeriod = 0.5 * (stableInterval + coldInterval) * (maxPermits - thresholdPermits)

得出 maxPermits 为:

1
maxPermits = thresholdPermits + 2.0 * warmupPeriod / (stableInterval + coldInterval)

这样,我们就得到了 thresholdPermits 和 maxPermits 的值。

接下来,我们来看一下冷却时间间隔,它指的是 storedPermits 中每个 permit 的增长速度,也就是我们前面说的 x=k 这条垂直线往右的移动速度,为了达到从 0 到 maxPermits 花费 warmupPeriodMicros 的时间,我们将其定义为:

1
2
3
4
@Override
double coolDownIntervalMicros() {
return warmupPeriodMicros / maxPermits;
}

贴一下代码,大家就知道了,在 resync 中用到的这个:

1
2
3
4
5
6
7
8
void resync(long nowMicros) {
if (nowMicros > nextFreeTicketMicros) {
// coolDownIntervalMicros 在这里使用
double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
storedPermits = min(maxPermits, storedPermits + newPermits);
nextFreeTicketMicros = nowMicros;
}
}

基于上面的分析,我们来看 SmoothWarmingUp 的其他源码。

首先,我们来看它的 doSetRate 方法,有了前面的介绍,这个方法的源码非常简单:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
double oldMaxPermits = maxPermits;
// coldFactor 是固定的 3
double coldIntervalMicros = stableIntervalMicros * coldFactor;
// 这个公式我们上面已经说了
thresholdPermits = 0.5 * warmupPeriodMicros / stableIntervalMicros;
// 这个公式我们上面也已经说了
maxPermits =
thresholdPermits + 2.0 * warmupPeriodMicros / (stableIntervalMicros + coldIntervalMicros);
// 计算那条斜线的斜率。数学知识,对边 / 临边
slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits);
if (oldMaxPermits == Double.POSITIVE_INFINITY) {
// if we don't special-case this, we would get storedPermits == NaN, below
storedPermits = 0.0;
} else {
storedPermits =
(oldMaxPermits == 0.0)
? maxPermits // initial state is cold
: storedPermits * maxPermits / oldMaxPermits;
}
}

setRate 方法非常简单,接下来,我们要分析的是 storedPermitsToWaitTime 方法,我们回顾一下下面的代码:

4

这段代码是 acquire 方法的核心,waitMicros 由两部分组成,一部分是从 storedPermits 中获取花费的时间,一部分是等待 freshPermits 产生花费的时间。在 SmoothBursty 的实现中,从 storedPermits 中获取 permits 直接返回 0,不需要等待。

而在 SmoothWarmingUp 的实现中,由于需要预热,所以从 storedPermits 中取 permits 需要花费一定的时间,其实就是要计算下图中,阴影部分的面积。

5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Override
long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
double availablePermitsAboveThreshold = storedPermits - thresholdPermits;
long micros = 0;
// 如果右边梯形部分有 permits,那么先从右边部分获取permits,计算梯形部分的阴影部分的面积
if (availablePermitsAboveThreshold > 0.0) {
// 从右边部分获取的 permits 数量
double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake);
// 梯形面积公式:(上底+下底)*高/2
double length =
permitsToTime(availablePermitsAboveThreshold)
+ permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake);
micros = (long) (permitsAboveThresholdToTake * length / 2.0);
permitsToTake -= permitsAboveThresholdToTake;
}
// 加上 长方形部分的阴影面积
micros += (long) (stableIntervalMicros * permitsToTake);
return micros;
}

// 对于给定的 x 值,计算 y 值
private double permitsToTime(double permits) {
return stableIntervalMicros + permits * slope;
}

到这里,SmoothWarmingUp 也已经说完了。

如果大家对于 Guava RateLimiter 还有什么疑惑,欢迎在留言区留言,对于 Sentinel 中的流控不感兴趣的读者,看到这里就可以结束了。

Sentinel 中的流控

Sentinel 是阿里开源的流控、熔断工具,这里不做过多的介绍,感兴趣的读者请自行了解。

在 Sentinel 的流控中,我们可以配置流控规则,主要是控制 QPS 和并发线程数,这里我们不讨论控制线程数,控制线程数的代码不在我们这里的讨论范围内,下面的介绍都是指控制 QPS。

RateLimiterController

RateLimiterController 非常简单,它通过使用 latestPassedTime