在高并发系统中，为了保证系统的稳定，经常需要对容量做一些控制，如限流来防止超出系统的性能瓶颈。限流算法是限流中比较经典的算法之一，Guava的RateLimiter基于令牌桶算法实现，并且十分方便。
这里基于RateLimiter学习下令牌桶算法
概念
image.png
令牌桶：存储固定数量的令牌，令牌通过固定的速率加入其中，超出数量的令牌进行丢弃。
当请求处理的时候，只有获得令牌的请求才能被处理，并且从令牌桶中去除对应的令牌数量。无法获得令牌(令牌消耗完毕)的请求会被限制，被丢弃或者加入等待队列。RateLimiter则是等待，而非丢弃。
RateLimiter
先贴一段Demo代码，理解RateLimiter如何使用，以及其内部的关键概念
package me.aihe;
import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.RateLimiter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
public class RaceLimiterDemo {
public static void main(String[] args) throws InterruptedException {
rateLimiterTest();
// stopWtachElapsedTest();
}
private static void stopWtachElapsedTest() throws InterruptedException {
Stopwatch stopwatch = Stopwatch.createStarted();
for (int i = 0; i < 10; i++) {
Thread.sleep(1000);
long elapsed = stopwatch.elapsed(MICROSECONDS);
System.out.println(elapsed);
}
}
private static void rateLimiterTest() throws InterruptedException {
int threadNum = 10;
ExecutorService threadPool = Executors.newFixedThreadPool(threadNum);
RateLimiter limiter = RateLimiter.create(1);
ThreadLocal<SimpleDateFormat> formatThreadLocal = ThreadLocal.withInitial(() -> new SimpleDateFormat("hh:mm:ss:SS"));
for (int i = 0; i < threadNum; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
limiter.acquire();
SimpleDateFormat simpleDateFormat = formatThreadLocal.get();
String currentFormattedDate = simpleDateFormat.format(new Date());
System.out.println(currentFormattedDate);
}
});
}
Thread.sleep(5 * 1000);
threadPool.shutdown();
}
}
抽象
// 当前存储了多少个令牌，可以理解为令牌桶算法的令牌桶
double storedPermits;
// 令牌桶最多可以存储多少个令牌
double maxPermits;
// 添加令牌的时间间隔，比如每秒5个令牌，值为200ms，其计算方式如下
// double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
double stableIntervalMicros;
// 下次计算令牌的时间，可以是当前值，也可能是未来的一个值。
// 比如当下如果为10，其值可能为15大于当前的值。
private long nextFreeTicketMicros = 0L;
更新令牌
resync(nowMicros)方法更新令牌数量.
acquire()方法会消耗令牌，其过程中同时会调用resync令牌数量。
acquire流程
1 获取排它锁，使得线程可以独立执行。
2 更新当前令牌数量，resync增加令牌。
3 计算本次需要消耗的令牌
- 如果要消耗的令牌数量大于令牌桶的令牌数。那么等待令牌差值 * 添加令牌时间间隔时间。
- 如果消耗的令牌数量小于桶中令牌数，无需等待
4 减去消耗的令牌数量。
5 线程等待第三步计算的需要等待的时间。
//1. 排它锁
final long reserve(int permits) {
checkPermits(permits);
synchronized (mutex()) {
return reserveAndGetWaitLength(permits, stopwatch.readMicros());
}
}
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
// 2 更新令牌数量
resync(nowMicros);
long returnValue = nextFreeTicketMicros;
// 3 消耗的令牌数量不能大于令牌桶中的令牌总数，因此取 min 值
double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
// 剩下的令牌数量为多少，如果消耗令牌数小于桶中令牌数，其值为0.
double freshPermits = requiredPermits - storedPermitsToSpend;
long waitMicros =
storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
+ (long) (freshPermits * stableIntervalMicros);
// 计算下一次更新令牌时间
this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
// 4 减去消耗令牌数，更新令牌桶的数量
this.storedPermits -= storedPermitsToSpend;
return returnValue;
}
void resync(long nowMicros) {
// if nextFreeTicket is in the past, resync to now
if (nowMicros > nextFreeTicketMicros) {
// 新的令牌数。(此刻 - 上次更新令牌时间) / 更新令牌时间间隔
double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
// 令牌桶数量，存储的令牌加上新生成的令牌数，但是其最大值不超过设置的maxPermits
storedPermits = min(maxPermits, storedPermits + newPermits);
nextFreeTicketMicros = nowMicros;
}
}
SleepingStopwatch
// 计算从生成StopWtach对象到此刻运行的总时间
protected long readMicros() {
return stopwatch.elapsed(MICROSECONDS);
}
private static void stopWtachElapsedTest() throws InterruptedException {
Stopwatch stopwatch = Stopwatch.createStarted();
for (int i = 0; i < 10; i++) {
Thread.sleep(1000);
long elapsed = stopwatch.elapsed(MICROSECONDS);
System.out.println(elapsed);
}
}
// 运算结果：从生成Stopwatch对象，到每次计算elapsed的时间。
1004892
2008624
3008745
4011957
5015946
6016365
关键思路
- 每次在消耗令牌之间，调用生成令牌逻辑。减少了要另起一个线程生成令牌的消耗。
- 新建RateLimiter对象时，通过doSetRate()初始化关键的属性，桶数量，生成令牌速率等
- 在上图中等待是通过Thread.sleep方式实现的。其中如果令牌数量不够的时候，内部并不是等待生成令牌之后再执行下一步。而是先更新令牌为负数，然后再进行sleep，令牌负数越大，后续请求等待的时间越长，它假设了令牌的生成速率是相同的。
使用场景
在我自己负责的业务系统中，通过RateLimiter控制调用下游的QPS，比如下游平台只提供了5W QPS的调用量，那么计算5W QPS分摊到每台机器上可以调用的QPS有多少。通过配置中心调整RateLimiter的限流速度，进而控制调用下游的QPS。
