Guava RateLimiter详解以及源码分析

Posted 2021-08-31 犀牛饲养员

文章目录

• • Guava RateLimiter详解以及源码分析
  • • RateLimiter使用场景介绍
    • RateLimiter使用示例
    • RateLimiter原理及源码解读
    • • SmoothBursty
      • SmoothWarmingUp

Guava RateLimiter详解以及源码分析

RateLimiter使用场景介绍

首先你需要明白限流的概念，在高并发、高流量的场景中，我们的系统有时候会通过限流的手段来防止自己的系统被外部的流量打挂，是一种自我保护措施。

有人可能会问，现在硬件资源成本越来越低，我难道不可以通过扩容的手段增加系统的承受能力吗？事实上任何系统的性能都有一个上限，当并发量超过这个上限之后，可能会对系统造成毁灭性地打击。因此在任何时刻我们都必须保证系统的并发请求数量不能超过某个阈值，限流就是为了完成这一目的。

有人可能接着问，什么时候我的系统需要限流，限制多少QPS呢？对不起，这个没有固定答案，通常需要根据业务场景并结合压力测试对系统做一个评估，然后给出合理的限流方案。

Guava RateLimiter是一个谷歌提供的限流工具，可以有效限定单个JVM实例上某个接口的流量。（单个JVM实例这个划重点）

有人可能又问了，Ratelimiter和一些主流的限流组件比如Sentinel有什么区别呢？

我个人觉得，Sentinel是个限流组件，它功能更加丰富，可以提供分布式环境下的限流方案。并且Sentinel还具有丰富的可视化管理工具，支持灵活的配置。

而Ratelimiter是个轻量级的限流工具，提供的是单机场景下的限流方案。不过需要说明的是，Sentinel在实现上参考了部分Ratelimiter的思路。

RateLimiter使用示例

首先引用guava的依赖，

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>19.0</version>
        </dependency>

然后我们封装一个简单限流服务，

public class AccessLimitService {

    //每秒发出10个令牌，也就是每100ms会向令牌桶中放置一个令牌
    RateLimiter rateLimiter = RateLimiter.create(10.0);

    /**
     * 尝试获取令牌
     * @return
     */
    public boolean tryAcquire(){
        return rateLimiter.tryAcquire();
    }

}

写个测试类，

public class AccessLimitTest {

    private static final AccessLimitService accessLimitService = new AccessLimitService();
    private static ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws InterruptedException {

        AccessLimitTest test = new AccessLimitTest();
        for (int i = 0; i < 10; i++) {
            fixedThreadPool.submit(() -> test.access());
        }

        fixedThreadPool.shutdown();
        fixedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);

    }

    private void access() {
        if (accessLimitService.tryAcquire()) {
            System.out.println("我被访问了,访问我的是：" + Thread.currentThread().getName());
            sleepUninterruptibly(20, MILLISECONDS);//模拟执行耗时
        } else {
            System.out.println("我被限制访问了,我是：" + Thread.currentThread().getName());
        }
    }
}

运行的结果，

我被访问了,访问我的是：pool-1-thread-1
我被限制访问了,我是：pool-1-thread-2
我被限制访问了,我是：pool-1-thread-4
我被限制访问了,我是：pool-1-thread-3
我被限制访问了,我是：pool-1-thread-8
我被限制访问了,我是：pool-1-thread-6
我被限制访问了,我是：pool-1-thread-5
我被限制访问了,我是：pool-1-thread-9
我被访问了,访问我的是：pool-1-thread-10
我被限制访问了,我是：pool-1-thread-7

除了上面使用的tryAcquire方法请求访问资源，RateLimiter还有一些其它的方法有类似的功能：

tryAcquire(int permits)

从RateLimiter 获取许可数，如果该许可数可以在无延迟下的情况下立即获取得到的话

tryAcquire(int permits, long timeout, TimeUnit unit)

