Guava限流器原理浅析

文章目录

基本知识
- 限流器的类图
- 使用示例
原理解析
- 限流整体流程
- 问题驱动
- - 1、限流器创建的时候会初始化令牌吗？
  - 2、令牌是如何放到桶里的？
  - 3、如果要获取的令牌数大于桶里的令牌数会怎么样
  - 4、令牌数量的更新会有并发问题吗
总结

实际工作中难免有限流的场景。我们熟知的限流算法有计数器限流（固定窗口、滑动窗口）算法、漏桶算法、令牌桶算法等。其具体实现也多种多样，本文就来简单窥探一下Guava的实现。

基本知识

限流器的类图

在这里插入图片描述
RateLimiter：限流器基类，定义限流器的创建、令牌的获取等操作。
SmoothRateLimiter：定义一种平滑的限流器，也是抽象类，继承RateLimiter。
SmoothBursty：普通的平滑限流器实现类，实现SmoothRateLimiter。以稳定的速率生成令牌，则会同时全部被获取到。比如令牌桶现有令牌数为5，这时连续进行10个请求，则前5个请求会全部直接通过，没有等待时间，之后5个请求则每隔200毫秒通过一次。
SmoothWarmingUp：预热的平滑限流器实现类，实现SmoothRateLimiter。随着请求量的增加，令牌生成速率会缓慢提升直到一个稳定的速率。比如令牌桶现有令牌数为5，这时连续进行10个请求，只会让第一个请求直接通过，之后的请求都会有等待时间，等待时间不断缩短，直到稳定在每隔200毫秒通过一次。这样，就会有一个预热的过程。

下文以SmoothBursty为例来分析限流原理。

使用示例

public class RateLimitTest {
    public static void main(String[] args) throws InterruptedException {
    	// 1、创建限流器，一秒内最多允许2个请求通过
        RateLimiter rateLimiter = RateLimiter.create(2);
        serial(rateLimiter);
    }

    private static void serial(RateLimiter rateLimiter) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            String time = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
            // 2、尝试获取令牌，不论是否能获取到都直接返回
            boolean res = rateLimiter.tryAcquire();
            // 获取令牌，如果获取不到就一直等待
			// rateLimiter.acquire();
            if (res) {
                System.out.println(time + ":请求被允许");
            } else {
                System.out.println(time + ":请求被限流");
            }
            Thread.sleep(250);
        }
    }
 }

执行结果：

15:52:08.583:请求被允许
15:52:08.852:请求被限流
15:52:09.108:请求被允许
15:52:09.361:请求被限流
15:52:09.617:请求被允许
15:52:09.872:请求被限流
15:52:10.127:请求被允许
15:52:10.378:请求被限流
15:52:10.629:请求被允许
15:52:10.882:请求被限流

可以看到同一秒内最多只有2个请求被允许。

原理解析

限流整体流程

在这里插入图片描述

创建限流器。此时桶里的令牌数为0。设置QPS=5（每秒最多允许5个请求），这个数字“5”带表了两层含义：
1）桶里最大只能容纳5个令牌。
2）一秒可以生成5个令牌，生成一个令牌需要1/5=0.2秒=200毫秒。
发起请求。此时距离限流器创建已经经过了一秒，桶里应该存在5个令牌，而本次请求需要获取并消耗1个令牌。
更新令牌数量。

上面只是描述了一个大致思路，还有很多细节问题需要考虑，下文就以问题来驱动原理探究。

问题驱动

限流器关键属性解释
SmoothRateLimiter.java

/**
 * 当前桶中已存在的令牌数，如果请求需要的令牌数小于已存在的令牌数，就允许通过
 */
double storedPermits;

/**
 * 令牌桶可以保存的最大令牌数
 */
double maxPermits;

/**
 * 多长时间可以生成一个令牌，单位是微秒。比如RateLimiter.create(5)，就意味着1秒生成5个令牌，那么生成一个令牌就需要200ms
 */
double stableIntervalMicros;

/**
 * 重要！！！下一个请求可以被允许获取令牌的时间点，单位是微秒。
 */
private long nextFreeTicketMicros = 0L;

1、限流器创建的时候会初始化令牌吗？

我们从限流器的创建源码着手分析。
RateLimiter.java

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

  static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
    // 创建一个普通平滑限流器
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
    // 关键：设置限流器速率相关信息
    rateLimiter.setRate(permitsPerSecond);
    return rateLimiter;
  }

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

  // 由子类即SmoothRateLimiter来实现
  abstract void doSetRate(double permitsPerSecond, long nowMicros);

