@Test
public void test_1(){
    // 1、创建一个RateLimiter，指定每秒放0.5个令牌（2秒放1个令牌
    RateLimiter rateLimiter = RateLimiter.create(0.5);

    int[] a = {1, 6, 2};
    for (int i = 0 ; i < a.length; i++){
        log.info("时间戳：{}, 获取 {} 个令牌需要 {}s", System.currentTimeMillis(), a[i], rateLimiter.acquire(a[i]));
    }

    // 1516166482561 acq 1: wait 0.0s
    // 1516166482563 acq 6: wait 1.997664s
    // 1516166484569 acq 2: wait 11.991958s
}

@GwtIncompatible
abstract class SmoothRateLimiter extends RateLimiter {
  //当前持有的令牌数
  double storedPermits;
  
  //令牌数上限
  double maxPermits;
  
  //生成一个令牌需要的时间
  double stableIntervalMicros;
  
  //下次请求能获取令牌的时间，是个时间戳
  private long nextFreeTicketMicros;

//根据指定的 QPS 数创建  RateLimiter
public static RateLimiter create(double permitsPerSecond) {
  return create(RateLimiter.SleepingStopwatch.createFromSystemTimer(), permitsPerSecond);
}

@VisibleForTesting
static RateLimiter create(RateLimiter.SleepingStopwatch stopwatch, double permitsPerSecond) {
  // 1、创建 SmoothBursty 限流器
  RateLimiter rateLimiter = new SmoothBursty(stopwatch, 1.0D /* maxBurstSeconds */);
  
  // 2、设置限流速率
  rateLimiter.setRate(permitsPerSecond);
  return rateLimiter;
}

final void doSetRate(double permitsPerSecond, long nowMicros) {
  this.resync(nowMicros);
  //生成一个令牌需要的时间 
  double stableIntervalMicros = (double)TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond;
  this.stableIntervalMicros = stableIntervalMicros;
  this.doSetRate(permitsPerSecond, stableIntervalMicros);
}

void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
  double oldMaxPermits = this.maxPermits;
  //设置最大令牌数（每秒最大生成令牌数）  this.maxBurstSeconds = 1 * 每秒生成令牌数
  this.maxPermits = this.maxBurstSeconds * permitsPerSecond;
  if (oldMaxPermits == 1.0D / 0.0) {
    
    this.storedPermits = this.maxPermits;
  } else {
    this.storedPermits = oldMaxPermits == 0.0D ? 0.0D : this.storedPermits * this.maxPermits / oldMaxPermits;
  }

}

public double acquire(int permits) {
  // 1、获取当前请求需要等待的时间（惰性计算）
  long microsToWait = this.reserve(permits);

  // 2、seleep.microToWait （时间窗口）
  this.stopwatch.sleepMicrosUninterruptibly(microsToWait);

  // 3、返回 microToWait 对应的秒级时间
  return 1.0D * (double)microsToWait / (double)TimeUnit.SECONDS.toMicros(1L);
}


final long reserve(int permits) {
  //检查参数是否ok
  checkPermits(permits);
  synchronized(this.mutex()) {
    //计算需要等待的诗句
    return this.reserveAndGetWaitLength(permits, this.stopwatch.readMicros());
  }
}

@Override
final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  //1.根据当前时间和预计下一秒时间判断有无新令牌产生，有则更新持有令牌数storedPermits 和 下次请求时间nextFreeTicketMicros
  resync(nowMicros); 
  long returnValue = nextFreeTicketMicros;

  //2.以下两句，根据请求需要的令牌数requiredPermits和storedPermits当前持有的令牌数storedPermits分别计算 持有令牌中可用的数量storedPermitsToSpend和需要预支的令牌数量freshPermits
  double storedPermitsToSpend = min(requiredPermits, this.storedPermits); 
  double freshPermits = requiredPermits - storedPermitsToSpend;

  //3.分别计算storedPermitsToSpend和freshPermits的等待时间
  long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
    + (long) (freshPermits * stableIntervalMicros);

  try {
    //4.更新nextFreeTicketMicros
    this.nextFreeTicketMicros = LongMath.checkedAdd(nextFreeTicketMicros, waitMicros); 
  } catch (ArithmeticException e) {
    this.nextFreeTicketMicros = Long.MAX_VALUE;
  }

  //5.更新storedPermits 
  this.storedPermits -= storedPermitsToSpend; 

  return returnValue;
}

