简单分析Guava中RateLimiter中的令牌桶算法的实现

       令牌桶算法是网络流量整形（Traffic Shaping）和速率限制（Rate Limiting）中最常使用的一种算法。典型情况下，令牌桶算法用来控制发送到网络上的数据的数目，并允许突发数据的发送。

    大小固定的令牌桶可自行以恒定的速率源源不断地产生令牌。如果令牌不被消耗，或者被消耗的速度小于产生的速度，令牌就会不断地增多，直到把桶填满。后面再产生的令牌就会从桶中溢出。最后桶中可以保存的最大令牌数永远不会超过桶的大小。                                                                          

                                                                                                                         ——摘自百度百科

那么什么是令牌桶算法呢？

        简单来说就是，生产者和消费者之间的事情，生产者往一个桶(Bucket)中丢令牌(Token),消费者从里面去捡令牌，生产者以一定的速率丢令牌，直到桶装满了，令牌就溢出了，消费者持续从桶里面捡令牌，没有令牌的话，就持续等待，直到有令牌出现。

 

 这里我们看下具体令牌桶算法的实现(Guava中的RateLimiter)，以及在实际生产中的应用场景(限制接口访问频次，保护后端系统)

       我们在暴露对外接口的时候，对于高频次访问的接口(例如查询接口),需要考虑到相关的保护措施，避免接口瞬时访问量过大，导致服务端不可用的场景产生。因此，我们可以使用RateLimiter，来做相关的频控。

  下面是RateLimiter的使用Demo：

   1、引入相关的依赖

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

 2、编写相关的Demo

public class RateLimiterTest {
    public static void main(String[] args) throws InterruptedException {
        RateLimiter limiter = RateLimiter.create(10);// 代码1
        Thread.currentThread().sleep(1000);//步骤1
        if (limiter.tryAcquire(20))//代码2
            System.out.println("======== Time1:" + System.currentTimeMillis() / 1000);
 
        Thread.currentThread().sleep(1001);
        if (limiter.tryAcquire(1))//代码3
            System.out.println("======== Time2:" + System.currentTimeMillis() / 1000);
 
        if (limiter.tryAcquire(5))
            System.out.println("======== Time3:" + System.currentTimeMillis() / 1000);
 
    }
}

3、运行结果如下

场景1:

======== Time1:1533114071
======== Time2:1533114072

场景2：修改代码:去掉步骤1，运行结果如下:

======== Time1:1533114155

场景3：修改相关代码如下：

public class RateLimiterTest {
    public static void main(String[] args) throws InterruptedException {
        RateLimiter limiter = RateLimiter.create(10);// 代码1
        Thread.currentThread().sleep(2000);
        if (limiter.tryAcquire(21))//代码2
            System.out.println("======== Time1:" + System.currentTimeMillis() / 1000);
 
        Thread.currentThread().sleep(1001);
        if (limiter.tryAcquire(1))//代码3
            System.out.println("======== Time2:" + System.currentTimeMillis() / 1000);
 
        if (limiter.tryAcquire(5))
            System.out.println("======== Time3:" + System.currentTimeMillis() / 1000);
 
    }
}

    结果如下：

======== Time1:1533114623

 下面我们来分析这三种情况产生的原因，顺便也分析下RateLimiter中的令牌桶算法是如何实现的。

在分析之前，说明一点，我之前一直以为令牌桶算法，是定时器机制，定时往桶里面放令牌，但是有些时候并不是这样的。先声明一下。

我们来分析下代码：

         代码行1：

 RateLimiter limiter = RateLimiter.create(10);

    这行代码，我们知道是创建一个每秒产生10个permit的速率器

      代码行2:
           limiter.tryAcquire(20)  //尝试从速率器中获取20个permit，获取成功 true；失败 false
      代码行3:
            limiter.tryAcquire(1) //尝试从速率器中获取1个permit，获取成功 true；失败 false

        为什么相同的代码，不同的休眠时间导致不同的结果呢？

结论：

    1、RateLimiter 速率器，通过预支将来的令牌来进行限制频控，什么意思呢？打个比方:速率器相当于工厂，获取令牌许可的线程相当于经销商，经销商过来取货，工厂每天的生产的货品是一定的(100吨/天)，A经销商来取货，第一天取了200吨货，工厂没有这么多货，怎么办呢？为了留住这个经销商，厂长做了决定，给200吨，现在的100吨先给你，明天的100吨也给你，然后把200吨货品的提取清单给了A经销商，A很满意的离开了。过了一会，B来了，B要10吨货物，这个时候，厂长就没有那么好说话了(谁让大客户已经到手了呢？),说10吨货物可以，你
后天来吧，明天和今天的活已经都卖完了。这个时候通过这种方式，来限制一天只卖/生产100吨的货物。

  根据这个结论我们来看相关的代码：

RateLimiter limiter = RateLimiter.create(10);

调用的是：

@VisibleForTesting
  static RateLimiter create(SleepingStopwatch stopwatch, double permitsPerSecond) {
    RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds 注意 这里的maxBurstSeconds指定的是1s 直接影响后面的maxPermit*/);
    rateLimiter.setRate(permitsPerSecond);//见下文代码
    return rateLimiter;
  }

setRate(permitsPerSecond)如下：

