前言

最近在学习限流算法，想把四种限流算法整合进入Dubbo里，于是用Dubbo的Filter来实现限流。

Dubbo版本

  1. <groupId>org.apache.dubbo</groupId>
  2.     <artifactId>dubbo</artifactId>
  3. <version>2.7.7</version>

准备工作

构建三个模块，dubbo-api(dubbo暴露的接口),dubbo-provider（dubbo提供者）,dubbo-consumer（dubbo消费者）

dubbo-api

  1. public interface DemoService {
  3.     String sayHello(String name);
  5. }

dubbo-provider

  1. @DubboService
  2. public class DemoServiceImpl implements DemoService {
  5.     @Value("${dubbo.application.name}")
  6.     private String serviceName;
  8.     @Override
  9.     public String sayHello(String name) {
  10.         return String.format("[%s] : Hello, %s", serviceName, name);
  11.     }
  13. }

Dubbo-consumer

  1. @RestController
  2. @RequestMapping("/test")
  3. @Api(tags = "Dubbo服务测试类")
  4. public class DemoController {
  6.       /**
  7.      * 为了调试限流,关闭超时报错,关闭重试机制
  8.      */
  9.     @DubboReference(timeout = 1000000,retries=0)
  10.     private DemoService demoService;
  12.     @ApiOperation(value = "服务调用测试", notes = "")
  13.     @GetMapping("sayHello")
  14.     public String sayHello() {
  15.         String name = demoService.sayHello("ZywooLee-debug");
  16.         return name;
  17.     }
  20. }

固定窗口限流算法（Dubbo内嵌）

Dubbo内嵌了令牌桶限流了，不过默认并没有开启，需要在服务提供者里实现SPI，在dubbo-provider/src/main/resources/META-INF/dubbo路径下创建文件org.apache.dubbo.rpc.Filter

文件内容：

  1. tps=org.apache.dubbo.rpc.filter.TpsLimitFilter

并在服务提供者上设置配置，filter会按照数组里配置的先后顺序执行（责任链设计模式），这里指定了tps（也就是SPI设置文件里的tps）

  1. //选择Dubbo令牌限流算法
  2. @DubboService(filter = {"tps"},parameters = {"tps","2","tps.interval","1000"})、
  3. public class DemoServiceImpl implements DemoService {
  4.     //...
  5. }

接下来查看我们设置的org.apache.dubbo.rpc.filter.TpsLimitFilter

  1. //TPS_LIMIT_RATE_KEY为“tps”，在org.apache.dubbo.rpc.Filter文件里指定的tps对应这里
  2. @Activate(group = CommonConstants.PROVIDER, value = TPS_LIMIT_RATE_KEY)
  3. public class TpsLimitFilter implements Filter {
  5.     private final TPSLimiter tpsLimiter = new DefaultTPSLimiter();
  7.     @Override
  8.     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
  10.         if (!tpsLimiter.isAllowable(invoker.getUrl(), invocation)) {
  11.             throw new RpcException(
  12.                     "Failed to invoke service " +
  13.                             invoker.getInterface().getName() +
  14.                             "." +
  15.                             invocation.getMethodName() +
  16.                             " because exceed max service tps.");
  17.         }
  19.         return invoker.invoke(invocation);
  20.     }
  22. }

