亲宝软件园·资讯

展开

Java实现限流算法

程序猿阿朗 人气:4

前言

最近几年，随着微服务的流行，服务和服务之间的依赖越来越强，调用关系越来越复杂，服务和服务之间的稳定性越来越重要。在遇到突发的请求量激增，恶意的用户访问，亦或请求频率过高给下游服务带来较大压力时，我们常常需要通过缓存、限流、熔断降级、负载均衡等多种方式保证服务的稳定性。其中限流是不可或缺的一环，这篇文章介绍限流相关知识。

1. 限流

限流顾名思义，就是对请求或并发数进行限制；通过对一个时间窗口内的请求量进行限制来保障系统的正常运行。如果我们的服务资源有限、处理能力有限，就需要对调用我们服务的上游请求进行限制，以防止自身服务由于资源耗尽而停止服务。

在限流中有两个概念需要了解。

2. 固定窗口算法

固定窗口算法又叫计数器算法，是一种简单方便的限流算法。主要通过一个支持原子操作的计数器来累计 1 秒内的请求次数，当 1 秒内计数达到限流阈值时触发拒绝策略。每过 1 秒，计数器重置为 0 开始重新计数。

2.1. 代码实现

下面是简单的代码实现，QPS 限制为 2，这里的代码做了一些优化，并没有单独开一个线程去每隔 1 秒重置计数器，而是在每次调用时进行时间间隔计算来确定是否先重置计数器。

/**
 * @author https://www.wdbyte.com
 */
public class RateLimiterSimpleWindow {
    // 阈值
    private static Integer QPS = 2;
    // 时间窗口（毫秒）
    private static long TIME_WINDOWS = 1000;
    // 计数器
    private static AtomicInteger REQ_COUNT = new AtomicInteger();
    
    private static long START_TIME = System.currentTimeMillis();
 
    public synchronized static boolean tryAcquire() {
        if ((System.currentTimeMillis() - START_TIME) > TIME_WINDOWS) {
            REQ_COUNT.set(0);
            START_TIME = System.currentTimeMillis();
        }
        return REQ_COUNT.incrementAndGet() <= QPS;
    }
 
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            Thread.sleep(250);
            LocalTime now = LocalTime.now();
            if (!tryAcquire()) {
                System.out.println(now + " 被限流");
            } else {
                System.out.println(now + " 做点什么");
            }
        }
    }
}

运行结果：

20:53:43.038922 做点什么
20:53:43.291435 做点什么
20:53:43.543087 被限流
20:53:43.796666 做点什么
20:53:44.050855 做点什么
20:53:44.303547 被限流
20:53:44.555008 被限流
20:53:44.809083 做点什么
20:53:45.063828 做点什么
20:53:45.314433 被限流

从输出结果中可以看到大概每秒操作 3 次，由于限制 QPS 为 2，所以平均会有一次被限流。看起来可以了，不过我们思考一下就会发现这种简单的限流方式是有问题的，虽然我们限制了 QPS 为 2，但是当遇到时间窗口的临界突变时，如 1s 中的后 500 ms 和第 2s 的前 500ms 时，虽然是加起来是 1s 时间，却可以被请求 4 次。

固定窗口算法

简单修改测试代码，可以进行验证：

// 先休眠 400ms，可以更快的到达时间窗口。
Thread.sleep(400);
for (int i = 0; i < 10; i++) {
    Thread.sleep(250);
    if (!tryAcquire()) {
        System.out.println("被限流");
    } else {
        System.out.println("做点什么");
    }
}

得到输出中可以看到连续 4 次请求，间隔 250 ms 没有却被限制。：

20:51:17.395087 做点什么
20:51:17.653114 做点什么
20:51:17.903543 做点什么
20:51:18.154104 被限流
20:51:18.405497 做点什么
20:51:18.655885 做点什么
20:51:18.906177 做点什么
20:51:19.158113 被限流
20:51:19.410512 做点什么
20:51:19.661629 做点什么