SmoothRateLimiter.java

@Override
  final void doSetRate(double permitsPerSecond, long nowMicros) {
    // 重点1：生成令牌，并同步下次可以获取令牌的时间
    resync(nowMicros);
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    // 将stableIntervalMicros从默认的0.0设置为 生成一个令牌所需的时间
    this.stableIntervalMicros = stableIntervalMicros;
    // 重点2
    doSetRate(permitsPerSecond, stableIntervalMicros);
  }

  // 重点1
    /** 限流器创建（doSetRate(double permitsPerSecond, long nowMicros)）
     * 以及 获取令牌（reserveEarliestAvailable(int requiredPermits, long nowMicros)）的时候都会调用这个方法
     * 如果是创建时调用 由于coolDownIntervalMicros返回值即stableIntervalMicros=0，所以当前storedPermits的计算结果仍为0
     *
     */
  void resync(long nowMicros) {
    if (nowMicros > nextFreeTicketMicros) {
      // 下一次可以获取令牌的时间到现在这段时间内，需要生成多少令牌，由于当前coolDownIntervalMicros()会返回0.0，所以计算结果为Infinity（无穷）
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
      // 保证桶里的令牌数不能超过最大允许的令牌数，因为newPermits=无穷，所以这里计算出桶里的令牌数应该是0
      storedPermits = min(maxPermits, storedPermits + newPermits);
      // 将nextFreeTicketMicros值设为限流器创建的时间
      nextFreeTicketMicros = nowMicros;
    }
  }

  // 由子类即SmoothBursty来实现
  abstract void doSetRate(double permitsPerSecond, double stableIntervalMicros);


static final class SmoothBursty extends SmoothRateLimiter {
    // 重点2
    @Override
    void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
      // 当前允许的最大令牌数，限流器创建时该值为0.0
      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 {
        // 如果最大允许的令牌数时0，则将桶里的令牌数也置为0
        storedPermits =
            (oldMaxPermits == 0.0)
                ? 0.0 // initial state
                : storedPermits * maxPermits / oldMaxPermits;
      }
    }
    
    @Override
    double coolDownIntervalMicros() {
      // 返回的就是生成一个令牌需要多长时间，该值在限流器创建的时候初始值为0.0
      return stableIntervalMicros;
    }
  }

通过上面源码中重点1和重点2的分析可以发现，在创建限流器的时候，当前桶中的令牌数一直是0。

结论：限流器创建的时候不会初始化令牌

2、令牌是如何放到桶里的？

我们经常看到对于令牌桶限流算法的描述是：将令牌每隔一段时间定时放入桶中。
乍一看也许需要一个定时器才能达到这个效果。但Guava的实现告诉我们其实不用这么复杂，只需要一个计数器（storedPermits）变量就能搞定。

想要知道令牌如何放到桶里，就需要从获取令牌的时候开始探索。

这有点奇怪对吗，正常是先把令牌放到桶里，然后才获取令牌，即有因才有果；但是我们却需要先知道如何获取令牌，才能知道令牌是如何放到桶里的。
在我看来，这正是Guava实现的巧妙之处。

RateLimiter.java

/**
* 尝试获取令牌
* @param permits 要获取的令牌数
* @param timeout 能获取到令牌的最大等待时间，等待时间超过这个时间就直接返回false。如果该值是0，不做任何等待，直接返回是否获取到令牌
*/
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)) {
        // 获取不了就返回false
        return false;
      } else {
        // 关键：如果在超时时间内能获取到令牌，计算需要等待的时间
        microsToWait = reserveAndGetWaitLength(permits, nowMicros);
      }
    }
    // 睡眠等待足够的时间
    stopwatch.sleepMicrosUninterruptibly(microsToWait);
    return true;
  }

  private boolean canAcquire(long nowMicros, long timeoutMicros) {
    // 获取最早可以获得令牌的时间
    return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
  }

  final long reserveAndGetWaitLength(int permits, long nowMicros) {
    // 关键：获取令牌并返回最早能获得令牌的时间
    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
    return max(momentAvailable - nowMicros, 0);
  }

  // 由子类即SmoothBursty实现
  abstract long queryEarliestAvailable(long nowMicros);
  
  // 由子类即SmoothBursty实现
  abstract long reserveEarliestAvailable(int permits, long nowMicros);

SmoothBursty.java

