【限流】4 种常见的限流算法

我就是我 2023-10-13 20:06 86阅读 0赞

## 1. 背景 ##

服务限流，是指通过控制请求的速率或次数来达到保护服务的目的，在微服务中，我们通常会将它和熔断、降级搭配在一起使用，来避免瞬时的大量请求对系统造成负荷，来达到保护服务平稳运行的目的

> 限流和熔断有什么区别？

限流发生在流量进来之前，超过的流量进行限制。

熔断是一种应对故障的机制，发生在流量进来之后，如果系统发生故障或者异常，熔断会自动切断请求，防止故障进一步扩展，导致服务雪崩

> 限流和削峰有什么区别？

削峰是对流量的平滑处理，通过缓慢地增加请求的处理速率来避免系统瞬时过载。

削峰大概就是水库，把流量储存起来，慢慢流，限流大概就是闸口，拒绝超出的流量。

## 2. 限流概述 ##

在大多数的微服务架构在设计之初，比如在技术选型阶段，架构师会从一个全局的视角去规划技术栈的组合，比如结合当前产品的现状考虑是使用 dubbo？还是 springcloud？作为微服务治理的底层框架。甚至为了满足快速的上线、迭代和交付，直接以 springboot 为基座进行开发，后续再引入新的技术栈等…

所以在谈论某个业务场景具体的技术解决方案时不可一概而论，而是需要结合产品和业务的现状综合评估，以限流来说，在下面的不同的技术架构下具体在选择的时候可能也不一样

### 2.1 dubbo 服务治理模式 ###

选择 dubbo 框架作为基础服务治理对于那种偏向内部平台的应用还是不错的，dubbo 底层走 netty，这一点相比http协议来说，在一定场景下还是具有优势的，如果选择 dubbo，在选择限流方案上可以做如下的参考。

#### 2.1.1 dubbo 框架级限流 ####

dubbo 官方提供了完善的服务治理，能够满足大多数开发场景中的需求，针对限流这个场景，具体来说包括如下手段，具体的配置，可以参考官方手册；

**客户端限流**

*  信号量限流 （通过统计的方式）
 *  连接数限流 (socket->tcp)

**服务端限流**

*  线程池限流 (隔离手段)
 *  信号量限流 (非隔离手段)
 *  接收数限流 (socket->tcp)

#### 2.1.2 线程池设置 ####

多线程并发操作一定离不开线程池，Dubbo 自身提供了支持了四种线程池类型支持。生产者 `<dubbo:protocol>` 标签中可配置线程池关键参数，线程池类型、阻塞队列大小、核心线程数量等，通过配置生产端的线程池数量可以在一定程度上起到限流的效果。

#### 2.1.3 集成第三方组件 ####

如果是 springboot 框架的项目，可以考虑直接引入地方的组件或 SDK，比如 hystrix，guava，sentinel 原生 SDK 等，如果技术实力足够强甚至可以考虑自己造轮子。

### 2.2 springcloud 服务治理模式 ###

如果你的服务治理框架选用的是 springcloud 或 springcloud-alibaba，其框架自身的生态中已经包含了相应的限流组件，可以实现开箱即用，下面列举几种常用的基于 springcloud 框架的限流组件。

#### 2.2.1 hystrix ####

Hystrix 是 Netflix 开源的一款容错框架，在 springcloud 早期推出市场的时候，作为 springcloud 生态中用于限流、熔断、降级的一款组件。

Hystrix 提供了限流功能，在 springcloud 架构的系统中，可以在网关启用 Hystrix，进行限流处理，每个微服务也可以各自启用Hystrix 进行限流。

> Hystrix 默认使用线程隔离模式，可以通过线程数+队列大小进行限流，具体参数配置可以参考官网相关资料。

#### 2.2.2 sentinel ####

