令牌桶算法是网络流量整形（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);// 代码1Thread.currentThread().sleep(1000);//步骤1if (limiter.tryAcquire(20))//代码2System.out.println("======== Time1:" + System.currentTimeMillis() / 1000);Thread.currentThread().sleep(1001);if (limiter.tryAcquire(1))//代码3System.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);// 代码1Thread.currentThread().sleep(2000);if (limiter.tryAcquire(21))//代码2System.out.println("======== Time1:" + System.currentTimeMillis() / 1000);Thread.currentThread().sleep(1001);if (limiter.tryAcquire(1))//代码3System.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);

调用的是：

@VisibleForTestingstatic 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()

 @Overridefinal void doSetRate(double permitsPerSecond, long nowMicros) {resync(nowMicros);//你可以认为这边是重设相关的nextFreeTicketMicros和storedPermits 这个函数是相关计算频控的重要组成部分  ------1double 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 nowif (nowMicros > nextFreeTicketMicros) {storedPermits = min(maxPermits,storedPermits+ (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros());nextFreeTicketMicros = nowMicros; }}

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

@Overridevoid doSetRate(double permitsPerSecond, double stableIntervalMicros) {double oldMaxPermits = this.maxPermits;maxPermits = maxBurstSeconds * permitsPerSecond;//设置最大可存储的令牌数 这里的maxBurstSeconds 就是之前设置的1s 所以maxPermits数值上等于我们设置的permitsPerSecondif (oldMaxPermits == Double.POSITIVE_INFINITY) {// if we don't special-case this, we would get storedPermits == NaN, belowstoredPermits = 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=MICROSECONDSlong timeoutMicros = max(unit.toMicros(timeout), 0);checkPermits(permits);//校验参数long microsToWait;synchronized (mutex()) {//互斥量 long nowMicros = stopwatch.readMicros();if (!canAcquire(nowMicros, timeoutMicros)) {//此处判断当前时间是否大于等于上次预支最早时间  ----1return false;} else {microsToWait = reserveAndGetWaitLength(permits, nowMicros);//当前线程获取到permit需要等待的时间 ---2}}stopwatch.sleepMicrosUninterruptibly(microsToWait);//线程等待 获取permitreturn true;}

我们来看1的实现部分:

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

@Overridefinal 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()

@Overridefinal long reserveEarliestAvailable(int requiredPermits, long nowMicros) {resync(nowMicros);//这里是重设相关的storedPermits和nenextFreeTicketMicros 这个在前文我们讲过 需要注意的是 这边的nextFreeTicketMicros设置的是nowMicros 可能会有人有疑问，nextFreeTicketMicros不是预支的最早获取permit的时间吗？怎么是nowMicros了呢？我们下面看long returnValue = nextFreeTicketMicros;//这里返回的其实就是nowMiscrosdouble storedPermitsToSpend = min(requiredPermits, this.storedPermits);//本次能消费的最多的permitdouble freshPermits = requiredPermits - storedPermitsToSpend;//需要预支的permitlong 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;//扣除本地消费的permitreturn returnValue;//返回当前时间}

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

Q:

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