Guava (一) ：RateLimiter

public class SemaphoreTest {
    static Semaphore semaphore = new Semaphore(10);
    static ExecutorService executor = Executors.newFixedThreadPool(20);
    public static void Test(){
        for (int i = 0; i < 20; i++) {
            int finalI = i;
            executor.execute(()->{
                try {
                    semaphore.acquire();
                    System.out.println("成员" + finalI + " 过来了");
                    TimeUnit.SECONDS.sleep(10);
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

        }
    }

    public static void main(String[] args) {
        Test();
    }
}

public class RateLimiterTest {
    final RateLimiter rateLimiter = RateLimiter.create(2.0);

    //每秒 2 个许可
    public void submitTasks(List<Runnable> tasks, Executor executor) {
        for (Runnable task : tasks) {
            rateLimiter.acquire(10);
            executor.execute(task);
        }
    }

    public void test() {
        double acquire = rateLimiter.acquire(10);
        System.out.println(acquire);
    }
    public static void main(String[] args) {
        RateLimiterTest t = new RateLimiterTest();
        t.test();
        t.test();
    }
}
//输出
0.0
4.997451

public abstract class RateLimiter {
  //SmoothBursty 模式
  public static RateLimiter create(double permitsPerSecond) {
    return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
  }
 
  //SmoothWarmingUp模式
  public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
    checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
    return create(
        permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
  }
  
  //获取令牌
  public double acquire() {}
  public double acquire(int permits) {}

  //是否能获取到令牌
  public boolean tryAcquire() {}
  public boolean tryAcquire(int permits) {}
  public boolean tryAcquire(long timeout, TimeUnit unit) {}
  public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {}
 
  //获取当前的速率 以及动态的设置当前的速率
  public final double getRate() {}
  public final void setRate(double permitsPerSecond) {}
}

//计时，RateLimiter 把实例化的时间设置为 0 值，后续都是取相对时间，用微秒表示。
private final SleepingStopwatch stopwatch;

//用来做锁，RateLimiter 依赖于 synchronized 来控制并发，各个属性甚至都没有用 volatile 修饰。
private volatile Object mutexDoNotUseDirectly;

// 当前还有多少 permits 没有被使用，被存下来的 permits 数量
double storedPermits;

// 最大允许缓存的 permits 数量，也就是 storedPermits 能达到的最大值
double maxPermits;

// 每隔多少时间产生一个 permit，
// 比如我们构造方法中设置每秒 5 个，也就是每隔 200ms 一个，这里单位是微秒，也就是 200,000
double stableIntervalMicros;

// 下一次可以获取 permits 的时间，这个时间是相对 RateLimiter 的构造时间的，是一个相对时间，理解为时间戳吧
private long nextFreeTicketMicros = 0L; 

//SmoothBursty 模式
public static RateLimiter create(double permitsPerSecond) {
    return create(permitsPerSecond, SleepingStopwatch.createFromSystemTimer());
  }

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

public final void setRate(double permitsPerSecond) {
    //<1> 检查参数
    checkArgument(
        permitsPerSecond > 0.0 && !Double.isNaN(permitsPerSecond), "rate must be positive");
    synchronized (mutex()) {
      //<2> 子类实现
      doSetRate(permitsPerSecond, stopwatch.readMicros());
    }
  }

//抽象方法 由子类实现
abstract void doSetRate(double permitsPerSecond, long nowMicros);

final void doSetRate(double permitsPerSecond, long nowMicros) {
    //<1> 同步
    resync(nowMicros);
    //<2> 计算属性 stableIntervalMicros  每生成一个permit 需要多少微妙
    double stableIntervalMicros = SECONDS.toMicros(1L) / permitsPerSecond;
    this.stableIntervalMicros = stableIntervalMicros;
    <3> 子类实现
    doSetRate(permitsPerSecond, stableIntervalMicros);
  }

//SmoothRateLimiter.resync
void resync(long nowMicros) {
    // 如果 nextFreeTicket 已经过掉了，想象一下很长时间都没有再次调用 limiter.acquire() 的场景
    // 需要将 nextFreeTicket 设置为当前时间，重新计算 storedPermits
    if (nowMicros > nextFreeTicketMicros) {
      //计算的新permits
      double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
      //给storedPermits 赋值
      storedPermits = min(maxPermits, storedPermits + newPermits);
      nextFreeTicketMicros = nowMicros;
    }
  }

void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
  double oldMaxPermits = this.maxPermits;
  //<1> 这里计算了，maxPermits 为 1 秒产生的 permits
  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 的值域变化了，需要等比例缩放
    storedPermits =
        (oldMaxPermits == 0.0)
            ? 0.0 // initial state
            : storedPermits * maxPermits / oldMaxPermits;
  }
}

//默认是获取一个
public double acquire() {
  return acquire(1);
}

public double acquire(int permits) {
  //<1> 预约，如果当前不能直接获取到 permits，需要等待
  // 返回值代表需要 sleep 多久
  long microsToWait = reserve(permits);
  //<2> sleep
  stopwatch.sleepMicrosUninterruptibly(microsToWait);
  // 返回 sleep 的时长
  return 1.0 * microsToWait / SECONDS.toMicros(1L);
}

//输出
0.0
4.997451

final long reserve(int permits) {
  //检查
  checkPermits(permits);
  //synchronized 保持同步
  synchronized (mutex()) {
    return reserveAndGetWaitLength(permits, stopwatch.readMicros());
  }
}

final long reserveAndGetWaitLength(int permits, long nowMicros) {
  // 返回 nextFreeTicketMicros
  long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
  // 计算时长
  return max(momentAvailable - nowMicros, 0);
}

final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  // 这里做一次同步，更新 storedPermits 和 nextFreeTicketMicros (如果需要)
  resync(nowMicros);
  // 返回值就是 nextFreeTicketMicros，注意刚刚已经做了 resync 了，此时它是最新的正确的值
  long returnValue = nextFreeTicketMicros;
  // storedPermits 中可以使用多少个 permits
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
  // storedPermits 中不够的部分
  double freshPermits = requiredPermits - storedPermitsToSpend;
  // 为了这个不够的部分，需要等待多久时间
  long waitMicros =
      storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) // 这部分固定返回 0
          + (long) (freshPermits * stableIntervalMicros);
  // 将 nextFreeTicketMicros 往前推
  this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
  // storedPermits 减去被拿走的部分
  this.storedPermits -= storedPermitsToSpend;
  return returnValue;
}

