浅谈限流算法

本文介绍几种常见的限流算法及其在Java下的实现方式

计数器

固定窗口计数器

为了实现流量控制，一个朴素的想法是直接对时间窗口内的请求数量进行统计。如果未达到阈值，则放行请求；反之则对请求进行拒绝。当该时间窗口过去后，则直接将计数器清零。即所谓的固定窗口计数器。该算法的示意图如下所示，可以看到在时间窗口3中，由于计数器值为5已经达到阈值，故对于第6个请求进行限流

而在具体实现过程中，可通过定时线程在到达一个新时间窗口时将计数器清零。Java实现如下所示

/**
 * 限流算法: 固定时间窗口计数器
 */
public class FixedWindowCounterLimiter {

    /**
     *  时间窗口大小, Unit: s
     */
    private int windowSize;

    /**
     * 限流阈值
     */
    private int limit;

    /**
     * 计数器
     */
    private AtomicInteger count;

    public FixedWindowCounterLimiter(int windowSize, int limit) {
        this.windowSize = windowSize;
        this.limit = limit;
        count = new AtomicInteger(0);

        // 启动一个线程, 定时清除计数器值
        new Thread( () -> {
            while (true) {
                try{
                    Thread.sleep(windowSize * 1000);
                }catch (InterruptedException e) {
                    System.out.println("Happen Exception: "+e.getMessage());
                }
                count.set(0);
            }
        }).start();
    }

    public boolean tryAcquire() {
        int num =  count.incrementAndGet();
        // 以达到当前窗口的请求阈值
        if( num > limit ) {
            return false;
        } else {
            return true;
        }
    }
}

测试代码如下所示

/**
 * 测试: 固定时间窗口计数器
 */
@Test
public void test1() throws Exception {
    // 限流配置, 2s内只允许通过5个
    FixedWindowCounterLimiter rateLimiter = new FixedWindowCounterLimiter(2, 5);

    // 请求总数
    int allNum = 3;
    // 通过数
    int passNum = 0;
    // 被限流数
    int blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++){
        if(rateLimiter.tryAcquire()){
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);

    // 延时以准备下一次测试
    Thread.sleep(5000);

    allNum = 14;
    passNum = 0;
    blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++){
        if(rateLimiter.tryAcquire()){
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);
}

测试结果如下，符合预期

滑动窗口计数器

在固定窗口计数器中存在一个明显的缺陷——临界问题。即分别对于时间窗口1、时间窗口2来说，其窗口内的请求数量均未超过阈值。但在下图红框部分的时间范围内，其放行的请求数量却超过了阈值要求

而这里的滑动窗口计数器即是对固定窗口计数器的改良。具体地，其在固定时间窗口的基础上，将其划分若干个小窗口，在各小窗口中对请求数分别进行统计。整个时间窗口随着时间的流逝，不断丢弃过期的小窗口，并将统计结果放在新的小窗口，这也是其被称为滑动窗口的由来。最后根据所有小窗口的统计值之和，来判断是放行请求还是进行限流。算法示意图如下所示，可以看到其在一个限流窗口时间范围内划分为3个子窗口。当子窗口数量越多，则子窗口的时间粒度也就越小，进而统计精度也就越高

在时间窗口刚刚开始滑动时，由于子窗口还未被完全填充。故为便于实现，推荐将对当前的统计计数值始终放在数组最后一个元素中,如上图的T1、T2时刻。具体实现如下所示

/**
 * 限流算法: 滑动窗口计数器
 */
public class SlidingWindowCounterLimiter {
    /**
     * 时间窗口大小, Unit: s
     */
    private int windowSize;

    /**
     * 用于统计的子窗口数量，默认为10
     */
    private int slotNum;

    /**
     * 子窗口的时间长度, Unit: ms
     */
    private int slotTime;

    /**
     * 限流阈值
     */
    private int limit;

    /**
     * 存放子窗口统计结果的数组
     * note: counters[0]记为数组左边, counters[size-1]记为数组右边
     */
    private int[] counters;

    private long lastTime;

    public SlidingWindowCounterLimiter(int windowSize, int limit) {
        this(windowSize, limit, 10);
    }

    public SlidingWindowCounterLimiter(int windowSize, int limit, int slotNum) {
        this.windowSize = windowSize;
        this.limit = limit;
        this.slotNum = slotNum;

        this.counters = new int[slotNum];
        // 计算子窗口的时间长度: 时间窗口 / 子窗口数量
        this.slotTime = windowSize * 1000 / slotNum;
        this.lastTime =  System.currentTimeMillis();
    }

    public synchronized boolean tryAcquire() {
        long currentTime = System.currentTimeMillis();
        // 计算滑动数, 子窗口统计时所对应的时间范围为左闭右开区间, 即[a,b)
        int slideNum = (int) Math.floor( (currentTime-lastTime)/slotTime );
        // 滑动窗口
        slideWindow(slideNum);
        // 统计滑动后的数组之和
        Integer sum = Arrays.stream(counters).sum();

        // 以达到当前时间窗口的请求阈值, 故被限流直接返回false
        if( sum > limit ) {
            return false;
        } else {    // 未达到限流, 故返回true
            counters[slotNum-1]++;
            return true;
        }
    }

