令牌桶算法

令牌桶算法介绍

令牌桶算法，又称token bucket。

从图中我们可以看到，令牌桶算法比漏桶算法稍显复杂。首先，我们有一个固定容量的桶，桶里存放着令牌（token）。桶一开始是空的，token以一个固定的速率r往桶里填充，直到达到桶的容量，多余的令牌将会被丢弃。每当一个请求过来时，就会尝试从桶里移除一个令牌，如果没有令牌的话，请求无法通过。

令牌桶算法实现

令牌桶算法实现代码如下：

package com.morris.user.demo;

import lombok.extern.slf4j.Slf4j;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 令牌桶算法
 */
@Slf4j
public class TokenBucketDemo {

    public static void main(String[] args) {

        TokenBucket sideWindow = new TokenBucket(2);
        for (int i = 0; i < 30; i++) {
            int finalI = i;
            new Thread(() -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(new Random().nextInt(3000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info(finalI + "--> " + sideWindow.canPass());
            }).start();
        }
    }
    
    public static class TokenBucket {
        // 上一次请求通过的时间
        private long lastRequestTime = System.currentTimeMillis();
        // 往桶内加入令牌的速率
        private final long rate;
        // 桶的容量
        private static final long CAPACITY = 10;
        // 桶内当前令牌的数量，要使用小数
        private double tokens = 0;

        public TokenBucket(int maxCount) {
            this.rate = maxCount;
        }

        public synchronized boolean canPass() {
            long now = System.currentTimeMillis();

            // 计算从lastRequestTime~now时间段应该增加的令牌数
            this.tokens = Math.min(CAPACITY, this.tokens + (now - lastRequestTime) * 1.0D / 1000 * this.rate);

            lastRequestTime = now;

            if (this.tokens < 1) {
                // 令牌不足
                return false;
            }
            this.tokens--;
            return true;
        }
    }
}

令牌桶限流原理

工作过程：

• 请求流量从令牌桶中获取令牌，持有令牌就放行，否则就会被拒绝
• 假如设置阈值每秒允许100个请求通过，则请求发送的速率为100个请求每秒，r=100/s
• 需要令牌生产的速率为1秒/r，即1/100，每10毫秒产生一个令牌
• 令牌桶容量为b，已满多余的令牌将被丢弃
• 令牌桶为空请求被拒
• 允许突发流量最大突发流量为令牌桶容量b
• 请求通过相应的令牌从令牌桶中移除

假如系统平时流量很低，突然陡增的流量需要缓慢增加。具体到令牌桶，可以通过控制令牌的生产速率来对流量进行控制。令牌生产速率如何控制？

我们在使用sentinel设置QPS的预热流控时，需要设置阈值count和预热时长warmUpPeriodInSec，下面梳理下与下图坐标图的关系。

坐标图说明：

参数	说明
x轴	表示令牌桶中的令牌数量
y轴	生产一个令牌需要的时间（秒）
stableInterval	稳定生产一个令牌需要的时间
coldInterval	生产一个令牌需要的最大时长，与冷启动因子coldFactor有关，可以通过-Dcsp.sentinel.flow.cold.factor设置，默认为3。
warmUpPeriodInSec	预热时长，默认为10秒。对应到坐标图中为（2）梯形面积
thresholdPermits(warningToken)	令牌桶中的一个阈值，超过该值时开启预热
maxPermits(maxToken)	令牌桶中最大令牌数

换算关系：

• count，已知由用户设置，例如每秒允许通过100个请求
• warmUpPeriodInSec，已知由用户设置，默认为10秒，时间区域上红色（2）梯形区域
• coldFactor，已知默认为3

stableInterval = 1/count

coldInterval = stableInterval * coldFactor

由于coldFactor默认为3，y轴stableInterval_{coldInterval的距离是0}stableInterval的距离两倍，所以从thresholdPermits到0的时间是从maxPermits到thresholdPermits时间的一半，也就是冷启动周期的一半。因为梯形的面积等于warmupPeriod，所以长方形面积是梯形面积的一半，长方形的面积是warmupPeriod/2。

根据长方形面积公式：长 * 宽  = 面积，可得：

坐标时间（1）长方形区域面积 = 长（thresholdPermits(warningToken)） * 宽（stableInterval）

坐标时间（1）长方形区域面积 = 0.5 * warmUpPeriodInSec

thresholdPermits(warningToken) = 0.5 * warmUpPeriodInSec / stableInterval

根据梯形的面积公式：（上低 + 下低）* 高 / 2，可得：

warmupPeriod = （stableInterval + coldInterval）* （maxPermits - thresholdPermits）/ 2

maxPermits = thresholdPermits + 2 * warmupPeriod / （stableInterval + coldInterval）

备注：由斜率公式k=(y1-y2)➗(x1-x2)，得出斜率如下:

slope = (coldInterval-stableInterval)➗(maxPermits(maxToken)-thresholdPermits(warningToken))

Sentinel每秒生产一次令牌，将新生产的令牌放入令牌桶，并记录本次生产令牌的时间，当下次生产时，根据当前时间与上一次生产令牌的时间间隔计算、以及每个令牌的生产间隔时间计算出本次需要生产的令牌数。

服务第一次启动时，或者接口很久没有被访问，都会导致当前时间与上次生产令牌的时间相差甚远，所以第一次生产令牌将会生产maxPermits个令牌，直接将令牌桶装满。由于令牌桶已满，接下来10s就是冷启动阶段。

由于冷启动阶段生产令牌的间隔时间比较正常消费速度慢，因此随着时间的推移，桶中的剩余令牌数就会趋近于thresholdPermits，生产令牌的时间间隔也会从coldInterval降低到stableInterval。当桶中剩余令牌数小于thresholdPermits时，冷启动结束，系统进入稳定状态，生产令牌的时间间隔为stableInterval，每秒生产的令牌数就等于QPS。

Sentinel并不会在请求通过时减少令牌桶中的令牌数量，而是在下一秒生产新的令牌时，再减去桶中与上一秒通过的请求数相等数量的令牌，这就是Sentinel官方介绍的令牌自动掉落。

Sentinel没有在每个请求通过时从令牌桶取走令牌，那么Sentinel是如何控制QPS的呢，我们再来看一张图：

• x1：当前令牌桶中超过thresholdPermits的令牌数量；

• y1：y1加上stableInterval等于当前令牌生产的时间间隔；

根据斜率和x1可算出y1：

y1 = slope * x1

y1加上stableInterval即为当前的令牌生产速率。

当前秒生产令牌的时间间隔为：

slope * (storedTokens - thresholdPermits) + stableInterval 

由于：stableInterval  = 1.0(1秒) / 限流阈值(count)

所以上述等式 =  slope * (storedTokens - thresholdPermits) + 1.0 / count

最后算得当前时间戳的QPS阈值为：

1.0 / slope * (storedTokens - thresholdPermits) + 1.0 / count

当令牌桶中的令牌数小于thresholdPermits(warningToken)时，令牌按照固定速率生产，请求流量稳定。当令牌数大于thresholdPermits(warningToken)时，开启预热。此段时期，生产的令牌的速率小于令牌滑落的速度，一段时间后，令牌小于等于thresholdPermits(warningToken)，请求回归到稳定状态，预热结束。

令牌桶算法在sentinel中的应用

漏桶算法主要用于流控规则中流控效果为Warm UP。

Sentinel中使用的令牌桶算法，是参考着Guava中的令牌桶算法来的。

WarmUpController的构造方法中会对告警阈值。斜率等参数进行计算并初始化。

public WarmUpController(double count, int warmUpPeriodInSec) {
    construct(count, warmUpPeriodInSec, 3);
}

private void construct(double count, int warmUpPeriodInSec, int coldFactor) {

    if (coldFactor <= 1) {
        throw new IllegalArgumentException("Cold factor should be larger than 1");
    }

    // 最大阈值
    this.count = count;

    // 冷启动因子3
    this.coldFactor = coldFactor;

    // 告警阈值，超过该值，开启冷启动
    // thresholdPermits = 0.5 * warmupPeriod / stableInterval.
    warningToken = (int)(warmUpPeriodInSec * count) / (coldFactor - 1);
    // / maxPermits = thresholdPermits + 2 * warmupPeriod / (stableInterval + coldInterval)
    // 令牌桶中允许的最大令牌数
    maxToken = warningToken + (int)(2 * warmUpPeriodInSec * count / (1.0 + coldFactor));

    // slope
    // slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits);
    // 斜率
    slope = (coldFactor - 1.0) / count / (maxToken - warningToken);

}

再来看下限流时是怎么做的？

com.alibaba.csp.sentinel.slots.block.flow.controller.WarmUpController#canPass(com.alibaba.csp.sentinel.node.Node, int, boolean)

public boolean canPass(Node node, int acquireCount, boolean prioritized) {
    // 当前的QPS
    long passQps = (long) node.passQps();

    // 上一个时间窗口的QPS
    long previousQps = (long) node.previousPassQps();

    // 计算令牌
    syncToken(previousQps);

    // 开始计算它的斜率
    // 如果进入了警戒线，开始调整他的qps
    long restToken = storedTokens.get();
    if (restToken >= warningToken) {
        // 开启预热
        long aboveToken = restToken - warningToken;
        // 消耗的速度要比warning快，但是要比慢
        // current interval = restToken*slope+1/count
        // 根据斜率计算出预热时的QPS
        double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count));
        if (passQps + acquireCount <= warningQps) {
            return true;
        }
    } else {
        // 不开启预热
        if (passQps + acquireCount <= count) {
            return true;
        }
    }

    return false;
}