final long queryEarliestAvailable(long nowMicros) {
    // 又是它！！！待会分析它到底是个什么东西
    return nextFreeTicketMicros;
  }

  /**
   * 获取令牌的核心方法
   *
   * @param requiredPermits 需要获取的令牌数
   * @param nowMicros
   * @return
   */
  @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);

    // 重点3：更新下一次可以获取令牌的时间 = 当前时间 + 要拿到还差的令牌数要等的时间
    this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
    // 重点4：更新桶里还剩的令牌数
    this.storedPermits -= storedPermitsToSpend;
    return returnValue;
  }

  void resync(long nowMicros) {
    if (nowMicros > nextFreeTicketMicros) {
      // 下一次可以获取令牌的时间到现在这段时间内，需要生成多少令牌
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
      // 重点1：生成令牌并放入桶中
      storedPermits = min(maxPermits, storedPermits + newPermits);
      // 重点2：将nextFreeTicketMicros值设为当前时间
      nextFreeTicketMicros = nowMicros;
    }
  }

通过上面源码中的重点1、重点2、重点3、重点4可以发现：

重点1是向桶里放令牌，既增加令牌计数器storedPermits
重点4是从桶里获取令牌，既减少令牌计数器storedPermits
重点2和重点3都是更新nextFreeTicketMicros

所以令牌的生成、获取都围绕着两个变量：storedPermits（当前桶里的令牌数）和nextFreeTicketMicros（下次可以获得令牌的时间）。

而这两个变量也正是Guava限流设计的巧妙之处：不必提前向桶里放入令牌，或通过一个单独的定时器向桶里放令牌，而是在获取令牌的时候增加令牌数量再减少令牌数量。

用图来更加直观的体现这里的逻辑。

nextFreeTicketMicros在源码中其实是用微秒级时间戳表示，为了方便理解，下面就用正常时间来表示。

在这里插入图片描述

创建限流器。RateLimiter rateLimiter = RateLimiter.create(5);即QPS=5，每秒生成5个令牌，生成1个令牌需要200毫秒，桶内最大令牌数=5。storedPermits（此时桶里的令牌数）=0，nextFreeTicketMicros（下次可以获取令牌的时间）=0。
请求A要获取1个令牌。rateLimiter.acquire();当前时间是2023-9-26 10:00:00。
发现当前时间 > nextFreeTicketMicros，两者相差的这段时间远远大于1秒，而1秒可以生成5个令牌（最多也只能存5个）。同时要把nextFreeTicketMicros设置为当前时间，意味着现在桶里已经有令牌了，现在马上就可以获取到令牌。此时storedPermits=5，nextFreeTicketMicros=2023-9-26 10:00:00。
获取到1个令牌，此时storedPermits=4，nextFreeTicketMicros=2023-9-26 10:00:00。
请求B要获取10个令牌。rateLimiter.acquire(10);当前时间是2023-9-26 10:00:01.001。
发现当前时间 > nextFreeTicketMicros，两者相差的这段时间大于1秒，1秒可以生成5个令牌，当前桶里还有4个，5+4=9，但桶最多只能存5个。同时要把nextFreeTicketMicros设置为当前时间，意味着现在桶里已经有令牌了，现在马上就可以获取到令牌。此时storedPermits=5，nextFreeTicketMicros=2023-9-26 10:00:01.001。
需要获取10个令牌，但是现在桶里只有5个，即使全部获取还欠5个，那就提前透支5个咯。意味着接下来这1秒生成的5个令牌是预留给当前请求的，其它请求1秒后才能再获取令牌。此时storedPermits=0，nextFreeTicketMicros=2023-9-26 10:00:02.001。
请求C要获取1个令牌。rateLimiter.acquire();当前时间是2023-9-26 10:00:01.999。
由于nextFreeTicketMicros=2023-9-26 10:00:02.001。还没到下次可以获取令牌的时间，就只能等待。
等待ing …
当前时间是2023-9-26 10:00:02.200。当前时间 > nextFreeTicketMicros，相差的这段时间是200毫秒，刚好能生成1个令牌。同时要把nextFreeTicketMicros设置为当前时间，意味着现在桶里已经有令牌了，现在马上就可以获取到令牌。此时storedPermits=1，nextFreeTicketMicros=2023-9-26 10:00:02.200。
获取到1个令牌，此时storedPermits=0，nextFreeTicketMicros=2023-9-26 10:02:200。

结论：令牌的生成其实是在令牌的获取逻辑中。

3、如果要获取的令牌数大于桶里的令牌数会怎么样

经过上面的分析可以得出结论：会透支/预支不足的令牌数。

4、令牌数量的更新会有并发问题吗

可以看一下获取令牌时的源码：

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