可以看到限流的是在org.apache.dubbo.rpc.filter.tps.DefaultTPSLimiter处理的

  1. public class DefaultTPSLimiter implements TPSLimiter {
  3.       //key->接口名称（如:com.dubbo.api.DemoService） value：StatItem类（令牌桶算法实现）
  4.     private final ConcurrentMap<String, StatItem> stats = new ConcurrentHashMap<String, StatItem>();
  6.     @Override
  7.     public boolean isAllowable(URL url, Invocation invocation) {
  8.           // 窗口内限制请求数
  9.         int rate = url.getParameter(TPS_LIMIT_RATE_KEY, -1);
  10.           // 窗口长度（单位：ms）
  11.         long interval = url.getParameter(TPS_LIMIT_INTERVAL_KEY, DEFAULT_TPS_LIMIT_INTERVAL);
  12.           // 服务名称
  13.         String serviceKey = url.getServiceKey();
  14.         if (rate > 0) {
  15.               // 从缓存中获取限流实现对象
  16.             StatItem statItem = stats.get(serviceKey);
  17.             if (statItem == null) {
  18.                 stats.putIfAbsent(serviceKey, new StatItem(serviceKey, rate, interval));
  19.                 statItem = stats.get(serviceKey);
  20.             } else {
  21.                 //限制请求数改变，重新构建限流实现对象
  22.                 if (statItem.getRate() != rate || statItem.getInterval() != interval) {
  23.                     stats.put(serviceKey, new StatItem(serviceKey, rate, interval));
  24.                     statItem = stats.get(serviceKey);
  25.                 }
  26.             }
  27.             return statItem.isAllowable();
  28.         } else {
  29.               // 限流数小于等于0，remove限流实现对象，放行
  30.             StatItem statItem = stats.get(serviceKey);
  31.             if (statItem != null) {
  32.                 stats.remove(serviceKey);
  33.             }
  34.         }
  36.         return true;
  37.     }
  39. }

限流的实现类是org.apache.dubbo.rpc.filter.tps.StatItem

  1. class StatItem {
  3.       //服务名称
  4.     private String name;
  6.       //上一次访问时间
  7.     private long lastResetTime;
  9.       //窗口长度（单位：ms）
  10.     private long interval;
  12.       //计数器
  13.     private LongAdder token;
  15.       //窗口内限制请求数
  16.     private int rate;
  18.     StatItem(String name, int rate, long interval) {
  19.         this.name = name;
  20.         this.rate = rate;
  21.         this.interval = interval;
  22.         this.lastResetTime = System.currentTimeMillis();
  23.         this.token = buildLongAdder(rate);
  24.     }
  26.     public boolean isAllowable() {
  27.         long now = System.currentTimeMillis();
  28.           //当前时间戳>上次访问时间+窗口长度，计数器初始化为限制请求数
  29.         if (now > lastResetTime + interval) {
  30.             token = buildLongAdder(rate);
  31.             lastResetTime = now;
  32.         }
  34.           //计数器小于0，此时间窗口内访问数量已用完
  35.         if (token.sum() < 0) {
  36.             return false;
  37.         }
  38.           //计数器次数-1
  39.         token.decrement();
  40.         return true;
  41.     }
  43.     public long getInterval() {
  44.         return interval;
  45.     }
  48.     public int getRate() {
  49.         return rate;
  50.     }
  53.     long getLastResetTime() {
  54.         return lastResetTime;
  55.     }
  57.     long getToken() {
  58.         return token.sum();
  59.     }
  61.     @Override
  62.     public String toString() {
  63.         return new StringBuilder(32).append("StatItem ")
  64.                 .append("[name=").append(name).append(", ")
  65.                 .append("rate = ").append(rate).append(", ")
  66.                 .append("interval = ").append(interval).append("]")
  67.                 .toString();
  68.     }
  70.     private LongAdder buildLongAdder(int rate) {
  71.         LongAdder adder = new LongAdder();
  72.         adder.add(rate);
  73.         return adder;
  74.     }
  76. }

自定义限流实现

因为是自行实现的限流，所以需要先扩展SPI，首先我先定义一个自己的ZywooLimitFilter，并在org.apache.dubbo.rpc.Filter文件里添加配置

  1. #这个是原先Dubbo提供的TPSLimit
  2. tps=org.apache.dubbo.rpc.filter.TpsLimitFilter
  3. #这个是我自行实现的Filter
  4. zywoo=com.dubbo.filter.ZywooLimitFilter