Sentinel 号称分布式系统的流量防卫兵，属于 springcloud-alibaba 生态中的重要组件，面向分布式服务架构的流量控制组件，主要以流量为切入点，从限流、流量整形、熔断降级、系统负载保护、热点防护等多个维度来帮助开发者保障微服务的稳定性。

### 2.3 网关层限流 ###

随着微服务规模的增加，整个系统中很多微服务都需要实现限流这种需求时，就可以考虑在网关这一层进行限流了，通常来说，网关层的限流面向的是通用的业务，比如那些恶意的请求，爬虫，攻击等，简单来说，网关层面的限流提供了一层对系统整体的保护措施

## 3. 四种常用限流算法 ##

### 3.1 算法概述 ###

这里列举几种常用的限流算法以供了解：

1.  固定窗口算法
2.  滑动窗口算法
3.  令牌桶算法
4.  漏桶算法

不管是哪种限流组件，其底层的限流实现算法大同小异，其通用流程为：  
![在这里插入图片描述][d25285b53c4d41b289670567088662fa.png]

1.  统计请求流量：记录请求的数量或速率，可以通过计数器、滑动窗口等方式进行统计。
2.  判断是否超过限制：根据设定的限制条件，判断当前请求流量是否超过限制。
3.  执行限流策略：如果请求流量超过限制，执行限流策略，如拒绝请求、延迟处理、返回错误信息等。
4.  更新统计信息：根据请求的处理结果，更新统计信息，如增加计数器的值、更新滑动窗口的数据等。
5.  重复执行以上步骤：不断地统计请求流量、判断是否超过限制、执行限流策略、更新统计信息

> 需要注意的是，具体的限流算法实现可能会根据不同的场景和需求进行调整和优化，比如使用令牌桶算法、漏桶算法等

### 3.2 算法实现 ###

接下来，我们来实现上述 4 种常见的限流算法，这里使用 Redis 作为分布式存储，Redission 作为 Redis 客户端

1、引入依赖

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.16.2</version>
    </dependency>

2、用单例模式获取RedissonClient

public class RedissonConfig {
        
    
        private static final String REDIS_ADDRESS = "redis://127.0.0.1:6379";
    
        private static volatile RedissonClient redissonClient;
    
        public static RedissonClient getInstance(){
        
            if (redissonClient == null){
        
                synchronized (RedissonConfig.class){
        
                    if (redissonClient == null){
        
                        Config config = new Config();
                        config.useSingleServer().setAddress(REDIS_ADDRESS);
                        redissonClient = Redisson.create(config);
                        return redissonClient;
                    }
                }
            }
            return redissonClient;
        }
    
    }

#### 3.2.1 固定窗口限流算法 ####

**固定窗口算法**（计数器算法）：是一种比较简单的限流算法，它把时间划分为固定的时间窗口，每个窗口内允许的请求次数设置限制。如果在一个时间窗口内，请求次数超过了上限，那么就会触发限流  
![在这里插入图片描述][1ab2030c1fa94c71b0a1633d4ed360f3.png]

##### 算法实现 #####

基于 Redisson 的实现固定窗口相当简单。在每个窗口期内，我们可以通过 `incrementAndGet` 操作来统计请求的数量。一旦窗口期结束，我们可以利用 Redis 的键过期功能来自动重置计数

public class FixedWindowRateLimiter {
        
    
        public static final String KEY = "fixedWindowRateLimiter:";
    
        // 请求限制数量
        private Long limit;
    
        // 窗口大小 秒
        private Long windowSize;
    
        public FixedWindowRateLimiter(Long limit, Long windowSize) {
        
            this.limit = limit;
            this.windowSize = windowSize;
        }
    
