Guava RateLimiter源码解析以及分布式限流总结

我就是我 2021-11-24 00:18 426阅读 0赞

在一个抽奖项目中为了应对流量洪峰使用了这个RateLimiter组件，在最近一项目中为了控制对下游服务的流量使用了分布式限流组件，这篇文章就是总结这两种，首先看一些基础的东西，以下内容来源于：[这里][Link 1]

在开发高并发系统时有三把利器用来保护系统：缓存、降级和限流

*  `缓存` 缓存的目的是提升系统访问速度和增大系统处理容量
 *  `降级` 降级是当服务出现问题或者影响到核心流程时，需要暂时屏蔽掉，待高峰或者问题解决后再打开
 *  `限流` 限流的目的是通过对并发访问/请求进行限速，或者对一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务、排队或等待、降级等处理

--------------------

# 常用限流算法 #

常用的限流算法有两种：漏桶算法和令牌桶算法

> 漏桶算法思路很简单，水（请求）先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出，可以看出漏桶算法能强行限制数据的传输速率。

对于很多应用场景来说，除了要求能够限制数据的平均传输速率外，还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了，令牌桶算法更为适合。

> 令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。

更多关于漏桶算法和令牌桶算法的介绍可以参考 [http://blog.csdn.net/charlesl...][http_blog.csdn.net_charlesl...]

# 漏桶算法VS令牌桶算法 #

1、令牌桶是按照固定速率往桶中添加令牌，请求是否被处理需要看桶中令牌是否足够，当令牌数减为零时则拒绝新的请求；

2、漏桶则是按照常量固定速率流出请求，流入请求速率任意，当流入的请求数累积到漏桶容量时，则新流入的请求被拒绝；

3、令牌桶限制的是平均流入速率（允许突发请求，只要有令牌就可以处理，支持一次拿3个令牌，4个令牌），并允许一定程度突发流量；

4、漏桶限制的是常量流出速率（即流出速率是一个固定常量值，比如都是1的速率流出，而不能一次是1，下次又是2），从而平滑突发流入速率；

5、令牌桶允许一定程度的突发，而漏桶主要目的是平滑流入速率；

6、两个算法实现可以一样，但是方向是相反的，对于相同的参数得到的限流效果是一样的。

# 令牌桶算法原理图 #

[图来自这里][Link 2]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70][]

# ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70 1][] #

# 源码解析 #

几个关键参数：

/**
       *当前存储的令牌数 
       */
      double storedPermits;
    
      /** 
       * 最大可存储令牌数 
       */
      double maxPermits;
    
      /**
       * 产生一个令牌的时间间隔
       */
      double stableIntervalMicros;
    
      /**
       * 下次可以拿到令牌的时间
       */
      private long nextFreeTicketMicros = 0L; // could be either in the past or future
    
      /** 
       * 限流器空闲时，保存多少秒产生的令牌器
       */
      final double maxBurstSeconds;

Guava 的 RateLimiter 有两种模式：

1、稳定模式，SmoothBursty，令牌生成的速度恒定。  
2、预热模式，SmoothWarmingUp，令牌生成速度缓慢提升直到维持在一个稳定值。比如在系统刚启动的时候，第一次访问的接口需要加载缓存等等，这时系统的处理速度较慢，就需要这种模式

默认是