与org.apache.dubbo.rpc.filter.TpsLimitFilter基本无异，只是指定了@Activate的value,还有将TPSLimiter的实现改为ZywooTPSLimiter,代码如下

  1. package com.dubbo.filter;
  3. import org.apache.dubbo.common.constants.CommonConstants;
  4. import org.apache.dubbo.common.extension.Activate;
  5. import org.apache.dubbo.rpc.*;
  6. import org.apache.dubbo.rpc.filter.tps.TPSLimiter;
  8. /**
  9.  * @Auther: Zywoo Lee
  10.  * @Date: 2022/4/1 20:46
  11.  * @Description: 自定义Filter实现
  12.  */
  13. @Activate(group = CommonConstants.PROVIDER, value = "zywoo")
  14. public class ZywooLimitFilter implements Filter {
  16.     //Dubbo默认实现
  17.     //private final TPSLimiter tpsLimiter = new DefaultTPSLimiter();
  19.     //自定义实现
  20.     private final TPSLimiter tpsLimiter = new ZywooTPSLimiter();
  22.     @Override
  23.     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
  24.         if (!tpsLimiter.isAllowable(invoker.getUrl(), invocation)) {
  25.             throw new RpcException(
  26.                     "Failed to invoke service " +
  27.                             invoker.getInterface().getName() +
  28.                             "." +
  29.                             invocation.getMethodName() +
  30.                             " because exceed max service tps.");
  31.         }
  33.         return invoker.invoke(invocation);
  34.     }
  35. }

com.dubbo.filter.ZywooLimitFilter集成了剩余三种限流算法

  1. public class ZywooTPSLimiter implements TPSLimiter {
  3.     private final ConcurrentMap<String, LimitItem> stats = new ConcurrentHashMap<>();
  5.     @Override
  6.     public boolean isAllowable(URL url, Invocation invocation) {
  7.         String serviceKey = url.getServiceKey();
  8.         //选择限流的策略
  9.         String strategy = url.getParameter("strategy");
  10.         //获取限流实现类
  11.         LimitItem limitItem = stats.get(serviceKey);
  12.         if (limitItem == null) {
  13.             switch (strategy) {
  14.                 //漏桶限流算法
  15.                 case "leakyBucket": {
  16.                     long rate = url.getParameter("rate", -1L);
  17.                     long capacity = url.getParameter("capacity", -1L);
  18.                     long currentTime = System.currentTimeMillis();
  19.                     stats.putIfAbsent(serviceKey, new LeakyBucketItem(serviceKey, rate, 0L, currentTime, capacity));
  20.                     break;
  21.                 }
  22.                 //固定窗口限流算法
  23.                 case "fixedWindow": {
  24.                     //默认1000毫秒滑动一次
  25.                     long windowUnit = url.getParameter("windowUnit", 1000L);
  26.                     //获取限流次数
  27.                     int limitCount = url.getParameter("limitCount", 0);
  28.                     long currentTime = System.currentTimeMillis();
  29.                     stats.putIfAbsent(serviceKey, new FixedWindowItem(serviceKey, windowUnit, currentTime, 0, limitCount));
  30.                 }
  31.                 //滑动窗口限流算法
  32.                 case "slidingWindow": {
  33.                     //限流个数
  34.                     int limitCount = url.getParameter("limitCount", 1);
  35.                     //时间窗数量
  36.                     int sampleCount = url.getParameter("sampleCount", 10);
  37.                     //时间窗长度
  38.                     int intervalInMs = url.getParameter("intervalInMs", 1000);
  39.                     stats.putIfAbsent(serviceKey,new SlidingWindowItem(limitCount,sampleCount,intervalInMs));
  40.                 }
  41.                 //无获取到限流算法,直接放行
  42.                 default: {
  43.                     return true;
  44.                 }
  45.             }
  46.             limitItem = stats.get(serviceKey);
  47.         }
  48.         return limitItem.isAllowed();
  49.     }
  50. }

滑动窗口限流算法（参考Sentinel）