        public boolean triggerLimit(String path) {
        
            RedissonClient redissonClient = RedissonConfig.getInstance();
            //加分布式锁，防止并发情况下窗口初始化时间不一致问题
            RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
            try {
        
                rLock.lock(100, TimeUnit.MILLISECONDS);
                String redisKey = KEY + path;
                RAtomicLong counter = redissonClient.getAtomicLong(redisKey);
                //计数
                long count = counter.incrementAndGet();
                //如果为1的话，就说明窗口刚初始化
                if (count == 1) {
        
                    //直接设置过期时间，作为窗口
                    counter.expire(windowSize, TimeUnit.SECONDS);
                }
                //触发限流
                if (count > limit) {
        
                    //触发限流的不记在请求数量中
                    counter.decrementAndGet();
                    return true;
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                rLock.unlock();
            }
            return false;
        }
    
    }

测试：

public class FixedWindowRateLimiterTest {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(20, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
            FixedWindowRateLimiter fixedWindowRateLimiter = new FixedWindowRateLimiter(10L,60L);
            //模拟不同窗口内的调用
            for (int i = 0; i < 3; i++) {
        
                CountDownLatch countDownLatch = new CountDownLatch(20);
                //20个线程并发调用
                for (int j = 0; j < 20; j++) {
        
                    threadPoolExecutor.execute(() -> {
        
                        boolean isLimit = fixedWindowRateLimiter.triggerLimit("/test");
                        System.out.println(isLimit);
                        countDownLatch.countDown();
                    });
                }
                countDownLatch.await();
                //休眠1min
                TimeUnit.MINUTES.sleep(1);
            }
        }
    
    }

固定窗口算法的优点是实现简单，占用空间小，但是它存在临界问题，由于窗口的切换是瞬间完成的，因此请求的处理并不平滑，可能会在窗口切换的瞬间出现流量的剧烈波动。

比如这个例子，假如在00:02，突然有大量请求过来，但是我们这时候计数重置了，那么就没法限制突发的这些流量。  
![在这里插入图片描述][97b4c85feb95478285269bf0cd91b5e5.png]

#### 3.2.2 滑动窗口算法 ####

为了缓解固定窗口的突发流量问题，可以采用滑动窗口算法，计算机网络中TCP的流量控制就是采用滑动窗口算法

**滑动窗口限流算法**：将一个大的时间窗口划分为多个小的时间窗口，每个小的窗口都有独立的计数。请求过来的时候，判断请求的次数是否超过整个窗口的限制。窗口的移动是每次向前滑动一个小的单元窗口

例如：下面这个滑动窗口，将大时间窗口1min分成了5个小窗口，每个小窗口的时间是12s。每个单元格有自己独立的计数器，每过12s就会向前移动一格。

假如有请求在00:01的时候过来，这时候窗口的计数就是3+12+9+15=39，也能起到限流的作用

![在这里插入图片描述][ebe014a5ea8a4dc5ae5c600b8fc0bc80.png]  
这就是为什么滑动窗口能解决临界问题，滑的格子越多，那么整体的滑动就会越平滑,限流的效果就会越精准。

##### 算法实现 #####

那么我们这里怎么实现滑动窗口限流算法呢？非常简单，我们可以直接使用Redis的有序集合（zset）结构。

我们使用时间戳作为 score 和 member，有请求过来的时候，就把当前时间戳添加到有序集合里。那么窗口之外的请求，我们可以根据窗口大小，计算出起始时间戳，删除窗口外的请求。这样，有序集合的大小，就是我们这个窗口的请求数了。  
![在这里插入图片描述][a763d957594449d0afc765d4e7a0f803.png]

public class SlidingWindowRateLimiter {
        
    
        public static final String KEY = "fixedWindowRateLimiter:";
    
        // 请求限制数量
        private Long limit;
    
        // 窗口大小 秒
        private Long windowSize;
    
        public SlidingWindowRateLimiter(Long limit, Long windowSize) {
        
            this.limit = limit;
            this.windowSize = windowSize;
        }
    