@Test
public void test_2() throws InterruptedException {
  //预热模式,设置预热时间和QPS，即在正式acquire前，限流器已经持有 5 * 4 = 20个令牌
  RateLimiter rateLimiter = RateLimiter.create(5, 4000, TimeUnit.MILLISECONDS);
  for (int i = 1; i < 50; i++) {
    System.out.println(System.currentTimeMillis() + " acq " + i + ": wait " + rateLimiter.acquire() + "s");
    if (i == 15) {
      Thread.sleep(2000);
      System.out.println(System.currentTimeMillis() + " acq " + 15 + ": wait " + rateLimiter.acquire() + "s");
    }
  }
}

1552395652026 acq 1: wait 0.0s
1552395652028 acq 2: wait 0.578357s
1552395652612 acq 3: wait 0.533835s
1552395653151 acq 4: wait 0.495191s
1552395653649 acq 5: wait 0.457239s
1552395654110 acq 6: wait 0.41631s
1552395654528 acq 7: wait 0.377524s
1552395654912 acq 8: wait 0.334018s
1552395655248 acq 9: wait 0.298249s
1552395655550 acq 10: wait 0.256165s
1552395655808 acq 11: wait 0.217752s
1552395656028 acq 12: wait 0.197672s
1552395656231 acq 13: wait 0.19451s
1552395656429 acq 14: wait 0.196465s
1552395656630 acq 15: wait 0.195714s
1552395658834 acq 15: wait 0.0s
1552395658834 acq 16: wait 0.34158s
1552395659180 acq 17: wait 0.296628s
1552395659482 acq 18: wait 0.256914s
1552395659744 acq 19: wait 0.216517s
1552395659965 acq 20: wait 0.195077s
1552395660164 acq 21: wait 0.195953s
1552395660365 acq 22: wait 0.195196s
1552395660564 acq 23: wait 0.196015s
1552395660764 acq 24: wait 0.195972s

static final class SmoothWarmingUp extends SmoothRateLimiter {
  //预热时间
  private final long warmupPeriodMicros;

  //预热区斜率
  private double slope;

  //区分冷区期和稳定器的阈值
  private double thresholdPermits;

  //code inteval/ stable inteval 的固定数值 硬编码写死的是 3
  private double coldFactor;

@VisibleForTesting
static RateLimiter create(
  SleepingStopwatch stopwatch, double permitsPerSecond, long warmupPeriod, TimeUnit unit,
  double coldFactor) {
  //1.创建SmoothWarmingUp限流器
  RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);

  //2.设置限流速率
  rateLimiter.setRate(permitsPerSecond);
  return rateLimiter;
}


public final void setRate(double permitsPerSecond) {
  Preconditions.checkArgument(permitsPerSecond > 0.0D && !Double.isNaN(permitsPerSecond), "rate must be positive");
  synchronized(this.mutex()) {
    this.doSetRate(permitsPerSecond, this.stopwatch.readMicros());
  }
}

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

SmoothWarmingUp(
  SleepingStopwatch stopwatch, long warmupPeriod, TimeUnit timeUnit, double coldFactor) {
  super(stopwatch);

  //1.设置预热时间
  this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod); 

  //3.设置coldFactor为3
  this.coldFactor = coldFactor;
}