从RateLimiter 获取指定许可数如果该许可数可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可数的话，那么立即返回false （无需等待）

tryAcquire(long timeout, TimeUnit unit)

从RateLimiter 获取许可如果该许可可以在不超过timeout的时间内获取得到的话，或者如果无法在timeout 过期之前获取得到许可的话，那么立即返回false（无需等待）

acquire()

从RateLimiter获取一个许可，该方法会被阻塞直到获取到请求

acquire(int permits)

从RateLimiter获取指定许可数，该方法会被阻塞直到获取到请求

总结下，acquire和tryAcquire在于前者是阻塞一直等到获取许可，而后者不管是否拿到都立即返回（或者超时后返回）。

可能有人有疑问，tryAcquire(int permits)使用的场景是啥，也就是什么情况下需要获取多个许可呢。我理解这是应对突发流量的时候，比如我的业务场景我有突发流量，可以提前预支多个令牌。（带来的结果就是后面的请求延后处理）

在实际的项目中，如果我们用到了Ratelimiter，建议把它做在切面里，然后需要限流的方法加上注解就可以了。

RateLimiter原理及源码解读

RateLimiter是基于令牌桶算法实现的限流方案，如下图所示：

解释下：

• 令牌桶是按照固定速率往桶中添加令牌，请求是否被接受需要看桶中令牌是否足够，当令牌数减为零时则拒绝新的请求。

• 令牌桶限制的是平均流入速率，允许突发请求，只要有令牌就可以处理，支持一次拿多个令牌。

Guava的RateLimiter基于令牌桶算法，实现了两种限流模式，平滑突发限流(SmoothBursty)和平滑预热限流(SmoothWarmingUp)。两者的区别是前者是稳定模式，令牌生成速度恒定。而SmoothWarmingUp是渐进模式，令牌生成速度缓慢提升直到维持在一个稳定值。

从类图结构也可以窥探一二，

RateLimiter 是一个抽象类，SmoothRateLimiter 继承自 RateLimiter，也是一个抽象类，SmoothBursty 和 SmoothWarmingUp 是具体的实现类。

下面我们就来分别看下两个具体实现类的源码。

RateLimiter提供了两个静态方法，分别用来创建两个实现类的实例。

  @VisibleForTesting
  static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
  }
  
  @VisibleForTesting
  static RateLimiter create(
      double permitsPerSecond,
      long warmupPeriod,
      TimeUnit unit,
      double coldFactor,
      SleepingStopwatch stopwatch) {
    RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
  }

SmoothBursty

先通过一个单元测试，看看SmoothBursty的效果，

public static void testSmoothBursty() {
        RateLimiter r = RateLimiter.create(5);
        while (true) {
            System.out.println("get 1 tokens: " + r.acquire() + "s");
        }
    }

输出的结果，

get 1 tokens: 0.0s
get 1 tokens: 0.197195s
get 1 tokens: 0.194382s
get 1 tokens: 0.196544s
get 1 tokens: 0.198446s
get 1 tokens: 0.198093s
get 1 tokens: 0.199798s

前面已经提到了acquire方法回阻塞一直到拿到令牌，返回的是所花的时间。所以从这个输出结果看出，基本上都是0.2s执行一次，符合代码中一秒发放5个令牌的设定。

再来看一个测试用例，

public static void testSmoothBursty2() {
        RateLimiter r = RateLimiter.create(2);
        while (true)
        {
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            try {
                Thread.sleep(2000);
            } catch (Exception e) {}
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("end");
        }
    }

输出时这样的，

get 1 tokens: 0.0s
get 1 tokens: 0.0s
get 1 tokens: 0.0s
get 1 tokens: 0.0s
end
get 1 tokens: 0.49959s
get 1 tokens: 0.0s
get 1 tokens: 0.0s
get 1 tokens: 0.0s