        public boolean triggerLimit(String path) {
        
            RedissonClient redissonClient = RedissonConfig.getInstance();
            // 使用分布式锁，避免并发设置初始值的时候，导致窗口计数被覆盖
            RLock rLock = redissonClient.getLock(KEY + "LOCK:" + path);
            // 窗口计数
            RScoredSortedSet<Long> counter = redissonClient.getScoredSortedSet(KEY + path);
            try {
        
                rLock.lock(200, TimeUnit.MILLISECONDS);
                long currentTimestamp = System.currentTimeMillis();
                // 窗口起始时间戳
                long windowStartTimestamp = currentTimestamp - windowSize * 1000;
                // 移除窗口外的时间戳，左闭右开
                counter.removeRangeByScore(0, true, windowStartTimestamp, false);
                // 将当前时间戳作为score,也作为member，
                // TODO:高并发情况下可能没法保证唯一，可以加一个唯一标识
                counter.add(currentTimestamp, currentTimestamp);
                //使用zset的元素个数，作为请求计数
                long count = counter.size();
                // 判断时间戳数量是否超过限流阈值
                if (count > limit) {
        
                    System.out.println("[triggerLimit] path:" + path + " count:" + count + " over limit:" + limit);
                    return true;
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                rLock.unlock();
            }
            return false;
        }
        
    }

这里还有一个小的可以完善的点，zset 在 member 相同的情况下，是会覆盖的，也就是说高并发情况下，时间戳可能会重复，那么就有可能统计的请求偏少，这里可以用时间戳+随机数来缓解，也可以生成唯一序列来解决，比如 UUID、雪花算法等等。

测试：

public class SlidingWindowRateLimiterTest {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
            SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(10L, 1L);
            // 模拟在不同时间片内的请求
            for (int i = 0; i < 8; i++) {
        
                CountDownLatch countDownLatch = new CountDownLatch(20);
                for (int j = 0; j < 20; j++) {
        
                    threadPoolExecutor.execute(() -> {
        
                        boolean isLimit = slidingWindowRateLimiter.triggerLimit("/test");
                        System.out.println(isLimit);
                        countDownLatch.countDown();
                    });
                }
                countDownLatch.await();
                //休眠10s
                TimeUnit.SECONDS.sleep(10L);
            }
        }
    
    }

用 Redis 实现了滑动窗口限流，解决了固定窗口限流的边界问题，当然这里也带来了新的问题，因为我们存储了窗口期的所有请求，所以高并发的情况下，可能会比较占内存

#### 3.2.3 漏桶算法 ####

我们可以看到，计数器类的限流，体现的是一个“戛然而止”，超过限制，立马决绝，但是有时候，我们可能只是希望请求平滑一些，追求的是“波澜不惊”，这时候就可以考虑使用其它的限流算法

**漏桶算法**：就是请求就像水一样以任意速度注入漏桶，而桶会按照固定的速率将水漏掉

![在这里插入图片描述][a5d5763ca6514c8ab6c2bae1d81678bd.png]  
当进水速率大于出水速率的时候，漏桶会变满，此时新进入的请求将会被丢弃。

漏桶算法的两大作用是 网络流量整形（Traffic Shaping）和 速度限制（Rate Limiting）

##### 算法实现 #####

在滑动窗口限流算法里我们用到了 `RScoredSortedSet`。这里也可以用这个结构，直接使用 `ZREMRANGEBYSCORE` 命令来删除旧的请求

*  进水就不用多说了，请求进来，判断桶有没有满，满了就拒绝，没满就往桶里丢请求。
 *  那么出水怎么办呢？得保证稳定速率出水，可以用一个定时任务，来定时去删除旧的请求

public class LeakyBucketRateLimiter {
        
    
        private RedissonClient redissonClient = RedissonConfig.getInstance();
        private static final String KEY_PREFIX = "LeakyBucket:";
    
        /**
         * 桶的大小
         */
        private Long bucketSize;
        /**
         * 漏水速率，单位:个/秒
         */
        private Long leakRate;
    