public final void setRate(double permitsPerSecond) {
    checkArgument(
        permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
    synchronized (mutex()) {
      doSetRate(permitsPerSecond, stopwatch.readMicros());//stopwatch.readMirco 获取的是创建以来的系统时间 这里调用SmoothRateLimiter.doSetRate()
    }
  }

 SmoothRateLimiter.doSetRate()

 @Override
  final void doSetRate(double permitsPerSecond, long nowMicros) {
    resync(nowMicros);//你可以认为这边是重设相关的nextFreeTicketMicros和storedPermits 这个函数是相关计算频控的重要组成部分  ------1
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    doSetRate(permitsPerSecond, stableIntervalMicros);//这个函数是RateLimiter创建时候 初始化maxpermits和StorePermits的相关部分 也是一个重要的部分 ---2
  }

我们来看1的实现：

    

/**
   * Updates {@code storedPermits} and {@code nextFreeTicketMicros} based on the current time.
   * 基于当前的时间 计算相关的storedPermits和nextFreeTicketMicros 
   *  storedPermits:当前存储的令牌数
   *  nextFreeTicketMicros:下次可以获取令牌的时间 其实这么讲不太准确 应该说是，上次令牌获取之后预支到下次可以获取令牌的最早时间
   *         此处再创建的时候 nextFreeTicketMicros基本就是创建时候的系统时间
   */
  void resync(long nowMicros) {
    // if nextFreeTicket is in the past, resync to now
    if (nowMicros > nextFreeTicketMicros) {
      storedPermits = min(maxPermits,
          storedPermits
            + (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());
      nextFreeTicketMicros = nowMicros; 
    }
  }

        我们可以看到，我们这里通过计算当前时间和下次可以获取令牌的时间，相互计算差值，然后除以一个令牌产生的时间间隔，来计算当前时段可以产生多少令牌，然后和我们的     maxPermits来取最小值，由此我们可以看到storedPermits最多只能存储maxPermits数量的令牌，这也是令牌桶大小所限制的。
    我们再来看2代码的实现:

@Override
    void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
      double oldMaxPermits = this.maxPermits;
      maxPermits = maxBurstSeconds * permitsPerSecond;//设置最大可存储的令牌数 这里的maxBurstSeconds 就是之前设置的1s 所以maxPermits数值上等于我们设置的permitsPerSecond
      if (oldMaxPermits == Double.POSITIVE_INFINITY) {
        // if we don't special-case this, we would get storedPermits == NaN, below
        storedPermits = maxPermits;
      } else {
        storedPermits = (oldMaxPermits == 0.0)
            ? 0.0 // initial state
            : storedPermits * maxPermits / oldMaxPermits;
      }
    }

到这里我们的初始化RateLimiter结束了。我们来明确其中的几个概念:

    maxPermits:最大存储的令牌数，即令牌桶的大小

    storedPermits:已存储的令牌数<=maxPermits,当然这个是通过计算算出来的

    nextFreeTicketMicros:上次获取令牌时预支的最早能够再次获取令牌的时间

    nowMicros:当前系统时间

好，我们接下来看如何获取令牌：

  代码2:

limiter.tryAcquire(20)

 具体的代码实现如下：

public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {//timeout = 0 unit=MICROSECONDS
    long timeoutMicros = max(unit.toMicros(timeout), 0);
    checkPermits(permits);//校验参数
    long microsToWait;
    synchronized (mutex()) {//互斥量 
      long nowMicros = stopwatch.readMicros();
      if (!canAcquire(nowMicros, timeoutMicros)) {//此处判断当前时间是否大于等于上次预支最早时间  ----1
        return false;
      } else {
        microsToWait = reserveAndGetWaitLength(permits, nowMicros);//当前线程获取到permit需要等待的时间 ---2
      }
    }
    stopwatch.sleepMicrosUninterruptibly(microsToWait);//线程等待 获取permit
    return true;
  }

我们来看1的实现部分:

private boolean canAcquire(long nowMicros, long timeoutMicros) {
    return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros;
  }

@Override
  final long queryEarliestAvailable(long nowMicros) {
    return nextFreeTicketMicros;
  }

有此可见，如果当前时间+超时时间>=预支的最早时间，那么是可以获取许可的，反之则不能获取许可

再看代码2的实现:

final long reserveAndGetWaitLength(int permits, long nowMicros) {
    long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
    return max(momentAvailable - nowMicros, 0);//计算需要等待的时间
  }

SmoothRateLimiter.reserveEarliestAvailable()

@Override
  final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
    resync(nowMicros);//这里是重设相关的storedPermits和nenextFreeTicketMicros 这个在前文我们讲过 需要注意的是 这边的nextFreeTicketMicros设置的是nowMicros 可能会有人有疑问，nextFreeTicketMicros不是预支的最早获取permit的时间吗？怎么是nowMicros了呢？我们下面看
    long returnValue = nextFreeTicketMicros;//这里返回的其实就是nowMiscros
    double storedPermitsToSpend = min(requiredPermits, this.storedPermits);//本次能消费的最多的permit
    double freshPermits = requiredPermits - storedPermitsToSpend;//需要预支的permit
    long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
        + (long) (freshPermits * stableIntervalMicros);//预支的生产的时间
 
    try {
      this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros);//这里就是重设了预支下次能够获取permit的最早时间了 这边将waitMiscros加上了
    } catch (ArithmeticException e) {
      this.nextFreeTicketMicros = Long.MAX_VALUE;
    }
    this.storedPermits -= storedPermitsToSpend;//扣除本地消费的permit
    return returnValue;//返回当前时间
  }

      这样就完成了前后两个permit之间获取的的联动性，并不是有一个定时任务往中间放permit，而是直接预支的后面消费者的时间来进行控制的，这样有一个好处就是，第一次获取permit的时候，其实可以获取N多个permit，并不做限制，只是这么多的permit会导致后面消费者卡死在那边，当然，消费者在timeOut范围内获取不到permit也就直接返回了。

 

Q:

    思考下 前后两个线程之间的同步部分，为什么还要等待一段时间？最多能储存多少permit？令牌桶有什么弊端(或者说RateLimiter可能存在的问题)？