这个结果说明什么呢？首先第一次马上拿到令牌这个没啥问题，然后sleep 2秒，这段时间会发放4个令牌（500ms发一个，第一个已经被拿了），后面三个可以马上获取说明RateLimiter会进行令牌的累积，如果获取令牌的频率比较低，则不会导致等待，直接获取令牌。

再来一个示例，

public static void testSmoothBursty3() {
        RateLimiter r = RateLimiter.create(5);
        while (true)
        {
            System.out.println("get 5 tokens: " + r.acquire(5) + "s");
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("get 1 tokens: " + r.acquire(1) + "s");
            System.out.println("end");
        }
    }

输出，

get 5 tokens: 0.0s
get 1 tokens: 0.997078s
get 1 tokens: 0.197288s
get 1 tokens: 0.199405s
end
get 5 tokens: 0.199609s
get 1 tokens: 0.999309s
get 1 tokens: 0.198634s
get 1 tokens: 0.1944s
end

这个测试用例就是前面提到的应对突发流量的情况，RateLimiter由于会累积令牌，所以可以应对突发流量。示例代码中，d第一个请求5个令牌会马上拿到，但是后面的请求就要付出"代价"，RateLimiter在没有足够令牌发放时，采用滞后处理的方式，也就是前一个请求获取令牌所需等待的时间由下一次请求来承受，也就是代替前一个请求进行等待。

接着看下具体的源码实现，我尽量写在注释里，比较简单的就不详述了。

static final class SmoothBursty extends SmoothRateLimiter {
    /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */
    //在ReteLimiter未使用时，最多保存几秒的令牌，默认是1
    final double maxBurstSeconds;

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

SmoothBursty 的 maxBurstSeconds 构造函数参数主要用于计算 maxPermits：

maxPermits = maxBurstSeconds * permitsPerSecond;。

该参数的作用在于，可以更为灵活地控制流量。比如某些接口限制为500次/20秒等。设置并不局限于qps。

上面的工厂方法(create)里都调用了setRate方法，RateLimiter 中 setRate 方法最终后调用 doSetRate 方法，doSetRate 是一个抽象方法，SmoothRateLimiter 抽象类中覆写了 doSetRate 方法：

@Override
  final void doSetRate(double permitsPerSecond, long nowMicros) {
    resync(nowMicros);
    /**
     * stableIntervalMicros 等于 1/qps，它代表系统在稳定期间，两次请求之间间隔的微秒数。
     * 例如：如果设置的 qps 为5，则 stableIntervalMicros 为200ms。
     */
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);
  }

resync方法也比较重要，这里说明下。

/** 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;
    }
  }

其中，nextFreeTicketMicros这个变量的意义是：

/**
   * The time when the next request (no matter its size) will be granted. After granting a request,
   * this is pushed further in the future. Large requests push this further than small requests.
   * 下一次请求可以获取令牌的起始时间
   * RateLimiter允许预消费，上次请求预消费令牌后，下次请求需要等待相应的时间到nextFreeTicketMicros时刻才可以获取令牌
   */
  private long nextFreeTicketMicros = 0L; // could be either in the past or future

resync 方法就是RateLimiter中惰性计算的实现。每一次请求来的时候（acquire和tryAcquire方法），都会调用到这个方法。其实现思路为，若当前时间晚于nextFreeTicketMicros，则计算该段时间内可以生成多少令牌，将生成的令牌加入令牌桶中并更新数据。这样一来，只需要在获取令牌时计算一次即可。

其它的方案，比如开启一个定时任务，由定时任务持续生成令牌也是可以的，不过比如系统中需要限流的接口比较多，需要针对每个接口定制限流方案，那对应就要多个定时任务，比较消耗资源。这也是RateLimiter的一个设计亮点。

具备了以上的背景知识后，再来继续看Ratelimiter暴露的接口。

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