    /**
     * 将数组元素全部向左移动num个位置
     * @param num
     */
    private void slideWindow(int num) {
        if( num == 0 ) {
            return;
        }

        // 数组中所有元素都会被移出, 故直接全部清零
        if( num >= slotNum ) {
            Arrays.fill(counters, 0);
        } else {
            // 对于a[0]~a[num-1]而言, 向左移动num个位置后, 则直接被移出了
            // 故从a[num]开始移动即可
            for(int index=num; index<slotNum; index++) {
                // 计算a[index]元素向左移动num个位置后的新位置索引
                int newIndex = index - num;
                counters[newIndex] = counters[index];
                counters[index] = 0;
            }
        }

        // 更新时间
        lastTime = lastTime + num * slotTime;
    }

}

测试代码如下所示

/**
 * 测试: 滑动时间窗口计数器
 */
@Test
public void test2() throws Exception {
    // 限流配置, 2s内只允许通过5个
    SlidingWindowCounterLimiter rateLimiter = new SlidingWindowCounterLimiter(2, 5);

    // 请求总数
    int allNum = 3;
    // 通过数
    int passNum = 0;
    // 被限流数
    int blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++){
        if(rateLimiter.tryAcquire()){
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);

    // 延时以准备下一次测试
    Thread.sleep(5000);

    allNum = 14;
    passNum = 0;
    blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++){
        if(rateLimiter.tryAcquire()){
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);
}

测试结果如下，符合预期

Leaky Bucket 漏桶

As a Meter Version

所谓Leaky Bucket漏桶，指的是向桶中以任意速率注水，而由于该桶底部有一个漏洞，其会以固定的速率流出水。显然如果注水速率过快，桶中水量超过桶容量时即会导致水溢出。而将这一原理应用于限流领域时，不断涌来的请求即相当于向桶中注水，如果桶中水量未超过桶容量，则放行请求；反之则对请求限流。而桶固定的漏水速率则可以理解为系统处理请求流量的能力

具体实现如下所示

/**
 * 漏桶算法: As a Meter Version
 */
public class LeakyBucketLimiter1 {

    /**
     * 桶容量, Unit: 个
     */
    private long capacity;

    /**
     * 出水速率, Unit: 个/秒
     */
    private long rate;

    /**
     * 桶的当前水量
     */
    private long water;

    /**
     * 上次时间
     */
    private long lastTime;

    public LeakyBucketLimiter1(long capacity, long rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.water = 0;
        this.lastTime = System.currentTimeMillis();
    }

    public synchronized boolean tryAcquire() {
        // 获取当前时间
        long currentTime = System.currentTimeMillis();
        // 计算流出的水量: (当前时间-上次时间) * 出水速率
        long outWater = (currentTime-lastTime)/1000 * rate;
        // 计算水量: 桶的当前水量 - 流出的水量
        water = Math.max(0, water-outWater);
        // 更新时间
        lastTime = currentTime;

        // 当前水量 小于 桶容量, 则请求放行, 返回true
        if( water<capacity ) {
            water++;
            return true;
        }else{
            // 当前水量 不小于 桶容量, 则进行限流, 返回false
            return false;
        }
    }

}

测试代码如下所示

/**
 * 测试: 漏桶算法(As a Meter Version)
 */
