Guava RateLimiter 限流原理详解-每日运维

1.前言

RateLimiter 是一个常用的单机限流工具，在网上也不乏相关的文章，笔者最近在项目里有一个单机限流的需求，使用了此组件，使用之余对它的原理也很感兴趣，于是去看了它的源代码实现，在这里进行一个记录和总结。这里的 guava 版本为 18.0。

2.主要结构

类的继承关系

public abstract class RateLimiter {
    ...
}

abstract class SmoothRateLimiter extends RateLimiter {
    ...
}

static final class SmoothBursty extends SmoothRateLimiter {
    ...
}
static final class SmoothWarmingUp extends SmoothRateLimiter {
    ...
}

以上是 RateLimiter类的一系列继承关系。

主要属性

// RateLimiter 类的属性

// 计时器，RateLimiter 对象创建时初始化，每个请求进来，依靠此属性得到相对时间
private final SleepingStopwatch stopwatch;
// 一个单例对象，进入 Synchronized 代码块时锁住的对象
private volatile Object mutexDoNotUseDirectly;

// SmoothRateLimiter 类的属性
// 存储的令牌数
double storedPermits;
// 允许的最大令牌数
double maxPermits;
// 每生成一个令牌所需的毫秒数
double stableIntervalMicros;
// 可以获得令牌的时间
private long nextFreeTicketMicros;

// SmoothBursty 类的属性
// 最大允许突发的秒数
final double maxBurstSeconds;

// SmoothWarmingUp 类的属性
// 热身时间
private final long warmupPeriodMicros;
// 梯形的斜率
private double slope;
// 令牌数量的阈值/拐点
private double halfPermits;

以上列出了类中的属性，在注释里已经有解释，SmoothWarmingUp 的属性比较难以理解，下面会详细解释。下面这个类图可以直观的展示这一系列属性和继承关系：

下面对 SmoothRateLimiter 的两个实现类的限流原理进行阐述。

3.SmoothBursty

SmoothBursty 类是一个有应对突发能力的令牌桶限流的实现，下面从源码级别看一下它是怎么做的。

限流原理

SmoothBursty 用来限流其实主要就是 acquire 方法，下面来看一下。

acquire 方法

下面是调用 acquire 方法的简化流程图：

其实流程上，就是这么简单，只不过得到 microsToWait 的过程比较复杂，下面具体看一下。

@CanIgnoreReturnValue
public double acquire() {
    return this.acquire(1);
}

@CanIgnoreReturnValue
public double acquire(int permits) {
// 得到需要等待的毫秒数
long microsToWait = this.reserve(permits);
// 线程休眠
this.stopwatch.sleepMicrosUninterruptibly(microsToWait);
return 1.0 * (double)microsToWait / (double)TimeUnit.SECONDS.toMicros(1L);
}

final long reserve(int permits) {
    checkPermits(permits);
    synchronized(this.mutex()) {
        return this.reserveAndGetWaitLength(permits, this.stopwatch.readMicros());
    }
}

acquire 方法会先调用 reserve 方法来得到需要等待的毫秒，接着调用 stopwatch 的方法进行线程的休眠（这里就不深入 stopwatch 的源码了，不是重点），那么看一下 reserve 方法，先调用 checkPermits 判断你要拿的令牌数是否合理，它的代码如下：

private static void checkPermits(int permits) {
Preconditions.checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
}

其实就是判断令牌数是否大于 0。接着进入 synchronized 代码块，锁的是 this.mutex() 这个对象，看一下它是啥：

private Object mutex() {
    Object mutex = this.mutexDoNotUseDirectly;
    if (mutex == null) {
        synchronized(this) {
            mutex = this.mutexDoNotUseDirectly;
            if (mutex == null) {
                this.mutexDoNotUseDirectly = mutex = new Object();
            }
        }
    }

    return mutex;
}

可以看到，其实就是一开始说的 mutexDoNotUseDirectly 对象，这里是一个典型的双段检查的单例写法。然后这里用 synchronized 来控制并发，我觉得也是合理的，毕竟可能剩下一个令牌，但是两个线程都进来了，这时候肯定得控制下并发的。接着它又调用 reserveAndGetWaitLength 方法，点进去看一下：

final long reserveAndGetWaitLength(int permits, long nowMicros) {
    long momentAvailable = this.reserveEarliestAvailable(permits, nowMicros);
    return Math.max(momentAvailable - nowMicros, 0L);
}