final void doSetRate(double permitsPerSecond, long nowMicros) {
  this.resync(nowMicros);
  // 如果QPS为5 则 stableIntervalMicros  = 1s/5 = 200ms
  double stableIntervalMicros = (double)TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond;
  this.stableIntervalMicros = stableIntervalMicros;
  this.doSetRate(permitsPerSecond, stableIntervalMicros);
}


@Override
void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
  double oldMaxPermits = maxPermits;

  //1.设置冷却期等待时间数值 coldIntervalMicros
  double coldIntervalMicros = stableIntervalMicros * coldFactor; 

  //2.设置冷却期的阈值，thresholdPermits 等于预热期产生令牌数的 一半 （4000000/200000 = 20 ， 20 * 0.5 = 10）
  thresholdPermits = 0.5 * warmupPeriodMicros / stableIntervalMicros;

  //3.设置持有令牌的最大值，为thresholdPermits的2倍 为20
  maxPermits = thresholdPermits + 2.0 * warmupPeriodMicros / (stableIntervalMicros + coldIntervalMicros);

  //4.设置预热区的斜率；纵坐标之差／横坐标之差 
  slope = (coldIntervalMicros - stableIntervalMicros) / (maxPermits - thresholdPermits);

  if (oldMaxPermits == Double.POSITIVE_INFINITY) {
    storedPermits = 0.0;
  } else {
    storedPermits = (oldMaxPermits == 0.0)
      ? maxPermits 
      : storedPermits * maxPermits / oldMaxPermits;
  }
}

@Override
long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {

  //1.获取当前持有令牌数和阈值的差值availablePermitsAboveThreshold
  double availablePermitsAboveThreshold = storedPermits - thresholdPermits;
  long micros = 0;

  //2.如果availablePermitsAboveThreshold>0,即当前持有令牌数>阈值，即到达冷区期；计算等待时间
  if (availablePermitsAboveThreshold > 0.0) {

    //3.计算WARM UP PERIOD部分计算的方法，这部分是一个梯形，梯形的面积计算公式是 “（上底 + 下底） * 高 / 2”
    double permitsAboveThresholdToTake = min(availablePermitsAboveThreshold, permitsToTake);
    micros = (long) (permitsAboveThresholdToTake * (permitsToTime(availablePermitsAboveThreshold)
                                                    + permitsToTime(availablePermitsAboveThreshold - permitsAboveThresholdToTake)) / 2.0);

    //4.剩余的令牌从 stable部分拿
    permitsToTake -= permitsAboveThresholdToTake;
  }
  //5.stable 部分令牌获取花费的时间
  micros += (stableIntervalMicros * permitsToTake);
  return micros;
}

if (获得许可) {
    执行代码块 1
} else {
    执行代码块 2
}

