服务限流之Guava源码解析

1、前言

在应用系统中有三把利器能够保护系统，提升系统健壮性：缓存、限流、降级，它们分别在不一样场景下使用，但最终但目的都是用来保护系统稳定运行。其各自的使用场景及用途以下：

1）缓存：提升系统响应速度，为数据库减压，避免过多查询数据库；

2）限流：限制并发量，限制某一段时间只有指定数量的请求进入后台服务器，遇到流量高峰期或者流量突增（流量尖刺）时，把流量速率限制在系统所能接受的合理范围以内，不至于让系统被高流量击垮；

3）降级：当访问某个服务出现问题或影响核心业务时，暂时屏蔽该服务，直到请求高峰事后或者服务正常时再打开。

2、经常使用限流算法

常见等限流算法主要有2种：令牌桶、漏桶。令牌桶算法是令牌以固定速率流入桶内，桶满了则丢弃令牌，桶内没有令牌时则阻塞，当桶内有令牌但不够时，不会阻塞，容许预先消费，可解决流量突发状况。漏桶法则容许以任意速率流入桶内，但流出速率是恒定的，桶满了则丢弃令牌，桶内没有令牌时则阻塞，它不能解决流量突发状况。具体流程如图所示：

1）令牌桶法

2）漏桶法

3、常见的限流场景

开发过程当中或多或少会接触到服务限流，好比tomcat限制最大链接数、数据库链接池限制链接数、nginx限制ip访问数、秒杀、抢购等。这些都是经过限制一个时间窗口内的请求数，当达到设置的最大请求数后，会让后续请求进入等待队列或直接拒绝，防止系统过载。这些限流是怎么作到的呢，或者说限流主要有哪些方式呢？不少人把限流概括成4种场景，其实也不外乎这4种场景，它们分别是：

1）限制总并发数或请求数：针对整个系统作拦截

2）限制接口的总并发或请求数：针对单个接口作拦截

3）限制接口每秒的请求数：针对整个接口作拦截，时间窗口为1秒

4）平滑限流接口的请求数：平滑突发限流（SmoothBursty）、平滑预热限流（SmoothWarmingUp）

4、Guava使用及原理解析

4.1 Guava使用

Guava是google提供的开源工具包，提供了限流工具类RateLimiter，该类基于令牌桶算法实现流量限制，只需在pom.xml文件引入guava依赖，使用起来很方便。

从执行结果能够看到第一次获取时等待时间为0.0s，后续每次获取令牌都需等待0.5秒，0.5表示每秒往容器里平滑的放2个凭证，每0.5秒放入一个，0.5是怎么得来的呢？其实就是1/2=0.5，而2便是RateLimiter.create(2)这个方法中的参数值。那么Guava是怎么实现的呢？

4.2 Guava实现原理

Guava的RateLimiter限流步骤主要分2步：初始化、获取凭证，初始化主要是建立RateLimiter对象，并初始化对应的属性值，其中比较重要的几个属性以下：

4.1.1 容器初始化

初始化过程会作以下赋值：storedPermits=0、nextFreeTicketMicros=当前时间、stableIntervalMicros=1/qps，具体流程以下：

static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) {
  // 建立RateLimiter对象
  RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
  // 初始化storedPermits、nextFreeTicketMicros、stableIntervalMicros的值
  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());
  }
}

 

final void doSetRate(double permitsPerSecond, long nowMicros) {
  // 计算storedPermits、nextFreeTicketMicros的值
  resync(nowMicros);
  // 计算每一个凭证流入的间隔时间
  double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
  this.stableIntervalMicros = stableIntervalMicros;
  doSetRate(permitsPerSecond, stableIntervalMicros);
}

 

void resync(long nowMicros) {
  // 若是当前时间可以获取凭证，每次获取凭证时都会计算下一次能获取凭证的时间
  if (nowMicros > nextFreeTicketMicros) {
    // 容器里的总凭证数，即result=容器中原有凭证数+该时间点应该往容器里放多少凭证
    storedPermits = min(maxPermits,
        storedPermits
          + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
    nextFreeTicketMicros = nowMicros;
  }
}

4.1.2 获取凭证

public double acquire(int permits) {
  // 计算须要等待的时间
  long microsToWait = reserve(permits);
  // 线程阻塞指定时间（tryAcquire()方法未非阻塞方法）
  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);
  // 2个数取最大值，当存在预先消费或提早获取凭证时momentAvailable>nowMicros
  return max(momentAvailable - nowMicros, 0);
}

 

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  // 计算storedPermits、nextFreeTicketMicros当值
  resync(nowMicros);
  long returnValue = nextFreeTicketMicros;
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
  // 计算容器中还需刷新的凭证数，即预先消费时会存在requiredPermits>storedPermitsToSpend
  double freshPermits = requiredPermits - storedPermitsToSpend;
  // 计算下一次获取凭证时需等待的时间，预先消费时该值为freshPermits * stableIntervalMicros
  long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
      + (long) (freshPermits * stableIntervalMicros);

  try {
    // 从新计算nextFreeTicketMicros的值
    this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros);
  } catch (ArithmeticException e) {
    this.nextFreeTicketMicros = Long.MAX_VALUE;
  }
  // 减小容器中的凭证数
  this.storedPermits -= storedPermitsToSpend;
  return returnValue;
}

 

void resync(long nowMicros) {
  // 若是当前时间可以获取凭证，每次获取凭证时都会计算下一次能获取凭证的时间 
  if (nowMicros > nextFreeTicketMicros) {
    // 容器里的总凭证数，即result=容器中原有凭证数+该时间点应该往容器里放多少凭证 
    storedPermits = min(maxPermits,
        storedPermits
          + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
    nextFreeTicketMicros = nowMicros;
  }
}