public static RateLimiter create(double permitsPerSecond) {
        return create(SleepingTicker.SYSTEM_TICKER, permitsPerSecond);
      }
    
      @VisibleForTesting
      static RateLimiter create(SleepingTicker ticker, double permitsPerSecond) {
        RateLimiter rateLimiter = new Bursty(ticker, 1.0 /* maxBurstSeconds */);
        rateLimiter.setRate(permitsPerSecond);
        return rateLimiter;
      }
    
      /**
       * 生成一个bursty对象，同时设置空闲时保存多少秒产生的令牌器
       * 这里很明显默认是1
       * 当速率限制器闲置时，允许许可数暴增到permitsPerSecond，随后的请求会被平滑地限制在稳定速率permitsPerSecond中
       **/
      private static class Bursty extends RateLimiter {
        /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */
        final double maxBurstSeconds;
    
        Bursty(SleepingTicker ticker, double maxBurstSeconds) {
          super(ticker);
          this.maxBurstSeconds = maxBurstSeconds;
        }
    
        @Override
        void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
          /**
           * 初始值是0
           **/
          double oldMaxPermits = this.maxPermits;
    
          /**
           * 此值是一个固定值，最大的令牌数
           **/
          maxPermits = maxBurstSeconds * permitsPerSecond;
    
          /**
           * 刚开始设置的时候是0
           * 下次执行的时候更新为最新的存储值
           **/
          storedPermits = (oldMaxPermits == 0.0)
              ? 0.0 // initial state
              : storedPermits * maxPermits / oldMaxPermits;
        }
    
        @Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          return 0L;
        }
      }
    
      /**
       * 更新RateLimite的稳定速率，参数permitsPerSecond 由构造RateLimiter的工厂方法提供。调用该方法后，当前限制线程不会被唤醒，因此他们不会注意到最新的速率；
       * 只有接下来的请求才会。需要注意的是，由于每次请求偿还了（通过等待，如果需要的话）上一次请求的开销，这意味着紧紧跟着的下一个请求不会被最新的速率影响到，
       * 在调用了setRate 之后；它会偿还上一次请求的开销，这个开销依赖于之前的速率。RateLimiter的行为在任何方式下都不会被改变，
       * 比如如果 RateLimiter 有20秒的预热期配置，在此方法被调用后它还是会进行20秒的预热。
      public final void setRate(double permitsPerSecond) {
        Preconditions.checkArgument(permitsPerSecond > 0.0
            && !Double.isNaN(permitsPerSecond), "rate must be positive");
        synchronized (mutex) {
          resync(readSafeMicros());
          double stableIntervalMicros = TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond;
          this.stableIntervalMicros = stableIntervalMicros;
          doSetRate(permitsPerSecond, stableIntervalMicros);
        }
      }
    
      /**
       * 根据令牌桶算法，桶中的令牌是持续生成存放的，有请求时需要先从桶中拿到令牌才能开始执行，谁来持续生成令牌存放呢？
       *
       * 一种解法是，开启一个定时任务，由定时任务持续生成令牌。这样的问题在于会极大的消耗系统资源，如，某接口需要分别对每个用户做访问频率限制，假设系统中存在6W用户，则至多需要开启6W个定时任务来维持每个桶中的令牌数，这样的开销是巨大的。
       * 
       * 另一种解法则是延迟计算，如上resync函数。该函数会在每次获取令牌之前调用，其实现思路为，若当前时间晚于nextFreeTicketMicros，则计算该段时间内可以生成多少令牌，将生成的令牌加入令牌桶中并更新数据。这样一来，只需要在获取令牌时计算一次即可。
      private void resync(long nowMicros) {
        // if nextFreeTicket is in the past, resync to now
        if (nowMicros > nextFreeTicketMicros) {
          storedPermits = Math.min(maxPermits,storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
          nextFreeTicketMicros = nowMicros;
        }
      }

以上的步骤就create了一个RateLimiter了，下边是使用时候的代码：

/**
       * 默认是请求一个令牌，单位是毫秒
       **/
      public boolean tryAcquire() {
        return tryAcquire(1, 0, TimeUnit.MICROSECONDS);
      }
    
      public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
        long timeoutMicros = unit.toMicros(timeout);
        /**
         * 校验大于0
         **/
        checkPermits(permits);
        long microsToWait;
        /**
         * 同步当前对象，相当于加了一个锁
        synchronized (mutex) {
          /**
           * 读取当前的毫秒
           **/
          long nowMicros = readSafeMicros();
          /**
           * 还没有到下次获取令牌的时间直接返回false
           **/
          if (nextFreeTicketMicros > nowMicros + timeoutMicros) {
            return false;
          } else {
            /**
             * 获得需要等待的时间
             * 主要计算下次获取的时间
             **/
            microsToWait = reserveNextTicket(permits, nowMicros);
          }
        }
        /**
         * 肯定不会阻塞，直接返回true
         **/
        ticker.sleepMicrosUninterruptibly(microsToWait);
        return true;
      }
    
     /**
      * 
      **/
     private long reserveNextTicket(double requiredPermits, long nowMicros) {
        resync(nowMicros);
        // 如果是过去时间，因为上面刚同步过，肯定为0，不需要等待；主要针对下一次是未来时间
        long microsToNextFreeTicket = nextFreeTicketMicros - nowMicros;
        // 存储的令牌有多少被使用
        double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits);
        // 需要等待新生成的令牌数（这里的等待其实是再还上一次预支的令牌，本次的预支不需要等待，留给一次再还）
        double freshPermits = requiredPermits - storedPermitsToSpend;
    
        // 以下函数原guava的实现里计算等待会加上，但只针对WarmingUp使用
        // storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
        long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) + (long) (freshPermits * stableIntervalMicros);
        /**
         * 更新下一次不需要等待时间
         * 如果参数是0.1，那么就意味着在1s内会有10个通过，这个值是一点点增加到10，然后维持在10不变，由于我们设置的是busry所以这里是流量徒增类型的
         **/
        this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;
        // 减扣消费的令牌数
        this.storedPermits -= storedPermitsToSpend;
        return microsToNextFreeTicket;
      }
    
      static abstract class SleepingTicker extends Ticker {
        abstract void sleepMicrosUninterruptibly(long micros);
    
        static final SleepingTicker SYSTEM_TICKER = new SleepingTicker() {
          @Override
          public long read() {
            return systemTicker().read();
          }
    
          /** 
           * 阻塞等待
           **/
          @Override
          public void sleepMicrosUninterruptibly(long micros) {
            if (micros > 0) {
              Uninterruptibles.sleepUninterruptibly(micros, TimeUnit.MICROSECONDS);
            }
          }
        };
      }