3. 滑动窗口算法

我们已经知道固定窗口算法的实现方式以及它所存在的问题，而滑动窗口算法是对固定窗口算法的改进。既然固定窗口算法在遇到时间窗口的临界突变时会有问题，那么我们在遇到下一个时间窗口前也调整时间窗口不就可以了吗？

下面是滑动窗口的示意图。

滑动窗口算法

上图的示例中，每 500ms 滑动一次窗口，可以发现窗口滑动的间隔越短，时间窗口的临界突变问题发生的概率也就越小，不过只要有时间窗口的存在，还是有可能发生时间窗口的临界突变问题。

3.1. 代码实现

下面是基于以上滑动窗口思路实现的简单的滑动窗口限流工具类。

package com.wdbyte.rate.limiter;
 
import java.time.LocalTime;
import java.util.concurrent.atomic.AtomicInteger;
 
/**
 * 滑动窗口限流工具类
 *
 * @author https://www.wdbyte.com
 */
public class RateLimiterSlidingWindow {
    /**
     * 阈值
     */
    private int qps = 2;
    /**
     * 时间窗口总大小（毫秒）
     */
    private long windowSize = 1000;
    /**
     * 多少个子窗口
     */
    private Integer windowCount = 10;
    /**
     * 窗口列表
     */
    private WindowInfo[] windowArray = new WindowInfo[windowCount];
 
    public RateLimiterSlidingWindow(int qps) {
        this.qps = qps;
        long currentTimeMillis = System.currentTimeMillis();
        for (int i = 0; i < windowArray.length; i++) {
            windowArray[i] = new WindowInfo(currentTimeMillis, new AtomicInteger(0));
        }
    }
 
    /**
     * 1. 计算当前时间窗口
     * 2. 更新当前窗口计数 & 重置过期窗口计数
     * 3. 当前 QPS 是否超过限制
     *
     * @return
     */
    public synchronized boolean tryAcquire() {
        long currentTimeMillis = System.currentTimeMillis();
        // 1. 计算当前时间窗口
        int currentIndex = (int)(currentTimeMillis % windowSize / (windowSize / windowCount));
        // 2.  更新当前窗口计数 & 重置过期窗口计数
        int sum = 0;
        for (int i = 0; i < windowArray.length; i++) {
            WindowInfo windowInfo = windowArray[i];
            if ((currentTimeMillis - windowInfo.getTime()) > windowSize) {
                windowInfo.getNumber().set(0);
                windowInfo.setTime(currentTimeMillis);
            }
            if (currentIndex == i && windowInfo.getNumber().get() < qps) {
                windowInfo.getNumber().incrementAndGet();
            }
            sum = sum + windowInfo.getNumber().get();
        }
        // 3. 当前 QPS 是否超过限制
        return sum <= qps;
    }
 
    private class WindowInfo {
        // 窗口开始时间
        private Long time;
        // 计数器
        private AtomicInteger number;
 
        public WindowInfo(long time, AtomicInteger number) {
            this.time = time;
            this.number = number;
        }
        // get...set...
    }
}

下面是测试用例，设置 QPS 为 2,测试次数 20 次，每次间隔 300 毫秒，预计成功次数在 12 次左右。

public static void main(String[] args) throws InterruptedException {
    int qps = 2, count = 20, sleep = 300, success = count * sleep / 1000 * qps;
    System.out.println(String.format("当前QPS限制为:%d,当前测试次数:%d,间隔:%dms,预计成功次数:%d", qps, count, sleep, success));
    success = 0;
    RateLimiterSlidingWindow myRateLimiter = new RateLimiterSlidingWindow(qps);
    for (int i = 0; i < count; i++) {
        Thread.sleep(sleep);
        if (myRateLimiter.tryAcquire()) {
            success++;
            if (success % qps == 0) {
                System.out.println(LocalTime.now() + ": success, ");
            } else {
                System.out.print(LocalTime.now() + ": success, ");
            }
        } else {
            System.out.println(LocalTime.now() + ": fail");
        }
    }
    System.out.println();
    System.out.println("实际测试成功次数:" + success);
}

