常用限流算法与Guava RateLimiter源码解析，

常用限流算法与Guava RateLimiter源码解析，

在分布式系统中，应对高并发访问时，缓存、限流、降级是保护系统正常运行的常用方法。当请求量突发暴涨时，如果不加以限制访问，则可能导致整个系统崩溃，服务不可用。同时有一些业务场景，比如短信验证码，或者其它第三方API调用，也需要提供必要的访问限制支持。还有一些资源消耗过大的请求，比如数据导出等（参考 记一次线上Java服务CPU 100%处理过程 ），也有限制访问频率的需求。

常见的限流算法有令牌桶算法，漏桶算法，与计数器算法。本文主要对三个算法的基本原理及Google Guava包中令牌桶算法的实现RateLimiter进行介绍，下一篇文章介绍最近写的一个以RateLimiter为参考的分布式限流实现及计数器限流实现。

令牌桶算法

令牌桶算法的原理就是以一个恒定的速度往桶里放入令牌，每一个请求的处理都需要从桶里先获取一个令牌，当桶里没有令牌时，则请求不会被处理，要么排队等待，要么降级处理，要么直接拒绝服务。当桶里令牌满时，新添加的令牌会被丢弃或拒绝。

令牌桶算法的处理示意图如下（图片来自网络）

令牌桶算法主要是可以控制请求的平均处理速率，它允许预消费，即可以提前消费令牌，以应对突发请求，但是后面的请求需要为预消费买单（等待更长的时间），以满足请求处理的平均速率是一定的。

漏桶算法

漏桶算法的原理是水（请求）先进入漏桶中，漏桶以一定的速度出水（处理请求），当水流入速度大于流出速度导致水在桶内逐渐堆积直到桶满时，水会溢出（请求被拒绝）。

漏桶算法的处理示意图如下（图片来自网络）

漏桶算法主要是控制请求的处理速率，平滑网络上的突发流量，请求可以以任意速度进入漏桶中，但请求的处理则以恒定的速度进行。

计数器算法

计数器算法是限流算法中最简单的一种算法，限制在一个时间窗口内，至多处理多少个请求。比如每分钟最多处理10个请求，则从第一个请求进来的时间为起点，60s的时间窗口内只允许最多处理10个请求。下一个时间窗口又以前一时间窗口过后第一个请求进来的时间为起点。常见的比如一分钟内只能获取一次短信验证码的功能可以通过计数器算法来实现。

Guava RateLimiter解析

Guava是Google开源的一个工具包，其中的RateLimiter是实现了令牌桶算法的一个限流工具类。在pom.xml中添加guava依赖，即可使用RateLimiter

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>29.0-jre</version>
</dependency>

如下测试代码示例了RateLimiter的用法,

public static void main(String[] args) {
    RateLimiter rateLimiter = RateLimiter.create(1); //创建一个每秒产生一个令牌的令牌桶
    for(int i=1;i<=5;i++) {
        double waitTime = rateLimiter.acquire(i); //一次获取i个令牌
        System.out.println("acquire:" + i + " waitTime:" + waitTime);
    }
}

运行后，输出如下，

acquire:1 waitTime:0.0
acquire:2 waitTime:0.997729
acquire:3 waitTime:1.998076
acquire:4 waitTime:3.000303
acquire:5 waitTime:4.000223

第一次获取一个令牌时，等待0s立即可获取到（这里之所以不需要等待是因为令牌桶的预消费特性），第二次获取两个令牌，等待时间1s，这个1s就是前面获取一个令牌时因为预消费没有等待延到这次来等待的时间，这次获取两个又是预消费，所以下一次获取（取3个时）就要等待这次预消费需要的2s了，依此类推。可见预消费不需要等待的时间都由下一次来买单，以保障一定的平均处理速率（上例为1s一次）。

RateLimiter有两种实现：

类结构如下

关键属性及方法解析（以 SmoothBursty 为例）

1.关键属性

/** 桶中当前拥有的令牌数. */
double storedPermits;

/** 桶中最多可以保存多少秒存入的令牌数 */
double maxBurstSeconds;