流量warmUp和徒增又是在哪里呢？

private static class WarmingUp extends RateLimiter {
    
        final long warmupPeriodMicros;
        /**
         * The slope of the line from the stable interval (when permits == 0), to the cold interval
         * (when permits == maxPermits)
         */
        private double slope;
        private double halfPermits;
    
        WarmingUp(SleepingTicker ticker, long warmupPeriod, TimeUnit timeUnit) {
          super(ticker);
          this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod);
        }
    
        @Override
        void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
          double oldMaxPermits = maxPermits;
          maxPermits = warmupPeriodMicros / stableIntervalMicros;
          halfPermits = maxPermits / 2.0;
          // Stable interval is x, cold is 3x, so on average it's 2x. Double the time -> halve the rate
          double coldIntervalMicros = stableIntervalMicros * 3.0;
          slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits;
          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;
          }
        }
    
        @Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          double availablePermitsAboveHalf = storedPermits - halfPermits;
          long micros = 0;
          // measuring the integral on the right part of the function (the climbing line)
          if (availablePermitsAboveHalf > 0.0) {
            double permitsAboveHalfToTake = Math.min(availablePermitsAboveHalf, permitsToTake);
            micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf)
                + permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);
            permitsToTake -= permitsAboveHalfToTake;
          }
          // measuring the integral on the left part of the function (the horizontal line)
          micros += (stableIntervalMicros * permitsToTake);
          return micros;
        }
    
        private double permitsToTime(double permits) {
          return stableIntervalMicros + permits * slope;
        }
      }
    
    
      private static class Bursty extends RateLimiter {
        /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */
        final double maxBurstSeconds;
    
        Bursty(SleepingTicker ticker, double maxBurstSeconds) {
          super(ticker);
          this.maxBurstSeconds = maxBurstSeconds;
        }
    
        @Override
        void doSetRate(double permitsPerSecond, double stableIntervalMicros) {
          double oldMaxPermits = this.maxPermits;
          maxPermits = maxBurstSeconds * permitsPerSecond;
          storedPermits = (oldMaxPermits == 0.0)
              ? 0.0 // initial state
              : storedPermits * maxPermits / oldMaxPermits;
        }
    
        @Override
        long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {
          return 0L;
        }
      }
    
      @VisibleForTesting
      static abstract class SleepingTicker extends Ticker {
        abstract void sleepMicrosUninterruptibly(long micros);
    
        static final SleepingTicker SYSTEM_TICKER = new SleepingTicker() {
          @Override
          public long read() {
            return systemTicker().read();
          }
    
          @Override
          public void sleepMicrosUninterruptibly(long micros) {
            if (micros > 0) {
              Uninterruptibles.sleepUninterruptibly(micros, TimeUnit.MICROSECONDS);
            }
          }
        };
      }
    }

看上边setRate方法，一个是有梯度的，一个直接到了最大值，然后在设置下一个获取时间的时候做限制。

# 分布式限流 #

总体思路就是利用redis实现，key以秒为单位，如果当前秒数内还额度的话，继续增加，没有的话设置当前秒数据，再有请求进入的话直接拒绝，在第一次的时候设置过期的时间：