滑动窗口算法我参考了sentinel来进行实现。主要是将WindowWrap里的value固定为LongAdder只记录小时间窗（时间窗长度/时间窗个数）里的访问次数（Sentinel里的value实际上是MetricBucket类，内部维护了LongAdder数组，来去记录这个时间里通过、阻塞、报错、限流的次数分别多少我这边精简实现，记录通过数量即可）

滑动窗口的实现com.dubbo.limit.slidingWindow.LeapArray

  1. /**
  2.  * @Auther: Zywoo Lee
  3.  * @Date: 2022/4/2 15:57
  4.  * @Description: 滑动窗口限流算法
  5.  */
  6. public class LeapArray implements LimitItem {
  7.     // 样本窗口长度
  8.     private int windowLengthInMs;
  9.     // 一个时间窗中包含的时间窗数量
  10.     private int sampleCount;
  11.     // 时间窗长度
  12.     private int intervalInMs;
  13.     //元素为WindowWrap样本窗口
  14.     private final AtomicReferenceArray<WindowWrap> array;
  16.     //修改锁
  17.     private final ReentrantLock updateLock = new ReentrantLock();
  19.     /**
  20.      *
  21.      * 功能描述: 构造函数
  22.      *
  23.      * @param: sampleCount 时间窗数量 intervalInMs 时间窗长度
  24.      * @return:
  25.      * @auther: Zywoo Lee
  26.      * @date: 2022/4/5 14:19
  27.      */
  28.     public LeapArray(int sampleCount, int intervalInMs){
  29.         this.windowLengthInMs = intervalInMs / sampleCount;
  30.         this.intervalInMs = intervalInMs;
  31.         this.sampleCount = sampleCount;
  32.         this.array = new AtomicReferenceArray(sampleCount);
  33.     }
  35.     public WindowWrap currentWindow(){
  36.         // 获取当前时间点所在的样本窗口
  37.         return currentWindow(System.currentTimeMillis());
  38.     }
  41.     @Override
  42.     public boolean isAllowed() {
  43.         //获取当前时间段在哪个滑动窗口
  45.         return true;
  46.     }
  49.     public WindowWrap currentWindow(long timeMillis){
  50.         if (timeMillis < 0) {
  51.             return null;
  52.         }
  54.         // 计算当前时间所在的样本窗口id，即在计算数组LeapArray中的索引
  55.         int idx = calculateTimeIdx(timeMillis);
  57.         // 计算当前样本窗口的开始时间点
  58.         long windowStart = calculateWindowStart(timeMillis);
  60.         while(true){
  61.             WindowWrap old = array.get(idx);
  62.             // 若当前时间所在样本窗口为null，说明该样本窗口还不存在，则创建一个
  63.             if (old==null){
  64.                 WindowWrap window = new WindowWrap(windowLengthInMs, windowStart,new LongAdder());
  65.                 if (array.compareAndSet(idx,null,window)){
  66.                     return window;
  67.                 }else{
  68.                     Thread.yield();
  69.                 }
  70.             }
  71.             // 若当前样本窗口的起始时间点与计算出的样本窗口起始时间点相同，
  72.             // 则说明这两个是同一个样本窗口
  73.             else if(windowStart==old.windowStart()){
  74.                 return old;
  75.             }
  76.             // 若当前样本窗口的起始时间点 大于 计算出的样本窗口起始时间点，
  77.             // 说明计算出的样本窗口已经过时了，需要将原来的样本窗口替换
  78.             else if (windowStart > old.windowStart()){
  79.                 if (updateLock.tryLock()) {
  80.                     try {
  81.                         // 替换掉老的样本窗口
  82.                         return resetWindowTo(old, windowStart);
  83.                     } finally {
  84.                         updateLock.unlock();
  85.                     }
  86.                 } else {
  87.                     // Contention failed, the thread will yield its time slice to wait for bucket available.
  88.                     Thread.yield();
  89.                 }
  90.             }
  91.             // 当前样本窗口的起始时间点 小于 计算出的样本窗口起始时间点，
  92.             // 这种情况一般不会出现，因为时间不会倒流。除非人为修改了系统时钟
  93.             else if (windowStart < old.windowStart()){
  94.                 return new WindowWrap(windowLengthInMs, windowStart,new LongAdder());
  95.             }
  96.         }
  97.     }
  100.     public int calculateTimeIdx(long timeMillis) {
  101.         // 计算出当前时间在哪个样本窗口
  102.         long timeId = timeMillis / windowLengthInMs;
  103.         // Calculate current index so we can map the timestamp to the leap array.
  104.         return (int)(timeId % array.length());
  105.     }
  107.     private long calculateWindowStart(long timeMillis) {
  108.         return timeMillis - timeMillis % windowLengthInMs;
  109.     }
  112.     private WindowWrap resetWindowTo(WindowWrap w, long startTime) {
  113.         System.out.println("重置时间窗");
  114.         // 更新窗口起始时间
  115.         w.resetTo(startTime);
  116.         // 将访问次数数据清零
  117.         w.value().reset();
  118.         return w;
  119.     }
  122.     public List values() {
  123.         return values(System.currentTimeMillis());
  124.     }
  126.     public List values(long timeMillis) {
  127.         if (timeMillis < 0) {
  128.             return new ArrayList();
  129.         }
  130.         int size = array.length();
  131.         List result = new ArrayList(size);
  133.         // 逐个遍历array中的每一个样本窗口实例
  134.         for (int i = 0; i < size; i++) {
  135.             WindowWrap windowWrap = array.get(i);
  136.             // 若当前遍历实例为空或已经过时，则继续下一个
  137.             if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
  138.                 continue;
  139.             }
  140.             // 将当前遍历的样本窗口统计的数据记录到result中
  141.             result.add(windowWrap.value());
  142.         }
  143.         return result;
  144.     }
  146.     public boolean isWindowDeprecated(long time, WindowWrap windowWrap) {
  147.         // 当前时间与当前样本窗口的时间差 大于 时间窗长度，
  148.         // 说明当前样本窗口已经过时
  149.         return time - windowWrap.windowStart() > intervalInMs;
  150.     }
  152.     public int getWindowLengthInMs() {
  153.         return windowLengthInMs;
  154.     }
  155. }