进来之后发现，又是调别的函数。。。不过通过它的返回值，还是可以得到点信息的，它的返回语句为：**return Math.max(momentAvailable - nowMicros, 0L); **先是调某方法（之后再说）得到 momenAvailable，这个变量一看名字，就知道它是表示可以获得令牌的时间，然后这个表达式的意思其实就是如果 momenAvailable 小于等于请求进来的时刻，那肯定是可以拿令牌的，返回 0；否则，返回从现在到可以拿令牌，需要等的时间。这里可能有人有疑问了，momenAvailable 怎么会大于 nowMicros 呢？带着这个疑惑，点进 reserveEarliestAvailable 看一下：

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
    // 更新令牌数，nextFreeTicketMicros
    this.resync(nowMicros);
    // 得到当前的 nextFreeTicketMicros
    long returnValue = this.nextFreeTicketMicros;
    // 计算目前可以提供的令牌数
    double storedPermitsToSpend = Math.min((double)requiredPermits, this.storedPermits);
    // 计算需要新提供多少令牌
    double freshPermits = (double)requiredPermits - storedPermitsToSpend;
    // 计算产生这些新提供的令牌，需要多少时间
    long waitMicros = this.storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) + (long)(freshPermits * this.stableIntervalMicros);
    // 将 nextFreeTicketMicros 向后推
    this.nextFreeTicketMicros = LongMath.saturatedAdd(this.nextFreeTicketMicros, waitMicros);
    // 扣减令牌数
    this.storedPermits -= storedPermitsToSpend;

    return returnValue;
}

这里是 SmoothBursty 的核心代码所在，进来先是一个 resync 操作，点进去看一下：

void resync(long nowMicros) {
if (nowMicros > this.nextFreeTicketMicros) {
    // 距离上一个 nextFreeTicketMicros 到现在，生成的令牌数
    double newPermits = (double)(nowMicros - this.nextFreeTicketMicros) / this.coolDownIntervalMicros();
    // 令牌最多只能有 maxPermits 个
    this.storedPermits = Math.min(this.maxPermits, this.storedPermits + newPermits);
    this.nextFreeTicketMicros = nowMicros;
}
}

当 nowMicros 大于 nextFreeTicketMicros 时，才会触发，这也很合理，因为这里面的操作主要是更新 storedPermits 和 nextFreeTicketMicros，nowMicros 大于 nextFreeTicketMicros 时，说明离上次获取令牌又过了一段时间了，这段时间又生成了一些令牌，所以需要加上这些令牌。 reserveEarliestAvailable 剩下的代码就比较简单了，就是算一下这次请求，需要从我的令牌存量里拿多少，最多只能拿 storedPermits 个，这也很合理，再多我也没有了啊。然后会算一下，超出存量的令牌数，以及产生这些令牌需要的时间 waitMicros，把 nextFreeTicketMicros 往后推 waitMicros。看了上面这些东西，可能还是不太理解它的工作机理，在这里细化一下。

正常的限流流程