/** 桶中能存储的最大令牌数，等于storedPermits*maxBurstSeconds. */
double maxPermits;

/** 放入令牌的时间间隔*/
double stableIntervalMicros;

/** 下次可获取令牌的时间点，可以是过去也可以是将来的时间点*/
private long nextFreeTicketMicros = 0L;

2.关键方法

调用 RateLimiter.create(double permitsPerSecond) 方法时，创建的是 SmoothBursty 实例，默认设置 maxBurstSeconds 为1s。SleepingStopwatch 是guava中的一个时钟类实现。

@VisibleForTesting
static RateLimiter create(double permitsPerSecond, SleepingStopwatch stopwatch) {
		RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0 /* maxBurstSeconds */);
		rateLimiter.setRate(permitsPerSecond);
		return rateLimiter;
}

SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {
	super(stopwatch);
	this.maxBurstSeconds = maxBurstSeconds;
}

并通过调用 SmoothBursty.doSetRate(double, long) 方法进行初始化，该方法中:

源码如下所示

@Override
final void doSetRate(double permitsPerSecond, long nowMicros) {
		resync(nowMicros);
		double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
		this.stableIntervalMicros = stableIntervalMicros;
		doSetRate(permitsPerSecond, stableIntervalMicros);
}

/** Updates {@code storedPermits} and {@code nextFreeTicketMicros} based on the current time. */
void resync(long nowMicros) {
		// if nextFreeTicket is in the past, resync to now
		if (nowMicros > nextFreeTicketMicros) {
		double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
		storedPermits = min(maxPermits, storedPermits + newPermits);
		nextFreeTicketMicros = nowMicros;
		}
}

@Override
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
		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 {
				storedPermits =
						(oldMaxPermits == 0.0)
								? 0.0 // initial state
								: storedPermits * maxPermits / oldMaxPermits;
		}
}

调用 acquire(int) 方法获取指定数量的令牌时，

源码如下所示

/** 获取指定数量（permits）的令牌，阻塞直到获取到令牌，返回等待的时间*/
@CanIgnoreReturnValue
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());
		}
}

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

/** 针对此次需要获取的令牌数更新下次可取令牌时间点与存储的令牌数，返回本次可取令牌的时间点*/
@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); // 需要等待的时间

		this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros); // 更新下次可取令牌的时间点
		this.storedPermits -= storedPermitsToSpend; // 更新当前存储的令牌数
		return returnValue;
}

acquire(int) 方法是获取不到令牌时一直阻塞，直到获取到令牌，tryAcquire(int,long,TimeUnit) 方法则是在指定超时时间内尝试获取令牌，如果获取到或超时时间到则返回是否获取成功

源码如下所示

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)) { //判断是否能在超时时间内获取指定数量的令牌
						return false;
				} else {
						microsToWait = reserveAndGetWaitLength(permits, nowMicros);
				}
		}
		stopwatch.sleepMicrosUninterruptibly(microsToWait);
		return true;
}

private boolean canAcquire(long nowMicros, long timeoutMicros) {
		return queryEarliestAvailable(nowMicros) - timeoutMicros <= nowMicros; //只要可取时间小于当前时间+超时时间，则可获取（可预消费的特性！）
}

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

以上就是 SmoothBursty 实现的基本处理流程。注意两点：

总结

本文介绍了限流的三种基本算法，其中令牌桶算法与漏桶算法主要用来限制请求处理的速度，可将其归为限速，计数器算法则是用来限制一个时间窗口内请求处理的数量，可将其归为限量（对速度不限制）。Guava 的 RateLimiter 是令牌桶算法的一种实现，但 RateLimiter 只适用于单机应用，在分布式环境下就不适用了。虽然已有一些开源项目可用于分布式环境下的限流管理，如阿里的Sentinel，但对于小型项目来说，引入Sentinel可能显得有点过重，但限流的需求在小型项目中也是存在的，下一篇文章就介绍下基于 RateLimiter 的分布式下的限流实现。

---

[转载请注明出处]
作者：雨歌
欢迎关注作者公众号：半路雨歌，查看更多技术干货文章