@Test
public void test3() throws Exception {
    // 漏桶配置, 桶容量:5个, 出水率: 1个/秒
    LeakyBucketLimiter1 rateLimiter = new LeakyBucketLimiter1(5, 1);

    // 请求总数
    int allNum = 3;
    // 通过数
    int passNum = 0;
    // 被限流数
    int blockNum = 0;
    // 模拟连续请求
    for(int i=0; i<allNum; i++) {
        if( rateLimiter.tryAcquire() ) {
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);

    // 延时以准备下一次测试
    Thread.sleep(8*1000);

    allNum = 22;
    passNum = 0;
    blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++) {
        if( rateLimiter.tryAcquire() ) {
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);
}

测试结果如下，符合预期

As a Queue Version

在As a Meter Version版本的漏桶中，当桶中水未满，请求即会直接被放行。而在漏桶的另外一个版本As a Queue Version中，如果桶中水未满，则该请求将会被暂时存储在桶中。然后以漏桶固定的出水速率对桶中存储的请求依次放行。对比两个版本的漏桶算法不难看出，As a Meter Version版本的漏桶算法可以应对、处理突发流量，只要桶中尚有足够空余即可立即放行请求；而对于As a Queue Version版本的漏桶，其只会以固定速率放行请求，无法充分利用后续系统的处理能力

具体实现如下所示

/**
 * 漏桶算法:  As a Queue Version
 */
public class LeakyBucketLimiter2 {

    /**
     * 定时任务线程池, 用于以指定速率rate从阻塞队列中获取用户请求进行放行、处理
     */
    private ScheduledExecutorService threadPool;

    /**
     * 阻塞队列, 用于存储用户请求
     */
    private ArrayBlockingQueue<UserRequest> queue;

    /**
     * 桶容量, Unit: 个
     */
    private int capacity;

    /**
     * 出水速率, Unit: 个/秒
     */
    private long rate;

    public LeakyBucketLimiter2(int capacity, long rate) {
        this.capacity = capacity;
        this.rate = rate;

        // 根据桶容量构建有界队列
        queue = new ArrayBlockingQueue<>( capacity );

        threadPool = Executors.newSingleThreadScheduledExecutor();
        // 根据出水速率rate计算从阻塞队列获取用户请求的周期， Unit: ms
        long period = 1000 / rate;
        threadPool.scheduleAtFixedRate( getTask() ,0, period, TimeUnit.MILLISECONDS);
    }

    public boolean tryAcquire(UserRequest userRequest) {
        // 添加失败表示用户请求被限流, 则返回false
        return queue.offer(userRequest);
    }

    private Runnable getTask() {
        return () -> {
            // 从阻塞队列获取用户请求
            UserRequest userRequest = queue.poll();
            if( userRequest!=null ) {
                userRequest.handle();
            }
        };
    }

    /**
     * 用户请求
     */
    @AllArgsConstructor
    public static class UserRequest {

        private static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss.SSS");

        private String name;

        public void handle() {
            String timeStr = formatter.format( LocalTime.now() );
            String msg = "<"+timeStr+"> "+name+" 开始处理";
            System.out.println(msg);
        }
    }

}

测试代码如下所示

/**
 * 测试: 漏桶算法(As a Queue Version)
 */
@Test
public void test4() throws Exception {
    // 漏桶配置, 桶容量:5个, 出水率: 2个/秒
    LeakyBucketLimiter2 rateLimiter = new LeakyBucketLimiter2(5, 2);

    // 请求总数
    int allNum = 7;
    // 通过数
    int passNum = 0;
    // 被限流数
    int blockNum = 0;
    // 模拟连续请求
    for(int i=1; i<=allNum; i++) {
        // 构建用户请求
        String name = "用户请求:" + i;
        LeakyBucketLimiter2.UserRequest userRequest = new LeakyBucketLimiter2.UserRequest(name);

        if( rateLimiter.tryAcquire( userRequest ) ) {
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);

    // 延时等待
    Thread.sleep(120*1000);
}

测试结果如下，符合预期

Token Bucket令牌桶

Token Bucket令牌桶的基本原理其实并不复杂，我们以固定速率发放令牌到令牌桶，直到达到桶的容量为止。请求每次会先到令牌桶中获取令牌，如果桶中尚有令牌、获取成功则放行请求；反之，则对请求进行限流。事实上，可以看到Token Bucket令牌桶与As a Meter Version版本的漏桶，其实是一体两面的。前者负责消耗令牌，后者负责注水。本质上是相同的，只不过是思维角度不一样。具体实现如下所示

/**
 * 令牌桶算法
 */
public class TokenBucketLimiter {

    /**
     * 桶容量, Unit: 个
     */
    private long capacity;

    /**
     * 令牌生成速率, Unit: 个/秒
     */
    private long rate;

    /**
     * 桶当前的令牌数量
     */
    private long tokens;

    /**
     * 上次时间
     */
    private long lastTime;

    public TokenBucketLimiter(long capacity, long rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.tokens = capacity;
        this.lastTime = System.currentTimeMillis();
    }

    public synchronized boolean tryAcquire() {
        // 获取当前时间
        long currentTime = System.currentTimeMillis();
        // 计算生成的令牌数量: (当前时间-上次时间) * 令牌生成速率
        long newTokenNum = (currentTime-lastTime)/1000 * rate;
        // 计算令牌数量: 桶当前的令牌数量 + 生成的令牌数量
        tokens = Math.min(capacity, tokens+newTokenNum);
        // 更新时间
        lastTime = currentTime;

        // 桶中仍有令牌, 则请求放行, 返回true
        if( tokens > 0 ) {
            tokens--;
            return true;
        }else{
            // 桶中没有令牌, 则进行限流, 返回false
            return false;
        }
    }

}

测试代码如下所示

/**
 * 测试: 令牌桶算法
 */
@Test
public void test5() throws Exception {
    // 令牌桶配置, 桶容量:5个, 令牌生成速率: 1个/秒
    TokenBucketLimiter rateLimiter = new TokenBucketLimiter(5, 1);

    // 请求总数
    int allNum = 3;
    // 通过数
    int passNum = 0;
    // 被限流数
    int blockNum = 0;
    // 模拟连续请求
    for(int i=0; i<allNum; i++) {
        if( rateLimiter.tryAcquire() ) {
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);

    // 延时以准备下一次测试
    Thread.sleep(8*1000);

    allNum = 22;
    passNum = 0;
    blockNum = 0;
    //模拟连续请求
    for(int i=0; i<allNum; i++) {
        if( rateLimiter.tryAcquire() ) {
            passNum++;
        }else{
            blockNum++;
        }
    }
    System.out.println("请求总数: "+allNum+", 通过数: "+passNum+", 被限流数: "+blockNum);
}

测试结果如下，符合预期

参考文献

1. 凤凰架构 周志明著