//SmoothWarmingUp 模式
public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit) {
    checkArgument(warmupPeriod >= 0, "warmupPeriod must not be negative: %s", warmupPeriod);
    return create(
        permitsPerSecond, warmupPeriod, unit, 3.0, SleepingStopwatch.createFromSystemTimer());
  }

double coolDownIntervalMicros() {
    return warmupPeriodMicros / maxPermits;
}

void resync(long nowMicros) {
  if (nowMicros > nextFreeTicketMicros) {
    // coolDownIntervalMicros 在这里使用
    double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();
    storedPermits = min(maxPermits, storedPermits + newPermits);
    nextFreeTicketMicros = nowMicros;
  }
}

void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
    double oldMaxPermits = maxPermits;
    // coldFactor 是固定的 3
    double coldIntervalMicros = stableIntervalMicros * coldFactor;
    // 这个公式我们上面已经说了
    thresholdPermits = 0.5 * warmupPeriodMicros / stableIntervalMicros;
    // 这个公式我们上面也已经说了
    maxPermits =
        thresholdPermits + 2.0 * warmupPeriodMicros / (stableIntervalMicros + coldIntervalMicros);
    // 计算那条斜线的斜率。数学知识，对边 / 临边
    slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits);
    if (oldMaxPermits == Double.POSITIVE_INFINITY) {
        // if we don't special-case this, we would get storedPermits == NaN, below
        storedPermits = 0.0;
    } else {
        storedPermits =
            (oldMaxPermits == 0.0)
                ? maxPermits // initial state is cold
                : storedPermits * maxPermits / oldMaxPermits;
    }
}

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) // 这部分固定返回 0
          + (long) (freshPermits * stableIntervalMicros);
  this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
  this.storedPermits -= storedPermitsToSpend;
  return returnValue;
}

long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
  double availablePermitsAboveThreshold = storedPermits - thresholdPermits;
  long micros = 0;
  // 如果右边梯形部分有 permits，那么先从右边部分获取permits，计算梯形部分的阴影部分的面积
  if (availablePermitsAboveThreshold > 0.0) {
    // 从右边部分获取的 permits 数量
    double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake);
    // 梯形面积公式：(上底+下底)*高/2
    double length =
        permitsToTime(availablePermitsAboveThreshold)
            + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake);
    micros = (long) (permitsAboveThresholdToTake * length / 2.0);
    permitsToTake -= permitsAboveThresholdToTake;
  }
  // 加上 长方形部分的阴影面积
  micros += (long) (stableIntervalMicros * permitsToTake);
  return micros;
}

// 对于给定的 x 值，计算 y 值
private double permitsToTime(double permits) {
  return stableIntervalMicros + permits * slope;
}