下面是测试的结果。

当前QPS限制为:2,当前测试次数:20,间隔:300ms,预计成功次数:12
16:04:27.077782: success, 16:04:27.380715: success, 
16:04:27.684244: fail
16:04:27.989579: success, 16:04:28.293347: success, 
16:04:28.597658: fail
16:04:28.901688: fail
16:04:29.205262: success, 16:04:29.507117: success, 
16:04:29.812188: fail
16:04:30.115316: fail
16:04:30.420596: success, 16:04:30.725897: success, 
16:04:31.028599: fail
16:04:31.331047: fail
16:04:31.634127: success, 16:04:31.939411: success, 
16:04:32.242380: fail
16:04:32.547626: fail
16:04:32.847965: success, 
实际测试成功次数:11

4. 滑动日志算法

滑动日志算法是实现限流的另一种方法，这种方法比较简单。基本逻辑就是记录下所有的请求时间点，新请求到来时先判断最近指定时间范围内的请求数量是否超过指定阈值，由此来确定是否达到限流，这种方式没有了时间窗口突变的问题，限流比较准确，但是因为要记录下每次请求的时间点，所以占用的内存较多。

4.1. 代码实现

下面是简单实现的 一个滑动日志算法，因为滑动日志要每次请求单独存储一条记录，可能占用内存过多。所以下面这个实现其实不算严谨的滑动日志，更像一个把 1 秒时间切分成 1000 个时间窗口的滑动窗口算法。

package com.wdbyte.rate.limiter;
 
import java.time.LocalTime;
import java.util.HashSet;
import java.util.Set;
import java.util.TreeMap;
 
/**
 * 滑动日志方式限流
 * 设置 QPS 为 2.
 *
 * @author https://www.wdbyte.com
 */
public class RateLimiterSildingLog {
 
    /**
     * 阈值
     */
    private Integer qps = 2;
    /**
     * 记录请求的时间戳,和数量
     */
    private TreeMap<Long, Long> treeMap = new TreeMap<>();
 
    /**
     * 清理请求记录间隔, 60 秒
     */
    private long claerTime = 60 * 1000;
 
    public RateLimiterSildingLog(Integer qps) {
        this.qps = qps;
    }
 
    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        // 清理过期的数据老数据，最长 60 秒清理一次
        if (!treeMap.isEmpty() && (treeMap.firstKey() - now) > claerTime) {
            Set<Long> keySet = new HashSet<>(treeMap.subMap(0L, now - 1000).keySet());
            for (Long key : keySet) {
                treeMap.remove(key);
            }
        }
        // 计算当前请求次数
        int sum = 0;
        for (Long value : treeMap.subMap(now - 1000, now).values()) {
            sum += value;
        }
        // 超过QPS限制，直接返回 false
        if (sum + 1 > qps) {
            return false;
        }
        // 记录本次请求
        if (treeMap.containsKey(now)) {
            treeMap.compute(now, (k, v) -> v + 1);
        } else {
            treeMap.put(now, 1L);
        }
        return sum <= qps;
    }
 
    public static void main(String[] args) throws InterruptedException {
        RateLimiterSildingLog rateLimiterSildingLog = new RateLimiterSildingLog(3);
        for (int i = 0; i < 10; i++) {
            Thread.sleep(250);
            LocalTime now = LocalTime.now();
            if (rateLimiterSildingLog.tryAcquire()) {
                System.out.println(now + " 做点什么");
            } else {
                System.out.println(now + " 被限流");
            }
        }
    }
}

代码中把阈值 QPS 设定为 3，运行可以得到如下日志：

20:51:17.395087 做点什么
20:51:17.653114 做点什么
20:51:17.903543 做点什么
20:51:18.154104 被限流
20:51:18.405497 做点什么
20:51:18.655885 做点什么
20:51:18.906177 做点什么
20:51:19.158113 被限流
20:51:19.410512 做点什么
20:51:19.661629 做点什么

