令牌桶算法与RateLimiter

在高并发系统中，为了保证系统的稳定，经常需要对容量做一些控制，如限流来防止超出系统的性能瓶颈。限流算法是限流中比较经典的算法之一，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  formatThreadLocal = ThreadLocal.withInitial(() -> new SimpleDateFormat("hh:mm:ss:SS"));

        for (int i = 0; i

抽象

// 当前存储了多少个令牌，可以理解为令牌桶算法的令牌桶
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

关键思路

每次在消耗令牌之间，调用生成令牌逻辑。减少了要另起一个线程生成令牌的消耗。
新建RateLimiter对象时，通过doSetRate()初始化关键的属性，桶数量，生成令牌速率等
在上图中等待是通过Thread.sleep方式实现的。其中如果令牌数量不够的时候，内部并不是等待生成令牌之后再执行下一步。而是先更新令牌为负数，然后再进行sleep，令牌负数越大，后续请求等待的时间越长，它假设了令牌的生成速率是相同的。

使用场景

在我自己负责的业务系统中，通过RateLimiter控制调用下游的QPS，比如下游平台只提供了5W QPS的调用量，那么计算5W QPS分摊到每台机器上可以调用的QPS有多少。通过配置中心调整RateLimiter的限流速度，进而控制调用下游的QPS。

概念