protected void syncToken(long passQps) {
    // 当前时间
    long currentTime = TimeUtil.currentTimeMillis();
    // 当前秒数
    currentTime = currentTime - currentTime % 1000;
    long oldLastFillTime = lastFilledTime.get();
    if (currentTime <= oldLastFillTime) {
        return;
    }

    // 当前令牌中的桶数
    long oldValue = storedTokens.get();
    // 计算新产生的令牌数
    long newValue = coolDownTokens(currentTime, passQps);

    // 将新产生的令牌放入令牌桶
    if (storedTokens.compareAndSet(oldValue, newValue)) {
        // 从令牌桶中减去已使用的令牌
        long currentValue = storedTokens.addAndGet(0 - passQps);
        if (currentValue < 0) {
            storedTokens.set(0L);
        }
        lastFilledTime.set(currentTime);
    }

}

private long coolDownTokens(long currentTime, long passQps) {
    long oldValue = storedTokens.get();
    long newValue = oldValue;

    // 添加令牌的判断前提条件:
    // 当令牌的消耗程度远远低于警戒线的时候
    if (oldValue < warningToken) {
        newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
    } else if (oldValue > warningToken) {
        if (passQps < (int)count / coldFactor) {
            newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);
        }
    }
    return Math.min(newValue, maxToken);
}