/**
   * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request
   * can be granted. Tells the amount of time slept, if any.
   *
   * @param permits the number of permits to acquire
   * @return time spent sleeping to enforce rate, in seconds; 0.0 if not rate-limited
   * @throws IllegalArgumentException if the requested number of permits is negative or zero
   * @since 16.0 (present in 13.0 with {@code void} return type})
   *
   * 获取指定数量的令牌，返回阻塞的时间
   */
  @CanIgnoreReturnValue
  public double acquire(int permits) {
    // 计算获取令牌所需等待的时间
    long microsToWait = reserve(permits);
    //线程sleep等待
    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);
    // 下一次请求可以获取令牌的起始时间
    // RateLimiter允许预消费，上次请求预消费令牌后，下次请求需要等待相应的时间到nextFreeTicketMicros时刻才可以获取令牌
    long returnValue = nextFreeTicketMicros;

    double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
    // freshPermits是需要预先支付的令牌，也就是目标令牌数减去目前即可得到的令牌数
    double freshPermits = requiredPermits - storedPermitsToSpend;
    /*
    突发流量的情况，突然涌入大量请求，而现有令牌数又不够用，因此会预先支付一定的令牌数
    waitMicros是产生预先支付令牌的数量时间，则将下次要添加令牌的时间应该计算时间加上watiMicros
    SmoothBuresty的storedPermitsToWaitTime直接返回0，所以watiMicros就是预先支付的令牌所需等待的时间
     */
    long waitMicros =
        storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
            + (long) (freshPermits * stableIntervalMicros);

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

代码里都有详细的注释，我再简单解释下，acquire函数用于获取permits个令牌，并计算需要等待多长时间，进而挂起等待并将该值返回，主要通过reserve返回需要等待的时间，reserve中通过调用reserveAndGetWaitLength获取等待时间。

tryAcquire类似就不说了。

SmoothWarmingUp

要讲平滑预热限流的代码，必须先上一幅图，这个是源码里的一个截图，

体现这个图的核心代码如下：

//这个方法前面也说了，不同的实现类不一样。对于SmoothWarmingUp等待时间就是计算上图中梯形或者正方形的面积
    @Override
    long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
      //当前存储的permits超出阈值的部分
      double availablePermitsAboveThreshold = storedPermits - thresholdPermits;
      long micros = 0;
      // measuring the integral on the right part of the function (the climbing line)
      //如果当前存储的令牌数超出thresholdPermits
      if (availablePermitsAboveThreshold > 0.0) {
        //在阈值右侧并且需要被消耗的令牌数量
        double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake);
        // TODO(cpovirk): Figure out a good name for this variable.

        /**
         * 梯形的面积
         *
         * 高 * (顶 + 底) / 2
         *
         * 高是 permitsAboveThresholdToTake 也就是右侧需要消费的令牌数
         * 底 较长 permitsToTime(availablePermitsAboveThreshold)
         * 顶 较短 permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake)
         */
        double length =
            permitsToTime(availablePermitsAboveThreshold)
                + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake);
        micros = (long) (permitsAboveThresholdToTake * length / 2.0);
        permitsToTake -= permitsAboveThresholdToTake;
      }
      // measuring the integral on the left part of the function (the horizontal line)
      //平稳时期的面积，正好是长乘宽，等待时间开始平稳
      micros += (long) (stableIntervalMicros * permitsToTake);
      return micros;
    }

/**
     * cool down的间隔时间
     */
    @Override
    double coolDownIntervalMicros() {
      return warmupPeriodMicros / maxPermits;
    }

简单来讲，SmoothWarmingUp的实现，将storedPermits分成两个区间：[0, thresholdPermits) 和 [thresholdPermits, maxPermits]，存在一个"热身"的阶段，thresholdPermits 是系统 stable 阶段和 cold 阶段的临界点。从 thresholdPermits 右边的部分拿走 permit 需要等待的时间更长（面积更大）。

---

参考:

• https://www.alibabacloud.com/blog/detailed-explanation-of-guava-ratelimiters-throttling-mechanism_594820