源码分析 RateLimiter SmoothBursty 实现原理(文末附流程图)

丁威

发布于 2020-03-31 12:05:18

1.4K0

发布于 2020-03-31 12:05:18

文章被收录于专栏：中间件兴趣圈

上篇详细介绍了Sentinel FlowSlot 限流实现原理(文末附流程图与总结)的限流实现机制，但主要介绍的策略限流的快速失败机制，在Sentinel 中除了快速失败，还提供了匀速排队，预热等限流策略，但我发现 Sentinel 的匀速排队、预热机制是基于 guava 的 RateLimiter，为了更加彻底的理解 Sentienl 限流相关的内容，从本文开始先来学习一下 RateLimiter 的相关实现原理。

温馨提示：文章的末尾会总结 SmoothBursty 的核心流程图与实现原理，本文将展示笔者是如何一步一步揭晓其实现原理的方法。

1、RateLimiter 类设计图

RateLimiter 限流抽象类，定义限流器的基本接口。
SmoothRateLimiter 平滑限流实现器，也是一个抽象类。
SmoothWarmingUp 自带预热机制的限流器实现类型。
SmoothBursty 适应于突发流量的限流器。

上述类这些属性，在讲解 SmoothBursty、SmoothWarmingUp 时再详细介绍。

温馨提示：可以看看这些类上的注释，先初步了解其设计思想。

2、寻找入口

我们首先从 guava 的测试用例中尝试寻找一下 RateLimiter 的测试类，测试代码如下。

public void testSimple() {
    RateLimiter limiter = RateLimiter.create(stopwatch, 5.0);
    limiter.acquire(); // R0.00, since it's the first request
    limiter.acquire(); // R0.20
    limiter.acquire(); // R0.20
    assertEvents("R0.00", "R0.20", "R0.20");
}

从这里基本可以看出，首先通过 RateLimiter.create 的静态方法创建一个限流器，然后应用程序在执行业务逻辑之前先调研限流器的 acquire 方法申请许可，接下来我们将循着这个流程来探讨其实现思路。

3、探究 SmoothBursty 实现原理

3.1 SmoothBursty 创建流程

从上面的示例来看，应用程序首先通过 RateLimiter 的静态方法创建一个限流器，其代码如下：

RateLimiter#create

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

代码@1：首先先介绍方法的参数：

SleepingStopwatch stopwatch 秒表，主要是实现当前从启动开始已消耗的时间，有点类似计算一个操作耗时，实现精度纳秒。
double permitsPerSecond 每秒的许可数，即通常我们说的限流TPS。

代码@2：创建 SmoothBursty 对象。

代码@3：调用 setRate API 设置其速率器。

接下来我们对其进行展开。

3.1.1 SmoothBursty 构造函数

SmoothBursty 构造函数

SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {
    super(stopwatch);
    this.maxBurstSeconds = maxBurstSeconds;
}

这里主要是为 stopWatch 与 maxBurstSeconds 赋值，其中 maxBurstSeconds 为允许的突发流量的时间，这里默认为 1.0，表示一秒，会影响最大可存储的许可数。

3.1.2 RateLimiter setRate 方法详解

RateLimiter#setRate

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

代码@1：该方法需要获取该类的监视器，在同步代码块中执行，实现线程安全性。

代码@2：调用 doSetRate 设置速率，将调用其具体实现类 SmoothRateLimiter 的 doSetRate 方法。

SmoothRateLimiter#doSetRate

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

代码@1：先来介绍一下该方法的参数的含义：

double permitsPerSecond 每秒的许可数，即TPS。
long nowMicros 系统已运行时间。

代码@2：基于当前时间重置 SmoothRateLimiter 内部的 storedPermits (已存储的许可数量) 与 nextFreeTicketMicros (下一次可以免费获取许可的时间) 值，所谓的免费指的是无需等待就可以获取设定速率的许可，该方法对理解限流许可的产生非常关键，稍后详细介绍。

代码@3：根据 TPS 算出一个稳定的获取1个许可的时间。以一秒发放5个许可，即限速为5TPS，那发放一个许可的世界间隔为 200ms，stableIntervalMicros 变量是以微秒为单位。

代码@4：调用 SmoothRateLimiter 的抽象方法 doSetRate 设置速率，这里会调用 SmoothBursty 的 doSetRate 方法。

在介绍 SmoothBursty 的 doSetRate 方法之前，我们先来看看 resync 方法的实现细节。

SmoothRateLimiter#resync

void resync(long nowMicros) {
    if (nowMicros > nextFreeTicketMicros) {  // @1 
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();  // @2
      storedPermits = min(maxPermits, storedPermits + newPermits);    // @3
      nextFreeTicketMicros = nowMicros;   // @4
    }
}

代码@1：如果当前已启动时间大于 nextFreeTicketMicros（下一次可以免费获取许可的时间），则需要重新计算许可，即又可以向许可池中添加许可。