//缓存，记录最近一次用光配额的秒数
     private final AtomicReference<String> exhausted = new AtomicReference<String>();
    
     public boolean tryAcquire() {
    
    
            long startNs = System.nanoTime();
            boolean acquired = true;
            boolean redisSuccess = true;
    
            try {
                long epochSecond = System.currentTimeMillis()/1000;
                String second = String.valueOf(epochSecond);
                /**
                 * 当前的秒数内的数据如果用完了，直接返回
                 **/
                if (second.equals(this.exhausted.get())) {
                    acquired = false;
                    return acquired;
                }
    
                String timeBucketKey = uniqueKey + ":" + epochSecond;
                String hashKey = String.valueOf(epochSecond + this.hashKeyFactor);
    
                Long inc = null;
                try {
                    inc = this.redisInvoker.inc(hashKey, timeBucketKey, 1);
                } catch (Exception e) {
                    logger.error("error inc, hashKey:" + hashKey + ", redisKey:" + timeBucketKey
                            + ", msg:" + e.getMessage(), e);
                    //异常认为acquire成功(不会降级)，避免redis挂掉导致消息全部被降级
                    redisSuccess = false;
                    return true;
                }
    
                /**
                 * 第一次的时候设置过期时间5s
                if (Long.valueOf(1).equals(inc)) {
                    redisKeyCleaner.execute(new RedisKeyCleaner.CleanJob(hashKey, timeBucketKey, 5, this.redisInvoker));
                }
    
                /**
                 * 增加的值小于QPS的时候放过
                 **/
                acquired = inc <= this.qps;
                /**
                 * 用完设置当前的秒数
                 **/
                if (!acquired) {
                    this.exhausted.set(second);
                }
               
                return acquired;
            } finally {
                long timeCost = System.nanoTime() - startNs;
                RateLimitMonitor monitor = monitorRef.get();
    
                if (acquired) {
                    monitor.recordAcquiredSuccessResult(timeCost);
                } else {
                    monitor.recordAcquiredFailedResult(timeCost);
                }
    
                if (!redisSuccess) {
                    monitor.recordRedisErrorResult();
                }
            }
        }

参考：

[https://blog.csdn.net/a295567172/article/details/83345881][https_blog.csdn.net_a295567172_article_details_83345881]

[https://massivetechinterview.blogspot.com/2015/10/develop-api-rate-limit-throttling-client.html][https_massivetechinterview.blogspot.com_2015_10_develop-api-rate-limit-throttling-client.html]

[http://ifeve.com/guava-ratelimiter/][http_ifeve.com_guava-ratelimiter]

[https://segmentfault.com/a/1190000012875897][Link 1]

[http://xiaobaoqiu.github.io/blog/2015/07/02/ratelimiter/][http_xiaobaoqiu.github.io_blog_2015_07_02_ratelimiter]

[http://blog.eveow.com/post/262][http_blog.eveow.com_post_262]

不经意间发现了这个：

### [http://jm.taobao.org/hire/][http_jm.taobao.org_hire] 牛逼牛逼 ###

[Link 1]: https://segmentfault.com/a/1190000012875897
[http_blog.csdn.net_charlesl...]: http://blog.csdn.net/charleslei/article/details/53152883
[Link 2]: https://blog.csdn.net/ShiXueTanLang/article/details/80960208
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70]: /images/20211122/33082689875d44a584e4e8778b8b6e15.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hb3llcWl1_size_16_color_FFFFFF_t_70 1]: /images/20211122/144a797e3c884951be61aae74619c5a5.png
[https_blog.csdn.net_a295567172_article_details_83345881]: https://blog.csdn.net/a295567172/article/details/83345881
[https_massivetechinterview.blogspot.com_2015_10_develop-api-rate-limit-throttling-client.html]: https://massivetechinterview.blogspot.com/2015/10/develop-api-rate-limit-throttling-client.html
[http_ifeve.com_guava-ratelimiter]: http://ifeve.com/guava-ratelimiter/
[http_xiaobaoqiu.github.io_blog_2015_07_02_ratelimiter]: http://xiaobaoqiu.github.io/blog/2015/07/02/ratelimiter/
[http_blog.eveow.com_post_262]: http://blog.eveow.com/post/262
[http_jm.taobao.org_hire]: http://jm.taobao.org/hire/