@Test
public void rateLimiterTest() {
  RateLimiter rateLimiter = RateLimiter.create(5);
  for (int i = 0; i < 10; i++) {
    //非阻塞方式获取令牌，立即返回成功或者失败
    if (rateLimiter.tryAcquire()) {
      System.out.println(LocalTime.now()+"拿到令牌通过");
    }
    else {
      System.out.println(LocalTime.now()+"未拿到令牌不通过");
    }
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  
}




// 尝试获取1个许可，如果获取到，则返回true，否则返回false
boolean tryAcquire();

// 尝试获取多个许可，如果获取到，则返回true，否则返回false
boolean tryAcquire(int permits)
// 示例tryAcquire(3)

// 在timeout时间内，尝试获取1个许可，如果获取到，则返回true，否则返回false
tryAcquire(Duration timeout);
// 示例：tryAcquire(Duration.ofSeconds(3))
tryAcquire(long timeout, TimeUnit unit);
// 示例：tryAcquire(3, TimeUnit.SECONDS);

// 在timeout时间内，尝试获取多个许可，如果获取到，则返回true，否则返回false
tryAcquire(int permits, Duration timeout)
tryAcquire(int permits, long timeout, TimeUnit unit)

@Test
public void testAcquire() {
  RateLimiter rateLimiter = RateLimiter.create(2);

  for (int i = 0; i < 10; i++) {
    double sleep = rateLimiter.acquire();
    System.out.println("now: " + LocalTime.now() + "  sleep: " + sleep);
  }
}

@Documented
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface EntryTimes {

    /** 方法允许进入的次数 */
    int value() default 1;
   
     /**  可以访问的前缀 url前缀  */
    String prefix() default "";
}

/**
 * 控制每个用户访问Controller方法的次数
 * Created by fengchuanbo on 2017/5/25.
 */
@Aspect
@Component
public class MethodEntryTimesLimitInterceptor {

  private static final String METHOD_CAN_ENTRY_TIMES_KEY = "method:entry:times:";

  @Resource
  private StringRedisTemplate stringRedisTemplate;

  /**
     * 需要有 EntryTimes 标注，并且第一个参数需要是 AuthUser 才可以
     */
  @Around("@annotation(com.duodian.admore.zhaobutong.annotation.EntryTimes)")
  public Object aroundAdvice(ProceedingJoinPoint pjp) throws Throwable {
    HttpServletRequest request = ((ServletRequestAttributes) 
                                  RequestContextHolder.getRequestAttributes()).getRequest();
    String token = request.getParameter("token");
    String aes = AesUtils.LoginDecrypt(token);
    Long userId = Long.valueOf(aes.split("#")[0]);
    MethodSignature signature = (MethodSignature) pjp.getSignature();
    Method method = signature.getMethod();
    EntryTimes annotation = method.getAnnotation(EntryTimes.class);
    int times = annotation.value();
    String key = METHOD_CAN_ENTRY_TIMES_KEY + ":" + annotation.prefix() + ":" +  userId;
    // 没有整个方法使用一个redis链接，是为了方式方法执行时间过长一直占用redis链接。
    Long increment = getEntryTimes(key);
    Object retVal;
    try {
      // 放在try里面，才能执行finally
      if (increment > times){
        // 设置十秒钟超时，防止finally在系统崩溃或重启时没有执行造成用户不能操作。
        expireKey(key,10);
        return Response.of(Code.ACTION_FREQUENT);
      }
      //调用核心逻辑
      retVal = pjp.proceed();
    }finally {
      deleteKey(key);
    }
    return retVal;
  }

  private Long getEntryTimes(String key){
    return stringRedisTemplate.opsForValue().increment(key,1);
  }

  private void deleteKey(String key){
    stringRedisTemplate.delete(key);
  }

  private void expireKey(String key, int seconds){
    stringRedisTemplate.boundValueOps(key).expire(seconds, TimeUnit.SECONDS);
  }
}

@Slf4j
public class SimpleRateLimiter {

    private Jedis jedis;

    public SimpleRateLimiter(Jedis jedis) {
        this.jedis = jedis;
    }

    /**
     *
     * @param userId 用户Id
     * @param actionKey 动作
     * @param period 过期时间 单位(秒)
     * @param maxCount 最大次数
     * @return 返回是否允许访问
     * 1、获取key = userId + "#" + actionKey;
     * 2、
     */
    public boolean isActionAllowed(String userId, String actionKey, int period, int maxCount) {
        // 1、获取key = userId + "#" + actionKey;
        String key = userId + "#" + actionKey;
        long now = System.currentTimeMillis();
        Pipeline pipe = jedis.pipelined();
        //开启事务
        pipe.multi();
        //第二个参数是 score，第三个参数是value
        pipe.zadd(key, now, "" + now);
        ////删除 now -period* 1000 毫秒之前的数据（也就是说保留periods秒之内的数据）
        // 看清楚了这是直接删除了
        pipe.zremrangeByScore(key, 0, now - period * 1000);
        //当前窗口的元素个数
        Response<Long> count = pipe.zcard(key);
        //一定要设置过期时间，可能大部分用户是冷用户，因为要维护period时间内的记录，所以key过期时间要稍微比period大
        pipe.expire(key, period + 1);
        //执行事务
        pipe.exec();
        try {
            pipe.close();
        } catch (IOException e) {
            log.info("SimpleRateLimiter#isActionAllowed pipe.close() error", e);
        }
      
         //通过统计滑动窗口内的行为数量与阈值 `max_count `进行比较就可以得出当前的行为是 允许。用代码表示如下:
        return count.get() <= maxCount;
    }

}

@Test
public void test() {
  Jedis jedis = new Jedis("127.0.0.1");
  SimpleRateLimiter limiter = new SimpleRateLimiter(jedis);
  for (int i = 0; i < 20; i++) {
    log.info("限流状态：[{}]", limiter.isActionAllowed("QYD", "reply", 60, 5));
  }

  try {
    Thread.sleep(10000);
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
}

@Data
public class FunnelDTO {
  /** 容量 */
  Integer capacity;
  /** 剩余容量 */
  Integer leftQuota;
  /** 流出速率 */
  Float leakingRate;
  /** 计算起始时间 */
  Long startTime;


  public FunnelDTO(int capacity, float leakingRate) {
    this.capacity = capacity;
    this.leftQuota = capacity;
    this.leakingRate = leakingRate;
    this.startTime = System.currentTimeMillis();
  }


  /**
     * 1、计算时间间隔
     * 2、计算时间间隔内流出的数量
     *  2.1、如果获取到的小于0， 说明间隔时间太长（没有往漏斗里放数据，漏斗空了），整数数字过大溢出了，则初始化数据
     * 2.2、如果获取到的流出量小于1，则说明腾出空间太小(最小单位是1)，则直接返回，不执行
     * 3、
     */
  public void makeSpace() {
    Long nowTime = System.currentTimeMillis();
    // 1、计算时间间隔
    Long deltaTs = nowTime - startTime;

    // 2、计算时间间隔内流出的数量
    Integer deltaQuota = (int) (deltaTs * leakingRate);
    // 2.1、如果获取到的小于0， 说明间隔时间太长（没有往漏斗里放数据，漏斗空了），整数数字过大溢出了，则初始化数据
    if (deltaQuota < 0) {
      this.leftQuota = capacity;
      this.startTime = nowTime;
      return;
    }
    // 2.2、如果获取到的流出量小于1，则说明腾出空间太小(最小单位是1)，则直接返回，不执行
    if (deltaQuota < 1) {
      return;
    }

    // 3、重新计算
    // 当前时间 = nowTime
    // 剩余容量 = 当前剩余容量 + 流出速率 * 间隔时间（如果超出了总容量，剩余容量 = 总容量）
    this.leftQuota += deltaQuota;
    this.startTime = nowTime;
    if (this.leftQuota > this.capacity) {
      this.leftQuota = this.capacity;
    }
  }

  /**
     * 判断是否能加入交易
     * 1、漏斗重新计算剩余容量和当前时间
     * 2、
     * @param quota 定额
     * @return
     */
  public boolean watering(int quota) {
    // 1、漏斗重新计算剩余容量和当前时间
    makeSpace();

    // 2、如果剩余容量大于当前要流入漏斗的量，则执行成功，返回true
    if (this.leftQuota >= quota) {
      this.leftQuota -= quota;
      return true;
    }
    return false;
  }

}

public class FunnelRateLimiter {


  private Map<String, FunnelDTO> funnels ;
  public FunnelRateLimiter(Map funnels) {
    this.funnels = funnels;
  }


  /**
     * 限流方法
     * @param userId
     * @param actionKey
     * @param capacity
     * @param leakingRate
     * @return
     * 1、根据用户Id和动作获取对应的漏斗
     * 2、往漏斗中放入定额数据，看是否能放下
     */
  public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
    // 1、根据用户Id和动作获取对应的漏斗
    String key = String.format("%s:%s", userId, actionKey);
    FunnelDTO funnel = funnels.get(key);
    if (funnel == null) {
      funnel = new FunnelDTO(capacity, leakingRate);
      funnels.put(key, funnel);
    }

    //2、往漏斗中放入定额数据
    return funnel.watering(1);
  }


}

package com.healerjean.proj.aspect;

import com.google.common.util.concurrent.RateLimiter;
import com.healerjean.proj.util.limit.RateLimiterPropertyUtil;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * RateLimiterAspect
 *
 * @author zhangyujin
 * @date 2022/9/8  21:44.
 */
@Component
@Aspect
public class RateLimiterAspect {
    /**
     * log
     */
    private static final Logger log = LoggerFactory.getLogger(RateLimiterAspect.class);

    /**
     *
     */
    private RateLimiterAspect() {
    }


    /**
     * @param pjp
     * @return
     * @throws Throwable
     */
    @Around("execution(* com.healerjean.proj.controller.*Controller.*(..))")
    public Object logAround(ProceedingJoinPoint pjp) throws Throwable {
        Object[] args = pjp.getArgs();
        Class<?> clazz = pjp.getTarget().getClass();
        log.info("开始拦截:" + clazz.getName());

        try {
            this.limit(clazz.getName());
        } catch (Exception var5) {
            log.error(var5.getMessage(), var5);
        }

        return pjp.proceed(args);
    }

    /**
     *
     * @param className
     */
    public void limit(String className) {
        RateLimiter rateLimiter = RateLimiterPropertyUtil.getMap().get(className);
        if (null != rateLimiter) {
            log.info("开始获取令牌:" + className + " rate:" + rateLimiter.getRate());
            double waitTime = rateLimiter.acquire();
            log.info(className + ":获取成功 waitTime :" + waitTime + "ms");
        }
    }
}

package com.healerjean.proj.util.limit;

import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RateLimiterPropertyUtil
 *
 * @author zhangyujin
 * @date 2022/9/8  21:37.
 */
@Service
@Slf4j
public class RateLimiterPropertyUtil {

    /**
     * MAP_RATE
     */
    private static ConcurrentHashMap<String, RateLimiter> MAP_RATE = new ConcurrentHashMap();
    /**
     * props
     */
    private static Properties PROPS;

    private RateLimiterPropertyUtil() {
    }

    @PostConstruct
    public void init(){
        load();
    }

    private static synchronized void load() {
        log.info("开始加载限流文件内容.......");
        PROPS = new Properties();
        InputStream in = null;
        try {
            in = RateLimiterPropertyUtil.class.getResourceAsStream("/rateLimiter.properties");
            PROPS.load(in);
            Set<Map.Entry<Object, Object>> set = PROPS.entrySet();

            for (Map.Entry<Object, Object> objectObjectEntry : set) {
                Map.Entry<Object, Object> entry = objectObjectEntry;
                String key = String.valueOf(entry.getKey()).trim();
                String value = String.valueOf(entry.getValue());
                log.info("限流策略：key:{}, value:{}", key, value);
                if (StringUtils.hasText(key) && StringUtils.hasText(value)) {
                    RateLimiter rateLimiter = RateLimiter.create(Double.valueOf(value));
                    MAP_RATE.put(key, rateLimiter);
                }
            }
        } catch (FileNotFoundException var17) {
            log.error("jdbc.properties文件未找到");
        } catch (IOException var18) {
            log.error("出现IOException");
        } finally {
            try {
                if (null != in) {
                    in.close();
                }
            } catch (IOException var16) {
                log.error("jdbc.properties文件流关闭出现异常");
            }
        }

        log.info("properties文件内容：" + PROPS);
    }

    public static ConcurrentHashMap<String, RateLimiter> getMap() {
        return MAP_RATE;
    }


}