5. 漏桶算法

漏桶算法中的漏桶是一个形象的比喻，这里可以用生产者消费者模式进行说明，请求是一个生产者，每一个请求都如一滴水，请求到来后放到一个队列（漏桶）中，而桶底有一个孔，不断的漏出水滴，就如消费者不断的在消费队列中的内容，消费的速率（漏出的速度）等于限流阈值。即假如 QPS  为 2，则每 1s / 2= 500ms 消费一次。漏桶的桶有大小，就如队列的容量，当请求堆积超过指定容量时，会触发拒绝策略。

下面是漏桶算法的示意图。

漏桶算法

由介绍可以知道，漏桶模式中的消费处理总是能以恒定的速度进行，可以很好的保护自身系统不被突如其来的流量冲垮；但是这也是漏桶模式的缺点，假设 QPS 为 2，同时 2 个请求进来，2 个请求并不能同时进行处理响应，因为每 1s / 2= 500ms 只能处理一个请求。

6. 令牌桶算法

令牌桶算法同样是实现限流是一种常见的思路，最为常用的 Google 的 Java 开发工具包 Guava 中的限流工具类 RateLimiter 就是令牌桶的一个实现。令牌桶的实现思路类似于生产者和消费之间的关系。

系统服务作为生产者，按照指定频率向桶（容器）中添加令牌，如 QPS 为 2，每 500ms 向桶中添加一个令牌，如果桶中令牌数量达到阈值，则不再添加。

请求执行作为消费者，每个请求都需要去桶中拿取一个令牌，取到令牌则继续执行；如果桶中无令牌可取，就触发拒绝策略，可以是超时等待，也可以是直接拒绝本次请求，由此达到限流目的。

下面是令牌桶限流算法示意图。

令牌桶算法

思考令牌桶的实现可以以下特点。

6.1. 代码实现

Google 的 Java 开发工具包 Guava 中的限流工具类 RateLimiter 就是令牌桶的一个实现，日常开发中我们也不会手动实现了，这里直接使用 RateLimiter 进行测试。

引入依赖：

<exclusion>
   <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
   <version>31.0.1-jre</version>
</exclusion>

RateLimiter 限流体验：

// qps 2
RateLimiter rateLimiter = RateLimiter.create(2);
for (int i = 0; i < 10; i++) {
    String time = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME);
    System.out.println(time + ":" + rateLimiter.tryAcquire());
    Thread.sleep(250);
}

代码中限制 QPS 为 2，也就是每隔 500ms 生成一个令牌，但是程序每隔 250ms 获取一次令牌，所以两次获取中只有一次会成功。

17:19:06.797557:true
17:19:07.061419:false
17:19:07.316283:true
17:19:07.566746:false
17:19:07.817035:true
17:19:08.072483:false
17:19:08.326347:true
17:19:08.577661:false
17:19:08.830252:true
17:19:09.085327:false

6.2. 思考

虽然演示了 Google Guava 工具包中的 RateLimiter 的实现，但是我们需要思考一个问题，就是令牌的添加方式，如果按照指定间隔添加令牌，那么需要开一个线程去定时添加，如果有很多个接口很多个 RateLimiter 实例，线程数会随之增加，这显然不是一个好的办法。显然 Google 也考虑到了这个问题，在 RateLimiter 中，是在每次令牌获取时才进行计算令牌是否足够的。它通过存储的下一个令牌生成的时间，和当前获取令牌的时间差，再结合阈值，去计算令牌是否足够，同时再记录下一个令牌的生成时间以便下一次调用。

下面是 Guava 中 RateLimiter 类的子类 SmoothRateLimiter 的 resync() 方法的代码分析，可以看到其中的令牌计算逻辑。