        public LeakyBucketRateLimiter(Long bucketSize, Long leakRate) {
        
            this.bucketSize = bucketSize;
            this.leakRate = leakRate;
            //这里启动一个定时任务，每s执行一次
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
            executorService.scheduleAtFixedRate(this::leakWater, 0, 1, TimeUnit.SECONDS);
        }
    
        /**
         * 漏水
         *
         * @author zzc
         * @date 2023/7/19 17:16
         */
        public void leakWater() {
        
            RSet<String> pathSet = redissonClient.getSet(KEY_PREFIX + ":pathSet");
            //遍历所有path,删除旧请求
            for(String path : pathSet){
        
                RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(KEY_PREFIX + path);
                // 获取当前时间
                long now = System.currentTimeMillis();
                // 删除旧的请求
                bucket.removeRangeByScore(0, true,now - 1000 * leakRate,true);
            }
        }
    
        public boolean triggerLimit(String path) {
        
            //加锁，防止并发初始化问题
            RLock rLock = redissonClient.getLock(KEY_PREFIX + "LOCK:" + path);
            try {
        
                rLock.lock(100,TimeUnit.MILLISECONDS);
                String redisKey = KEY_PREFIX + path;
                RScoredSortedSet<Long> bucket = redissonClient.getScoredSortedSet(redisKey);
                //这里用一个set，来存储所有path
                RSet<String> pathSet = redissonClient.getSet(KEY_PREFIX + ":pathSet");
                pathSet.add(path);
                // 获取当前时间
                long now = System.currentTimeMillis();
                // 检查桶是否已满
                if (bucket.size() < bucketSize) {
        
                    // 桶未满，添加一个元素到桶中
                    bucket.add(now, now);
                    return false;
                }
                // 桶已满，触发限流
                System.out.println("[triggerLimit] path:"+path+" bucket size:"+bucket.size());
                return true;
            } finally {
        
                rLock.unlock();
            }
        }
    
    }

在代码实现里，我们用了 RSet 来存储 path，这样一来，一个定时任务，就可以搞定所有 path 对应的桶的出水，而不用每个桶都创建一个一个定时任务。

这里我直接用 ScheduledExecutorService 启动了一个定时任务，1s跑一次，当然集群环境下，每台机器都跑一个定时任务，对性能是极大的浪费，而且不好管理，我们可以用分布式定时任务，比如 xxl-job 去执行 leakWater。

测试：

public class LeakyBucketRateLimiterTest {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
            LeakyBucketRateLimiter leakyBucketRateLimiter = new LeakyBucketRateLimiter(10L, 1L);
            for (int i = 0; i < 8; i++) {
        
                CountDownLatch countDownLatch = new CountDownLatch(20);
                for (int j = 0; j < 20; j++) {
        
                    threadPoolExecutor.execute(() -> {
        
                        boolean isLimit = leakyBucketRateLimiter.triggerLimit("/test");
                        System.out.println(isLimit);
                        countDownLatch.countDown();
                    });
                }
                countDownLatch.await();
                //休眠10s
                TimeUnit.SECONDS.sleep(10L);
            }
        }
    
    }

漏桶算法能够有效防止网络拥塞，实现也比较简单。

但是，因为漏桶的出水速率是固定的，假如突然来了大量的请求，那么只能丢弃超量的请求，即使下游能处理更大的流量，没法充分利用系统资源。

#### 3.2.4 令牌桶算法 ####

**令牌桶算法**：系统以一种固定的速率向桶中添加令牌，每个请求在发送前都需要从桶中取出一个令牌，只有取到令牌的请求才被通过。因此，令牌桶算法允许请求以任意速率发送，只要桶中有足够的令牌  
![在这里插入图片描述][58d66e338ab64149b1d646ca50b40d77.png]

##### 算法实现 #####

首先是要发放令牌，要固定速率，那我们又得开个线程，定时往桶里投令牌，然后 Redission 提供了令牌桶算法的实现。