统计类com.dubbo.limit.slidingWindow.WindowWrap

  1. public class WindowWrap {
  3.     /**
  4.      * 样本窗口长度
  5.      */
  6.     private final long windowLengthInMs;
  8.     /**
  9.      * 样本窗口的起始时间戳
  10.      */
  11.     private long windowStart;
  13.     /**
  14.      * 当前样本窗口中的统计数据
  15.      */
  16.     private LongAdder value;
  19.     public WindowWrap(long windowLengthInMs, long windowStart,LongAdder value) {
  20.         this.windowLengthInMs = windowLengthInMs;
  21.         this.windowStart = windowStart;
  22.         this.value = value;
  23.     }
  25.     /**
  26.      *
  27.      * 功能描述:重置设置此窗口的开始时间戳
  28.      *
  29.      * @param:
  30.      * @return:
  31.      * @auther: Zywoo Lee
  32.      * @date: 2022/4/5 14:11
  33.      */
  34.     public WindowWrap resetTo(long startTime){
  35.         this.windowStart = startTime;
  36.         return this;
  37.     }
  39.     /**
  40.      *
  41.      * 功能描述:判断时间戳是否在此时间窗口内
  42.      *
  43.      * @param:
  44.      * @return:
  45.      * @auther: Zywoo Lee
  46.      * @date: 2022/4/5 14:11
  47.      */
  48.     public boolean isTimeInWindow(long timeMillis) {
  49.         return windowStart <= timeMillis && timeMillis < windowStart + windowLengthInMs;
  50.     }
  53.     public long windowStart() {
  54.         return windowStart;
  55.     }
  58.     public LongAdder value() {
  59.         return value;
  60.     }
  63. }