void resync(long nowMicros) { // 当前微秒时间
    // 当前时间是否大于下一个令牌生成时间
    if (nowMicros > this.nextFreeTicketMicros) { 
       // 可生成的令牌数 newPermits = （当前时间 - 下一个令牌生成时间）/ 令牌生成时间间隔。
       // 如果 QPS 为2，这里的 coolDownIntervalMicros 就是 500000.0 微秒(500ms)
        double newPermits = (double)(nowMicros - this.nextFreeTicketMicros) / this.coolDownIntervalMicros();
    // 更新令牌库存 storedPermits。
       this.storedPermits = Math.min(this.maxPermits, this.storedPermits + newPermits);
    // 更新下一个令牌生成时间 nextFreeTicketMicros
       this.nextFreeTicketMicros = nowMicros;
    }
}

7. Redis 分布式限流

Redis 是一个开源的内存数据库，可以用来作为数据库、缓存、消息中间件等。Redis 是单线程的，又在内存中操作，所以速度极快，得益于 Redis 的各种特性，所以使用 Redis 实现一个限流工具是十分方便的。

下面的演示都基于Spring Boot 项目，并需要以下依赖。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

配置 Redis 信息。

spring:
  redis:
    database: 0
    password: 
    port: 6379
    host: 127.0.0.1
    lettuce:
      shutdown-timeout: 100ms
      pool:
        min-idle: 5
        max-idle: 10
        max-active: 8
        max-wait: 1ms

7.1. 固定窗口限流

Redis 中的固定窗口限流是使用 incr 命令实现的，incr 命令通常用来自增计数；如果我们使用时间戳信息作为 key，自然就可以统计每秒的请求量了，以此达到限流目的。

这里有两点要注意。

由于 Redis 中实现事务的复杂性，所以这里直接只用 lua 脚本来实现原子操作。下面是 lua 脚本内容。

local count = redis.call("incr",KEYS[1])
if count == 1 then
  redis.call('expire',KEYS[1],ARGV[2])
end
if count > tonumber(ARGV[1]) then
  return 0
end
return 1

下面是使用 Spring Boot 中 RedisTemplate 来实现的 lua 脚本调用测试代码。

/**
 * @author https://www.wdbyte.com
 */
@SpringBootTest
class RedisLuaLimiterByIncr {
    private static String KEY_PREFIX = "limiter_";
    private static String QPS = "4";
    private static String EXPIRE_TIME = "1";
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
 
    @Test
    public void redisLuaLimiterTests() throws InterruptedException, IOException {
        for (int i = 0; i < 15; i++) {
            Thread.sleep(200);
            System.out.println(LocalTime.now() + " " + acquire("user1"));
        }
    }
 
    /**
     * 计数器限流
     *
     * @param key
     * @return
     */
    public boolean acquire(String key) {
        // 当前秒数作为 key
        key = KEY_PREFIX + key + System.currentTimeMillis() / 1000;
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Long.class);
        //lua文件存放在resources目录下
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("limiter.lua")));
        return stringRedisTemplate.execute(redisScript, Arrays.asList(key), QPS, EXPIRE_TIME) == 1;
    }
}

代码中虽然限制了 QPS 为 4，但是因为这种限流实现是把毫秒时间戳作为 key 的，所以会有临界窗口突变的问题，下面是运行结果，可以看到因为时间窗口的变化，导致了 QPS 超过了限制值 4。

17:38:23.122044 true
17:38:23.695124 true
17:38:23.903220 true
# 此处有时间窗口变化，所以下面继续 true
17:38:24.106206 true
17:38:24.313458 true
17:38:24.519431 true
17:38:24.724446 true
17:38:24.932387 false
17:38:25.137912 true
17:38:25.355595 true
17:38:25.558219 true
17:38:25.765801 true
17:38:25.969426 false
17:38:26.176220 true
17:38:26.381918 true

7.3. 滑动窗口限流

通过对上面的基于 incr 命令实现的 Redis 限流方式的测试，我们已经发现了固定窗口限流所带来的问题，在这篇文章的第三部分已经介绍了滑动窗口限流的优势，它可以大幅度降低因为窗口临界突变带来的问题，那么如何使用 Redis 来实现滑动窗口限流呢？

