令牌桶算法限流

我就是我 2022-04-23 06:06 256阅读 0赞

## 限流 ##

限流是对某一时间窗口内的请求数进行限制，保持系统的可用性和稳定性，防止因流量暴增而导致的系统运行缓慢或宕机。常用的限流算法有令牌桶和和漏桶，而Google开源项目Guava中的RateLimiter使用的就是令牌桶控制算法。

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

*  缓存：缓存的目的是提升系统访问速度和增大系统处理容量
 *  降级：降级是当服务器压力剧增的情况下，根据当前业务情况及流量对一些服务和页面有策略的降级，以此释放服务器资源以保证核心任务的正常运行
 *  限流：限流的目的是通过对并发访问/请求进行限速，或者对一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务、排队或等待、降级等处理

我们经常在调别人的接口的时候会发现有限制，比如微信公众平台接口、百度API Store、聚合API等等这样的，对方会限制每天最多调多少次或者每分钟最多调多少次

我们自己在开发系统的时候也需要考虑到这些，比如我们公司在上传商品的时候就做了限流，因为用户每一次上传商品，我们需要将商品数据同到到美团、饿了么、京东、百度、自营等第三方平台，这个工作量是巨大，频繁操作会拖慢系统，故做限流。

以上都是题外话，接下来我们重点看一下令牌桶算法

## 令牌桶算法 ##

下面是从网上找的两张图来描述令牌桶算法：

> ![874963-20180727184550598-845634570.png][]   ![874963-20180727184559491-371099708.png][]

## RateLimiter ##