这里先不考虑 nextFreeTicketMicros 向后推的情况，先看正常情况。比方说，现在限制每秒的 qps 为200，然后调用 create(double permitsPerSecond) 来创建一个 RateLimiter 对象，这个创建的代码我就不说了，总之，创建完成后，stableIntervalMicros 会更新为 5，maxPermits 会更新为 200，storedPermits 什么的也会被更新。一秒后，进来一个请求，acquire(1)，那么经过一个很长的调用链，它会调用 resync(nowMicros)，由于现在是正常情况，所以上一次的 nextFreeTicketMicros 也是通过调用 resync 设置的（在创建过程中也会调用 resync），在创建过程中，已经更新了一次 storedPermits，现在进来了一个请求，会算一下这期间产生的令牌，并将 nextFreeTicketMicros 更新为 nowMicros，只要这个流量比较小，每次进来都会把 nextFreeTicketMicros 更新为 nowMicros，再累加 storedPermits，这是没问题的。这些情况下，reserveEarliestAvailable 返回的值都是它们自己的 nowMicros(因为 nextFreeTicketMicros 就是按 nowMicros 设置的），在 reserveAndGetWaitLength 处，会返回 0，最上面的 microsToWait 也为 0，不需要休眠。

nextFreeTicketMicros 向后推的情况

假如某次请求为 acquire(600)，经过 resync，此时的 nextFreeTicketMicros 为 2000，storedPermits 为200，那这个存货不够啊，nextFreeTicketMicros 往后推 2000，变为了 4000。这里回过头看 reserveEarliestAvailable 这个方法，它是先用 **long returnValue = this.nextFreeTicketMicros; **存了一下旧的 nextFreeTicketMicros，然后再计算 nextFreeTicketMicros 要不要往后推，最后返回的是旧的 nextFreeTicketMicros。在这里，就可以明白 SmoothBursty 里的 bursty(突发）的意思了。这次拿 600 个令牌，即使存货不够，也可以拿到，而且 reserveEarliestAvailable 返回的是旧的 nextFreeTicketMicros，并不是往后推了的 nextFreeTicketMicros，这也很好理解，毕竟要支持突发嘛，nextFreeTicketMicros 往后推是我拿了令牌之后的结果，与我无关，我返回旧的 nextFreeTicketMicros 就行了，毕竟我调 reserveEarliestAvailable，要获得的是 momentAvailable，我可以拿令牌的时刻。这里也可以理解 SmoothBursty 是如何限制流量的。acquire(600) 这个请求成功了，那么假如 3000 时刻，来了一个 acquire(200) 进来，这时候 resync 并不会累计令牌，直接到 reserveEarliestAvailable 的下面的代码，会把 nextFreeTicketMicros 再往后推 1000，变为 5000，在 reserveAndGetWaitLength 方法会返回 1000，表示此线程要休眠 1000 毫秒（很合理，你进来时，nextFreeTicketMicros 为 4000，你是 3000 时进来的，当然等 1000 了）。在 2000 到 4000 这段时间里，任何进来的请求都会休眠，将 nextFreeTicketMicros 不断后推，那你从 2000 到 4000 这个大的时间跨度来看，qps 也是符合的（2000 时，预支了 400 个令牌，接下来的 2 秒都不会有请求通过，这两秒的令牌消耗量为 400，确实控制了 qps 为 200）。极端点想，你也可以一下子 acquire(20000)，这会把 nextFreeTicketMicros 往后推 100 秒，虽然瞬间流量为 20000，但是在 100 秒这个时间跨度内，qps 还是符合的。这样就既支持了突发，又支持了这段时间的限流。

tryAcquire 方法

看完了 acquire 方法，对于 SmoothBursty 限流已经理解了，但是有个问题，如果真的不幸遇到了前一个请求 acquire(20000) 这种情况，那我这个请求岂不是要休眠 100 秒，确实控制了流量，但这对我的业务肯定是不合理的，所以这里还提供了 tryAcquire 方法，下面是它的简化的流程图：

看一下它具体的代码：

public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
    long timeoutMicros = Math.max(unit.toMicros(timeout), 0L);
    checkPermits(permits);
    long microsToWait;
    synchronized(this.mutex()) {
        long nowMicros = this.stopwatch.readMicros();
        // 规定时间拿不到令牌，直接返回 false
        if (!this.canAcquire(nowMicros, timeoutMicros)) {
            return false;
        }

        microsToWait = this.reserveAndGetWaitLength(permits, nowMicros);
    }

    this.stopwatch.sleepMicrosUninterruptibly(microsToWait);
    return true;
}
// 判断规定时间内能不能拿到令牌
private boolean canAcquire(long nowMicros, long timeoutMicros) {
    return this.queryEarliestAvailable(nowMicros) - timeoutMicros  0.0) {
        // 当前的需求令牌数，梯形部分可以提供多少
        double permitsAboveHalfToTake = Math.min(availablePermitsAboveHalf, permitsToTake);
        // 梯形面积代表总时间
        micros = (long)(permitsAboveHalfToTake * (this.permitsToTime(availablePermitsAboveHalf) + this.permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);
        // 矩形部分需要提供的令牌数量
        permitsToTake -= permitsAboveHalfToTake;
    }

    micros = (long)((double)micros + this.stableIntervalMicros * permitsToTake);
    return micros;
        }

// 计算传入的相对横坐标对应的纵坐标也就是等待时间
private double permitsToTime(double permits) {
            return this.stableIntervalMicros + permits * this.slope;
        }

可以看到，在 SmoothBursty 里这部分等待时间直接为 0，而SmoothWarmingUp 里却很长，下面解释一下这个代码

其实 SmoothWarmingUp 的思想，看这个图就可以了，它的思想就是，如果你系统存的令牌也就是 storedPermits 越多，那么你的系统就越冷，那么你获取令牌的等待时间就应该更长，这个图横坐标就是系统存的令牌，纵坐标就是获取一个令牌需要额外等的时间。存货小于 halfPermits，额外的等待时间就是固定的 stableInterval 了，大于 halfPermits 时，随着存货越来越少，额外等待的时间也越来越少预热的关键属性就是 warmupPeriodMicros，这个越长，Warm up Period 就越大，额外的等待时间就越长

小结

SmoothWarmingUp 和 SmoothBursty 限流原理基本一致，只是在获取令牌时需要一些额外时间以达到预热的效果

5.总结

本文从源码角度分析了 SmoothBursty 和SmoothWarimingUp 是如何限流的，它们本质上是令牌桶的实现，通过 nextFreeTicketMicros 的设置，不仅支持突发流量，同时也可以保证突发流量之后的一段时间内的平均 qps 符合要求。根据每个类的目标不同，实现上也略有不同，如果有理解的不对的地方，友好交流。

Guava RateLimiter 限流原理详解

类的继承关系

主要属性

限流原理

acquire 方法

正常的限流流程

nextFreeTicketMicros 向后推的情况

tryAcquire 方法

小结

相关文章

发布评论取消回复

泡泡

百度搜索：蓝易云 – Docker的安装与卸载(Centos、Ubuntu )

PostgreSQL系数据库使用COPY导数时如何实现增量及重复数据更新导入

【OceanBase DBA早下班系列】—— obdiag 收集的 SQL Monitor Report 如何解读

Guava RateLimiter 限流原理详解

类的继承关系

主要属性

限流原理

acquire 方法

正常的限流流程

nextFreeTicketMicros 向后推的情况

tryAcquire 方法

小结

相关文章

发布评论 取消回复

泡泡

百度搜索：蓝易云 – Docker的安装与卸载(Centos、Ubuntu )

PostgreSQL系数据库使用COPY导数时如何实现增量及重复数据更新导入

【OceanBase DBA早下班系列】—— obdiag 收集的 SQL Monitor Report 如何解读

发布评论取消回复