测试用例

  1.     @org.junit.Test
  2.     public void SlidingWindowTest() throws InterruptedException {
  3.         //时间窗长度为1s,分割为10个滑动窗口,时间窗内只能有150个放行
  4.         LimitItem slidingWindowItem = new SlidingWindowItem(150,10,1000);
  5.         AtomicInteger allowed = new AtomicInteger(0);
  6.         AtomicInteger limited = new AtomicInteger(0);
  7.         long beginTime = System.currentTimeMillis();
  8.         for (int i = 0; i < 2000; i++) {
  9.             new Thread(() -> {
  10.                 //System.out.println(Thread.currentThread().getName()+":"+leakyBucketItem.isAllowed());
  11.                 if (slidingWindowItem.isAllowed()) {
  12.                     allowed.addAndGet(1);
  13.                 } else {
  14.                     limited.addAndGet(1);
  15.                 }
  16.             }, i + "").start();
  17.             //每20次休眠100毫秒
  18.             if (i % 20 == 0) {
  19.                 Thread.sleep(100);
  20.             }
  21.         }
  22.         long endTime = System.currentTimeMillis();
  23.         //等待1s让所有线程都请求完再展示结果
  24.         Thread.sleep(1000);
  25.         System.out.println("执行时间:" + (endTime - beginTime) + "毫秒");
  26.         System.out.println("限制次数:" + limited.get() + "通过次数:" + allowed.get());
  27.     }

首次测试结果

执行时间:10570毫秒
限制次数:414通过次数:1586

1586/10.570=150.047304，约等于每秒150，符合预期

第二次测试结果

执行时间:10593毫秒
限制次数:415通过次数:1585

1585/10.593=149.627112，约等于每秒150，符合预期

第三次测试结果

执行时间:10640毫秒
限制次数:409通过次数:1591

1591/10.64=149.530075，约等于每秒150，符合预期

令牌桶限流算法（Guava）

令牌桶限流算法我引入了Guava的RateLimiter

版本

  1. <dependency>
  2.     <groupId>com.google.guava</groupId>
  3.     <artifactId>guava</artifactId>
  4.     <version>31.0-jre</version>
  5.     <type>pom</type>
  6. </dependency>
  1. /**
  2.  * @Auther: Zywoo Lee
  3.  * @Date: 2022/4/10 15:00
  4.  * @Description: Guava令牌桶限流算法
  5.  */
  6. public class GuavaRateLimit implements LimitItem{
  8.     /**
  9.      * 服务名称
  10.      */
  11.     String name;
  13.     /**
  14.      * Guava限流实现
  15.      */
  16.     RateLimiter limit;
  18.     /**
  19.      *
  20.      * @param name 服务名称
  21.      * @param permitsPerSecond 每秒限制通过数
  22.      */
  23.     public GuavaRateLimit(String name,double permitsPerSecond) {
  24.         this.name = name;
  25.         limit = RateLimiter.create(permitsPerSecond);
  26.     }
  28.     @Override
  29.     public boolean isAllowed() {
  30.         return limit.tryAcquire();
  31.     }
  32. }

调用核心链路：

  1. com.dubbo.limit.GuavaRateLimit#isAllowed
  2. com.google.common.util.concurrent.RateLimiter#tryAcquire()
  3. com.google.common.util.concurrent.RateLimiter#tryAcquire(int, long, java.util.concurrent.TimeUnit)
  4. com.google.common.util.concurrent.RateLimiter#reserveAndGetWaitLength
  5. com.google.common.util.concurrent.SmoothRateLimiter#reserveEarliestAvailable