[https://github.com/google/guava][https_github.com_google_guava]

![874963-20180727185243164-1080939791.png][]

RateLimiter的代码不长，注释加代码432行，看一下RateLimiter怎么用

![复制代码][copycode.gif]

1 package com.cjs.example;
     2 
     3 import com.google.common.util.concurrent.RateLimiter;
     4 import org.springframework.web.bind.annotation.RequestMapping;
     5 import org.springframework.web.bind.annotation.RestController;
     6 
     7 import java.text.SimpleDateFormat;
     8 import java.util.Date;
     9 
    10 @RestController
    11 public class HelloController {
    12 
    13     private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    14 
    15     private static final RateLimiter rateLimiter = RateLimiter.create(2);
    16 
    17     /**
    18      * tryAcquire尝试获取permit，默认超时时间是0，意思是拿不到就立即返回false
    19      */
    20     @RequestMapping("/sayHello")
    21     public String sayHello() {
    22         if (rateLimiter.tryAcquire()) { //  一次拿1个
    23             System.out.println(sdf.format(new Date()));
    24             try {
    25                 Thread.sleep(500);
    26             } catch (InterruptedException e) {
    27                 e.printStackTrace();
    28             }
    29         }else {
    30             System.out.println("limit");
    31         }
    32         return "hello";
    33     }
    34 
    35     /**
    36      * acquire拿不到就等待，拿到为止
    37      */
    38     @RequestMapping("/sayHi")
    39     public String sayHi() {
    40         rateLimiter.acquire(5); //  一次拿5个
    41         System.out.println(sdf.format(new Date()));
    42         return "hi";
    43     }
    44 
    45 }

![复制代码][copycode.gif]

![874963-20180727185756001-2126438531.png][]

![874963-20180727192937920-1827417915.png][]

![874963-20180727193010457-248832890.png][]

![874963-20180727193056009-1747942169.png][]

### 关于RateLimiter： ###

*  A rate limiter。每个acquire()方法如果必要的话会阻塞直到一个permit可用，然后消费它。获得permit以后不需要释放。
 *  RateLimiter在并发环境下使用是安全的：它将限制所有线程调用的总速率。注意，它不保证公平调用。
 *  RateLimiter在并发环境下使用是安全的：它将限制所有线程调用的总速率。注意，它不保证公平调用。Rate limiter（直译为：速度限制器）经常被用来限制一些物理或者逻辑资源的访问速率。这和java.util.concurrent.Semaphore正好形成对照。
 *  一个RateLimiter主要定义了发放permits的速率。如果没有额外的配置，permits将以固定的速度分配，单位是每秒多少permits。默认情况下，Permits将会被稳定的平缓的发放。
 *  可以配置一个RateLimiter有一个预热期，在此期间permits的发放速度每秒稳步增长直到到达稳定的速率

### 基本用法： ###

![复制代码][copycode.gif]

final RateLimiter rateLimiter = RateLimiter.create(2.0); // rate is "2 permits per second"
    void submitTasks(List<Runnable> tasks, Executor executor) {
        for (Runnable task : tasks) {
            rateLimiter.acquire(); // may wait
            executor.execute(task);
        }
    }

![复制代码][copycode.gif]

### 实现 ###

![874963-20180727190953899-767102701.png][]

SmoothBursty以稳定的速度生成permit

SmoothWarmingUp是渐进式的生成，最终达到最大值趋于稳定

> ![874963-20180727191654729-920034681.png][]

### 源码片段解读： ###

![复制代码][copycode.gif]

public abstract class RateLimiter {
    
        /**
         * 用给定的吞吐量（“permits per second”）创建一个RateLimiter。
         * 通常是QPS
         */
        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;
        }
        
        /**
         * 用给定的吞吐量(QPS)和一个预热期创建一个RateLimiter
         */
        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());
        }
    
        static RateLimiter create(
                double permitsPerSecond,
                long warmupPeriod,
                TimeUnit unit,
                double coldFactor,
                SleepingStopwatch stopwatch) {
            RateLimiter rateLimiter = new SmoothWarmingUp(stopwatch, warmupPeriod, unit, coldFactor);
            rateLimiter.setRate(permitsPerSecond);
            return rateLimiter;
        }
    
        private final SleepingStopwatch stopwatch;
    
        //    锁
        private volatile Object mutexDoNotUseDirectly;
    
        private Object mutex() {
            Object mutex = mutexDoNotUseDirectly;
            if (mutex == null) {
                synchronized (this) {
                    mutex = mutexDoNotUseDirectly;
                    if (mutex == null) {
                        mutexDoNotUseDirectly = mutex = new Object();
                    }
                }
            }
            return mutex;
        }
        
        /**
         * 从RateLimiter中获取一个permit，阻塞直到请求可以获得为止
         * @return 休眠的时间，单位是秒，如果没有被限制则是0.0
         */
        public double acquire() {
            return acquire(1);
        }
      
        /**
         * 从RateLimiter中获取指定数量的permits，阻塞直到请求可以获得为止
         */
        public double acquire(int permits) {
            long microsToWait = reserve(permits);
            stopwatch.sleepMicrosUninterruptibly(microsToWait);
            return 1.0 * microsToWait / SECONDS.toMicros(1L);
        }
        
        /**
         * 预定给定数量的permits以备将来使用
         * 直到这些预定数量的permits可以被消费则返回逝去的微秒数
         */
        final long reserve(int permits) {
            checkPermits(permits);
            synchronized (mutex()) {
                return reserveAndGetWaitLength(permits, stopwatch.readMicros());
            }
        }
        
        private static void checkPermits(int permits) {
            checkArgument(permits > 0, "Requested permits (%s) must be positive", permits);
        }
        
        final long reserveAndGetWaitLength(int permits, long nowMicros) {
            long momentAvailable = reserveEarliestAvailable(permits, nowMicros);
            return max(momentAvailable - nowMicros, 0);
        }
    }
    
    
    abstract class SmoothRateLimiter extends RateLimiter {
        
        /** The currently stored permits. */
        double storedPermits;
    
        /** The maximum number of stored permits. */
        double maxPermits;
    
        /**
         * The interval between two unit requests, at our stable rate. E.g., a stable rate of 5 permits
         * per second has a stable interval of 200ms.
         */
        double stableIntervalMicros;
        
        /**
         * The time when the next request (no matter its size) will be granted. After granting a request,
         * this is pushed further in the future. Large requests push this further than small requests.
         */
        private long nextFreeTicketMicros = 0L; // could be either in the past or future
        
        final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {
            resync(nowMicros);
            long returnValue = nextFreeTicketMicros;
            double storedPermitsToSpend = min(requiredPermits, this.storedPermits);    //    本次可以获取到的permit数量
            double freshPermits = requiredPermits - storedPermitsToSpend;    //    差值，如果存储的permit大于本次需要的permit数量则此处是0，否则是一个正数
            long waitMicros =
                storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)
                    + (long) (freshPermits * stableIntervalMicros);    //    计算需要等待的时间（微秒）
    
            this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros, waitMicros);
            this.storedPermits -= storedPermitsToSpend;    //    减去本次消费的permit数
            return returnValue;
        }
        
        void resync(long nowMicros) {
            // if nextFreeTicket is in the past, resync to now
            if (nowMicros > nextFreeTicketMicros) {    //    表示当前可以获得permit
                double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();    //    计算这段时间可以生成多少个permit
                storedPermits = min(maxPermits, storedPermits + newPermits);    //    如果超过maxPermit，则取maxPermit，否则取存储的permit+新生成的permit
                nextFreeTicketMicros = nowMicros;    //    设置下一次可以获得permit的时间点为当前时间
            }
        }
    }