这里主要使用 ZSET 有序集合来实现滑动窗口限流，ZSET 集合有下面几个特点：

基于上面的四点特性，可以编写出基于 ZSET 的滑动窗口限流 lua 脚本。

--KEYS[1]: 限流 key
--ARGV[1]: 时间戳 - 时间窗口
--ARGV[2]: 当前时间戳（作为score）
--ARGV[3]: 阈值
--ARGV[4]: score 对应的唯一value
-- 1. 移除时间窗口之前的数据
redis.call('zremrangeByScore', KEYS[1], 0, ARGV[1])
-- 2. 统计当前元素数量
local res = redis.call('zcard', KEYS[1])
-- 3. 是否超过阈值
if (res == nil) or (res < tonumber(ARGV[3])) then
    redis.call('zadd', KEYS[1], ARGV[2], ARGV[4])
    return 1
else
    return 0
end

下面是使用 Spring Boot 中 RedisTemplate 来实现的 lua 脚本调用测试代码。

@SpringBootTest
class RedisLuaLimiterByZset {
 
    private String KEY_PREFIX = "limiter_";
    private String QPS = "4";
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
 
    @Test
    public void redisLuaLimiterTests() throws InterruptedException, IOException {
        for (int i = 0; i < 15; i++) {
            Thread.sleep(200);
            System.out.println(LocalTime.now() + " " + acquire("user1"));
        }
    }
 
    /**
     * 计数器限流
     *
     * @param key
     * @return
     */
    public boolean acquire(String key) {
        long now = System.currentTimeMillis();
        key = KEY_PREFIX + key;
        String oldest = String.valueOf(now - 1_000);
        String score = String.valueOf(now);
        String scoreValue = score;
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Long.class);
        //lua文件存放在resources目录下
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("limiter2.lua")));
        return stringRedisTemplate.execute(redisScript, Arrays.asList(key), oldest, score, QPS, scoreValue) == 1;
    }
}

代码中限制 QPS 为 4，运行结果信息与之一致。

17:36:37.150370 true
17:36:37.716341 true
17:36:37.922577 true
17:36:38.127497 true
17:36:38.335879 true
17:36:38.539225 false
17:36:38.745903 true
17:36:38.952491 true
17:36:39.159497 true
17:36:39.365239 true
17:36:39.570572 false
17:36:39.776635 true
17:36:39.982022 true
17:36:40.185614 true
17:36:40.389469 true

这里介绍了 Redis 实现限流的两种方式，当然使用 Redis 也可以实现漏桶和令牌桶两种限流算法，这里就不做演示了，感兴趣的可以自己研究下。

8. 总结

这篇文章介绍实现限流的几种方式，主要是窗口算法和桶算法，两者各有优势。

单机限流与分布式限流

上面演示的基于代码形式的窗口算法和桶算法限流都适用于单机限流，如果需要分布式限流可以结合注册中心、负载均衡计算每个服务的限流阈值，但这样会降低一定精度，如果对精度要求不是太高，可以使用。

而 Redis 的限流，由于 Redis 的单机性，本身就可以用于分布式限流。使用 Redis 可以实现各种可以用于限流算法，如果觉得麻烦也可以使用开源工具如 redisson，已经封装了基于 Redis 的限流。

其他限流工具

文中已经提到了 Guava 的限流工具包，不过它毕竟是单机的，开源社区中也有很多分布式限流工具，如阿里开源的 Sentinel 就是不错的工具，Sentinel 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。

一如既往，文章中的代码存放在：github.com/niumoo/JavaNotes

参考

Redis INCR:https://redis.io/commands/incr

Rate Limiting Wikipedia：https://en.wikipedia.org/wiki/Rate_limiting

SpringBoot Redis:https://www.cnblogs.com/lenve/p/10965667.html

加载全部内容
相关教程
猜你喜欢
用户评论