代码@2：根据当前时间可增加的许可数量，在 SmoothBursty 的 coolDownIntervalMicros 方法返回的就是上文提到的 stableIntervalMicros (发放一个许可所需要的时间)，故本次可以增加的许可数的算法也好理解，即用当前时间戳减去 nextFreeTicketMicros 的差值，再除以发送一个许可所需要的时间即可。

代码@3：计算当前可用的许可。

代码@4：更新下一次可增加计算许可的时间。

接下来再继续看 SmoothBursty 的 doSetRate 方法。

SmoothBursty#doSetRate

void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
    double oldMaxPermits = this.maxPermits;
    maxPermits = maxBurstSeconds * permitsPerSecond;
    if (oldMaxPermits == Double.POSITIVE_INFINITY) {
        storedPermits = maxPermits;
    } else {
        storedPermits =
            (oldMaxPermits == 0.0)
                ? 0.0 // initial state
                : storedPermits * maxPermits / oldMaxPermits;
    }
}

这里主要是初始化 storedPermits 的值，该限速器支持在运行过程中动态改变 permitsPerSecond 的值。

3.2 SmoothBursty acquire 工作流程

RateLimiter 中的 acquire 方法如下：

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

代码@1：根据当前剩余的许可与本次申请的许可来判断本次申请需要等待的时长，如果返回0则表示无需等待。

代码@2：如果需要等待的时间不为0，表示触发限速，睡眠指定时间后唤醒。

代码@3：返回本次申请等待的时长。

接下来重点介绍 reserve 方法的实现原理。

RateLimiter#reserve

final long reserve(int permits) {
    checkPermits(permits);
    synchronized (mutex()) {  // @1
      return reserveAndGetWaitLength(permits, stopwatch.readMicros()); // @2
    }
}

代码@1：限速器主要维护的重要数据字段( storedPermits )，对其进行维护时都需要先获取锁。

代码@2：调用内部方法 reserveAndGetWaitLength 来计算需要等待时间。

继续跟踪 reserveAndGetWaitLength 方法。

final long reserveAndGetWaitLength(int permits, long nowMicros) {
    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);   // @1
    return max(momentAvailable - nowMicros, 0);  // @2
}

代码@1：根据当前拥有的许可数量、当前时间判断待申请许可最早能得到满足的最早时间，用momentAvailable 表示。

代码@2：然后计算 momentAvailable 与 nowMicros 的差值与0做比较，得出需要等待的时间。

继续跟踪 reserveEarliestAvailable方法，该方法在 RateLimiter 中一个抽象方法，具体实现在其子类 SmoothRateLimiter 中。

SmoothRateLimiter#reserveEarliestAvailable

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

    this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);  // @5
    this.storedPermits -= storedPermitsToSpend;    // @6
    return returnValue;
}

代码@1：在尝试申请许可之前，先根据当前时间即发放许可速率更新 storedPermits 与 nextFreeTicketMicros（下一次可以免费获取许可的时间）。

代码@2：计算本次能从 storedPermits 中消耗的许可数量，取需要申请的许可数量与当前可用的许可数量的最小值，用 storedPermitsToSpend 表示。

代码@3：如果需要申请的许可数量( requiredPermits )大于当前剩余许可数量( storedPermits )，则还需要等待新的许可生成，用 freshPermits 表示，即如果该值大于0，则表示本次申请需要阻塞一定时间。

代码@4：计算本次申请需要等待的时间，storedPermitsToWaitTime 方法在 SmoothBursty 的实现中默认返回 0，即 SmoothBursty 的等待时间主要来自按照速率生成 freshPermits 个许可的时间，生成一个许可的时间为 stableIntervalMicros，故需要等待的时长为 freshPermits * stableIntervalMicros。

代码@5：更新 nextFreeTicketMicros 为当前时间加上需要等待的时间。

代码@6：更新 storedPermits 的值，即减少本次已消耗的许可数量。

代码@7：请注意这里返回的 returnValue 的值，并没有包含由于剩余许可需要等待创建新许可的时间，即允许一定的突发流量，故本次计算需要的等待时间将对下一次请求生效，这也是框架作者将该限速器取名为 SmoothBursty 的缘由。

SmoothBursty 的 acquire 方法就介绍到这里了。

4、总结

由于源码分析会显得枯燥与不直观，我们先给出如下流程图：

SmoothBursty 的核心设计思想基本与令牌桶类似，但还是有些不同。基本思想：

SmoothBursty 以指定的速率生成许可，在 SmoothBursty 中用 storedPermits 表示。
当一个请求需要申请许可时，如果需要申请的许可数小于 storedPermits ，则消耗指定许可，直接返回，无需等待。
当一个请求需要申请的许可大于 storedPermits 时，则计算需要等待的时间，更新下一次许可可发放时间，直接返回，即当请求消耗掉所有许可后，当前请求并不会阻塞，而是影响下一个请求，即支持突发流量。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-03-22，如有侵权请联系 cloudcommunity@tencent.com 删除

double