com.google.common.util.concurrent.RateLimiter#tryAcquire(int, long, java.util.concurrent.TimeUnit)

  1. public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
  2.   long timeoutMicros = max(unit.toMicros(timeout), 0);
  3.   checkPermits(permits);
  4.   long microsToWait;
  5.   synchronized (mutex()) {
  6.     long nowMicros = stopwatch.readMicros();
  7.     if (!canAcquire(nowMicros, timeoutMicros)) {
  8.       return false;
  9.     } else {
  10.       //获取令牌所需要的时长（单位：微秒）
  11.       microsToWait = reserveAndGetWaitLength(permits, nowMicros);
  12.     }
  13.   }
  14.   //休眠
  15.   stopwatch.sleepMicrosUninterruptibly(microsToWait);
  16.   return true;
  17. }

com.google.common.util.concurrent.SmoothRateLimiter#reserveEarliestAvailable

  1. @Override
  2. final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
  3.   // 同步时间和桶里的令牌数量
  4.   resync(nowMicros);
  5.   // 当前请求获取成功的时间为 nextFreeTicketMicros
  6.   // 与请求数量 requiredPermits 无关
  7.   long returnValue = nextFreeTicketMicros;
  8.   // 计算当前立即可得的许可数量
  9.   double storedPermitsToSpend = min(requiredPermits, this.storedPermits);
  10.   // 当前请求需要新生成的许可数量
  11.   double freshPermits = requiredPermits - storedPermitsToSpend;
  12.   // 生成上述数量新许可需要的时间
  13.   long waitMicros =
  14.       storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
  15.           + (long) (freshPermits * stableIntervalMicros);
  16.   // 新许可需要的时间由下一个请求承担
  17.   this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
  18.   this.storedPermits -= storedPermitsToSpend;
  19.   return returnValue;
  20. }

测试用例

  1. /**
  2.  *
  3.  * 功能描述:Guava令牌桶算法测试 
  4.  *
  5.  * @param: 
  6.  * @return: 
  7.  * @auther: Zywoo Lee
  8.  * @date: 2022/4/10 17:54
  9.  */
  10. @org.junit.Test
  11. public void GuavaRateLimitTest() throws InterruptedException {
  12.     //每秒发放两个令牌
  13.     LimitItem leakyBucketItem = new GuavaRateLimit("ZywooTest", 2);
  14.     AtomicInteger allowed = new AtomicInteger(0);
  15.     AtomicInteger limited = new AtomicInteger(0);
  16.     long beginTime = System.currentTimeMillis();
  17.     for (int i = 0; i < 1000; i++) {
  18.         new Thread(() -> {
  19.             //System.out.println(Thread.currentThread().getName()+":"+leakyBucketItem.isAllowed());
  20.             if (leakyBucketItem.isAllowed()) {
  21.                 allowed.addAndGet(1);
  22.             } else {
  23.                 limited.addAndGet(1);
  24.             }
  25.         }, i + "").start();
  26.         //每50次休眠500毫秒
  27.         if (i % 50 == 0) {
  28.             Thread.sleep(500);
  29.         }
  30.     }
  31.     long endTime = System.currentTimeMillis();
  32.     //等待1s让所有线程都请求完再展示结果
  33.     Thread.sleep(1000);
  34.     System.out.println("执行时间:" + (endTime - beginTime) + "毫秒");
  35.     System.out.println("限制次数:" + limited.get() + "通过次数:" + allowed.get());
  37. }

第一次执行结果

执行时间:10219毫秒
限制次数:979通过次数:21，符合预期

第二次执行结果

执行时间:10224毫秒
限制次数:979通过次数:21，符合预期

第三次执行结果

执行时间:10231毫秒
限制次数:979通过次数:21，符合预期