![复制代码][copycode.gif]

RateLimiter实现的令牌桶算法，不仅可以应对正常流量的限速，而且可以处理突发暴增的请求，实现平滑限流。

通过代码，我们可以看到它可以预消费，怎么讲呢

nextFreeTicketMicros表示下一次请求获得permits的最早时间。每次授权一个请求以后，这个值会向后推移（PS：想象一下时间轴）即向未来推移。因此，大的请求会比小的请求推得更。这里的大小指的是获取permit的数量。这个应该很好理解，因为上一次请求获取的permit数越多，那么下一次再获取授权时更待的时候会更长，反之，如果上一次获取的少，那么时间向后推移的就少，下一次获得许可的时间更短。可见，都是有代价的。正所谓：要浪漫就要付出代价。

还要注意到一点，就是获取令牌和处理请求是两个动作，而且，并不是每一次都获取一个，也不要想当然的认为一个请求获取一个permit（或者叫令牌），可以再看看前面那幅图

### Stopwatch ###

一个以纳秒为单位度量流逝时间的对象。它是一个相对时间，而不是绝对时间。

Stopwatch stopwatch = Stopwatch.createStarted();
    System.out.println("hahah");
    stopwatch.stop();
    Duration duration = stopwatch.elapsed();
    System.out.println(stopwatch);

## Semaphore（信号量） ##

> A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each acquire() blocks if necessary until a permit is available, and then takes it. Each release() adds a permit, potentially releasing a blocking acquirer. However, no actual permit objects are used; the Semaphore just keeps a count of the number available and acts accordingly.

一个信号量维护了一系列permits。

每次调用acquire()方法获取permit，如果必要的话会阻塞直到有一个permit可用为止。

调用release()方法则会释放自己持有的permit，即用完了再还回去。

信号量限制的是并发访问临界资源的线程数。

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

### 漏桶 ###

漏桶的出水速度是恒定的，那么意味着如果瞬时大流量的话，将有大部分请求被丢弃掉（也就是所谓的溢出）。

### 令牌桶 ###

生成令牌的速度是恒定的，而请求去拿令牌是没有速度限制的。这意味，面对瞬时大流量，该算法可以在短时间内请求拿到大量令牌，而且拿令牌的过程并不是消耗很大的事情。

最后，不论是对于令牌桶拿不到令牌被拒绝，还是漏桶的水满了溢出，都是为了保证大部分流量的正常使用，而牺牲掉了少部分流量，这是合理的，如果因为极少部分流量需要保证的话，那么就可能导致系统达到极限而挂掉，得不偿失。

## 小定律：排队理论 ##