public class TokenBucketRateLimiter {
        
    
        public static final String KEY = "TokenBucketRateLimiter:";
    
        /**
         * 阈值
         */
        private Long limit;
    
        /**
         * 添加令牌的速率，单位：个/秒
         */
        private Long tokenRate;
    
        public TokenBucketRateLimiter(Long limit, Long tokenRate) {
        
            this.limit = limit;
            this.tokenRate = tokenRate;
        }
    
        /**
         * 限流算法
         */
        public boolean triggerLimit(String path){
        
            RedissonClient redissonClient = RedissonConfig.getInstance();
            RRateLimiter rateLimiter = redissonClient.getRateLimiter(KEY + path);
            // 初始化，设置速率模式，速率，间隔，间隔单位
            rateLimiter.trySetRate(RateType.OVERALL, limit, tokenRate, RateIntervalUnit.SECONDS);
            // 获取令牌
            return rateLimiter.tryAcquire();
        }
    
    }

测试：

public class TokenBucketRateLimiterTest {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 50, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
            TokenBucketRateLimiter tokenBucketRateLimiter = new TokenBucketRateLimiter(10L, 1L);
            for (int i = 0; i < 8; i++) {
        
                CountDownLatch countDownLatch = new CountDownLatch(20);
                for (int j = 0; j < 20; j++) {
        
                    threadPoolExecutor.execute(() -> {
        
                        boolean isLimit = tokenBucketRateLimiter.triggerLimit("/test");
                        System.out.println(isLimit);
                        countDownLatch.countDown();
                    });
                }
                countDownLatch.await();
                //休眠10s
                TimeUnit.SECONDS.sleep(10L);
            }
        }
    
    }

## 总结 ##

在这篇文章里，我们对四种限流算法进行了分布式实现，采用了非常好用的 `Redission` 客户端。

当然我们也有不完善的地方：

*  并发处理采用了分布式锁，高并发情况下，对性能有一定损耗，逻辑最好还是直接采用Lua脚本实现，来提高性能
 *  可以提供更加优雅的调用方式，比如利用 aop 实现注解式调用，代码设计也可以更加优雅，继承体系可以完善一下
 *  没有实现限流的拒绝策略，比如抛异常、缓存、丢进MQ打散……限流是一种方法，最终的目的还是尽可能保证系统平稳

除此之外，市面上也有很多好用的开源限流工具：

*  Guava RateLimiter ，基于令牌桶算法限流，当然是单机的；
 *  Sentinel ，基于滑动窗口限流，支持单机，也支持集群
 *  网关限流，很多网关自带限流方法，比如Spring Cloud Gateway、Nginx

[d25285b53c4d41b289670567088662fa.png]: https://img-blog.csdnimg.cn/d25285b53c4d41b289670567088662fa.png
[1ab2030c1fa94c71b0a1633d4ed360f3.png]: https://img-blog.csdnimg.cn/1ab2030c1fa94c71b0a1633d4ed360f3.png
[97b4c85feb95478285269bf0cd91b5e5.png]: https://img-blog.csdnimg.cn/97b4c85feb95478285269bf0cd91b5e5.png
[ebe014a5ea8a4dc5ae5c600b8fc0bc80.png]: https://img-blog.csdnimg.cn/ebe014a5ea8a4dc5ae5c600b8fc0bc80.png
[a763d957594449d0afc765d4e7a0f803.png]: https://img-blog.csdnimg.cn/a763d957594449d0afc765d4e7a0f803.png
[a5d5763ca6514c8ab6c2bae1d81678bd.png]: https://img-blog.csdnimg.cn/a5d5763ca6514c8ab6c2bae1d81678bd.png
[58d66e338ab64149b1d646ca50b40d77.png]: https://img-blog.csdnimg.cn/58d66e338ab64149b1d646ca50b40d77.png