漏桶限流算法（自定义，待完善）

  1. /**
  2.  * @Auther: Zywoo Lee
  3.  * @Date: 2022/4/1 20:56
  4.  * @Description: 漏桶限流实现类
  5.  */
  6. public class LeakyBucketItem implements LimitItem {
  8.     /**
  9.      * 服务名称
  10.      */
  11.     private String name;
  13.     /**
  14.      * 每秒处理数(出水率)
  15.      */
  16.     private long rate;
  18.     /**
  19.      * 当前剩余水量
  20.      */
  21.     private LongAdder currentWater;
  23.     /**
  24.      * 最后刷新时间
  25.      */
  26.     private long refreshTime;
  28.     /**
  29.      * 桶容量
  30.      */
  31.     private long capacity;
  33.     public LeakyBucketItem(String name, long rate, long refreshTime, long capacity) {
  34.         this.name = name;
  35.         this.rate = rate;
  36.         this.currentWater =  new LongAdder();
  37.         this.refreshTime = refreshTime;
  38.         this.capacity = capacity;
  39.     }
  41.     @Override
  42.     public boolean isAllowed() {
  43.         //当前时间
  44.         long currentTime = System.currentTimeMillis();
  45.         //流出的水量 = (当前时间-上次刷新时间)*出水率
  46.         long outWater = (currentTime - refreshTime) / 1000 * rate;
  47.         if (currentWater.longValue()<outWater){
  48.             currentWater.reset();
  49.         }else{
  50.             //当前水量=原先剩余水量-流出的水量
  51.             currentWater.add(-outWater);
  52.         }
  55.         //如果当前剩余水量小于桶的容量,则请求放行
  56.         if (currentWater.longValue() < capacity) {
  57.             currentWater.add(1);
  58.             refreshTime = currentTime;
  59.             System.out.println("漏桶限流算法:<接受请求>当前容量:" + capacity + "剩余容量:" + currentWater.longValue());
  60.             return true;
  61.         }
  62.         System.out.println("漏桶限流算法:<拒绝请求>当前容量:" + capacity + "剩余容量:" + currentWater.longValue());
  63.         return false;
  66.     }
  68. }

测试用例

  1.     @org.junit.Test
  2.     public void LeakyBucketTest() throws InterruptedException {
  3.         // 桶容量10,每秒流出速率5
  4.         LimitItem leakyBucketItem = new LeakyBucketItem("ZywooTest", 5, 0L, System.currentTimeMillis(), 10);
  5.         AtomicInteger allowed = new AtomicInteger(0);
  6.         AtomicInteger limited = new AtomicInteger(0);
  7.         long beginTime = System.currentTimeMillis();
  8.         for (int i = 0; i < 1000; i++) {
  9.             new Thread(() -> {
  10.                 //System.out.println(Thread.currentThread().getName()+":"+leakyBucketItem.isAllowed());
  11.                 if (leakyBucketItem.isAllowed()) {
  12.                     allowed.addAndGet(1);
  13.                 } else {
  14.                     limited.addAndGet(1);
  15.                 }
  16.             }, i + "").start();
  17.             //每50次休眠500毫秒
  18.             if (i % 50 == 0) {
  19.                 Thread.sleep(500);
  20.             }
  21.         }
  22.         long endTime = System.currentTimeMillis();
  23.         //等待1s让所有线程都请求完再展示结果
  24.         Thread.sleep(1000);
  25.         System.out.println("执行时间:" + (endTime - beginTime) + "毫秒");
  26.         System.out.println("限制次数:" + limited.get() + "通过次数:" + allowed.get());
  28.     }

设置桶容量10，每秒流出速率5，设置1000个线程，每请求50次休眠500毫秒。

第一次执行结果

执行时间:10214毫秒
限制次数:945通过次数:55

也就是说执行时间为50.829秒，在首个执行的1s内，前100个请求有10个请求是能放行的，在之后的9s~10s（10.2s减去1s在这个范围内），每秒只有5个请求能放行的（每秒流出速率5）。预期：10+9 5=55或10+10 5=60。通过次数55在范围内，符合预期。

第二次执行结果

执行时间:10224毫秒
限制次数:945通过次数:55

符合预期

第三次执行结果

执行时间:10241毫秒
限制次数:940通过次数:60

符合预期

Github地址

https://github.com/leeyiyu/dubbo-limit