[https://en.wikipedia.org/wiki/Little%27s\_law][https_en.wikipedia.org_wiki_Little_27s_law]

> the long-term average number L of customers in a stationary system is equal to the long-term average effective arrival rate λ multiplied by the average time W that a customer spends in the system. Expressed algebraically the law is：
> 
> ![874963-20180727195445375-1406272747.png][]

在一个固定系统中，顾客的长期平均数量**L**等于顾客的长期平均到达速率**λ**乘以顾客在系统中平均花费的时间**W**。用公式表示为：

![874963-20180727195520820-1749595692.png][]

虽然这看起来很容易，但这是一个非常显著的举世瞩目的结果，因为这种关系“不受到达过程的分布，服务分布，服务顺序，或其他任何因素的影响”。这个结果适用于任何系统，特别是适用于系统内的系统。唯一的要求是系统必须是稳定的非抢占式的。

### 例子 ###

### 例1：找响应时间 ###

假设有一个应用程序没有简单的方法来度量响应时间。如果系统的平均数量和吞吐量是已知的，那么平均响应时间就是：

mean response time = mean number in system / mean throughput

平均响应时间 = 系统的平均数量 / 平均吞吐量.

### 例2：顾客在店里 ###

想象一下，一家小商店只有一个柜台和一个可供浏览的区域，每次只能有一个人在柜台，并且没有人不买东西就离开。

所以这个系统大致是：进入 --> 浏览 --> 柜台结账 --> 离开

在一个稳定的系统中，人们进入商店的速度就是他们到达商店的速度（我们叫做到达速度），它们离开的速度叫做离开速度。

相比之下，到达速度超过离开速度代表是一个不稳定的系统，这就会造成等待的顾客数量将逐渐增加到无穷大。

前面的小定律告诉我们，商店的平均顾客数量L等于有效的到达速度λ乘以顾客在商店的平均停留时间W。用公式表示为：

![874963-20180727195826804-163021892.png][]

假设，顾客以每小时10个的速度到达，并且平均停留时间是0.5小时。那么这就意味着，任意时间商店的平均顾客数量是5

![874963-20180727195921801-228132830.png][]

现在假设商店正在考虑做更多的广告，把到达率提高到每小时20。商店必须准备好容纳平均10人，或者必须将每个顾客在商店中的时间减少到0.25小时。商店可以通过更快地结帐或者增加更多的柜台来达到后者的目的。

我们可以把前面的小定律应用到商店系统中。例如，考虑柜台和在柜台前排的队。假设平均有2个人在柜台前排队，我们知道顾客到达速度是每小时10，所以顾客平均必须停留时间为0.2小时。

![874963-20180727200004348-640104162.png][]

## 最后 ##

这是单机（单进程）的限流，是JVM级别的的限流，所有的令牌生成都是在内存中，在分布式环境下不能直接这么用。

如果我们能把permit放到Redis中就可以在分布式环境中用了。

文章来源：[https://www.cnblogs.com/cjsblog/p/9379516.html][https_www.cnblogs.com_cjsblog_p_9379516.html]

[https://www.cnblogs.com/googlemeoften/p/6020718.html][https_www.cnblogs.com_googlemeoften_p_6020718.html]

[874963-20180727184550598-845634570.png]: https://images2018.cnblogs.com/blog/874963/201807/874963-20180727184550598-845634570.png
[874963-20180727184559491-371099708.png]: /images/20220323/0327dd4852524ebf860cdbae7a3c2413.png
[https_github.com_google_guava]: https://github.com/google/guava
[874963-20180727185243164-1080939791.png]: /images/20220323/9d85eab0b340460c9ab4b799780ab792.png
[copycode.gif]: /images/20220323/a7babe629c26466f8681e603b236480b.png
[874963-20180727185756001-2126438531.png]: /images/20220323/6770263252644fada103f5ffefc901dd.png
[874963-20180727192937920-1827417915.png]: /images/20220323/2728a8e02840471aa54bb89db4b3a936.png
[874963-20180727193010457-248832890.png]: /images/20220323/9f960495672e4b6396a9cc461f45b4b6.png
[874963-20180727193056009-1747942169.png]: /images/20220323/fe93cf41e1f7403883a1648b86adcad6.png
[874963-20180727190953899-767102701.png]: /images/20220323/200dd5cca41f4b178b40d48540b9ccc6.png
[874963-20180727191654729-920034681.png]: /images/20220323/73462ac6f4a245d4b0226f641fb202dd.png
[https_en.wikipedia.org_wiki_Little_27s_law]: https://en.wikipedia.org/wiki/Little%27s_law
[874963-20180727195445375-1406272747.png]: /images/20220323/871cc8c7a5dc4b62b4e6b8840a9b3312.png
[874963-20180727195520820-1749595692.png]: /images/20220323/792c5443092f4209924b259a74667bad.png
[874963-20180727195826804-163021892.png]: /images/20220323/cbd521d66a86482bbbf8812bffc2f8f9.png
[874963-20180727195921801-228132830.png]: /images/20220323/02f05082c0574d75a2b35adff2cb98ba.png
[874963-20180727200004348-640104162.png]: /images/20220323/bd34595155684a278e1655721dc21cc9.png
[https_www.cnblogs.com_cjsblog_p_9379516.html]: https://www.cnblogs.com/cjsblog/p/9379516.html
[https_www.cnblogs.com_googlemeoften_p_6020718.html]: https://www.cnblogs.com/googlemeoften/p/6020718.html