常用的限流方案思路和实现

限流方案

1、计数器（固定窗口）

1.1、简介

计数器固定窗口算法是最基础也是最简单的一种限流算法。原理就是对一段固定时间窗口内的请求进行计数，如果请求数超过了阈值，则舍弃该请求；如果没有达到设定的阈值，则接受该请求，且计数加1。当时间窗口结束时，重置计数器为0。
在这里插入图片描述

1.2、代码实现

import java.util.concurrent.atomic.AtomicInteger;

public class CountLimiter {
    /**
     * 窗口大小，单位为毫秒
     */
    private int window;
    /**
     * 窗口内允许的请求次数
     */
    private int limit;
    /**
     * 当前窗口内的请求次数
     */
    private AtomicInteger count;
    /**
     * 窗口是否正在运行
     */
    private volatile boolean isRunning = true;

    public CountLimiter() {
        this(60, 10);
    }

    /**
     * 构造函数
     * @param window 窗口大小，单位为毫秒
     * @param limit 窗口内允许的请求次数
     */
    public CountLimiter(int window, int limit) {
        this.window = window;
        this.limit = limit;
        count = new AtomicInteger(0);
    }

    /**
     * 尝试获取令牌
     * @return 获取成功返回true，失败返回false
     */
    public boolean tryAcquire() {
        int current = count.incrementAndGet();
        if (current > limit) {
            return false;
        } else {
            return true;
        }
    }

    /**
     * 启动窗口线程
     */
    public void start() {
        // 启动一个线程，每过window毫秒将count置为0
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    try {
                        Thread.sleep(window);
                        count.set(0);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }

    /**
     * 关闭窗口线程
     */
    public void stop() {
        isRunning = false;
        System.out.println("限流关闭，关闭时间" + System.currentTimeMillis());
    }

    public static void main(String[] args) {
        CountLimiter limiter = new CountLimiter(100, 3);
        limiter.start();
        int count = 0;

        for (int i = 0; i < 10; ++i) {
            if (limiter.tryAcquire()) {
                count++;
                System.out.println("请求通过, 当前时间" + System.currentTimeMillis());
            }
        }
        limiter.stop();
        System.out.println("请求通过：" + count);
    }
}

运行结果：

请求通过, 当前时间1709273340687
请求通过, 当前时间1709273340693
请求通过, 当前时间1709273340693
限流关闭，关闭时间1709273340693
请求通过：3

1.3、特点分析

实现简单，容易理解
窗口界限处可能会出现两倍于阈值的流量
窗口时间内可能会出现服务不可用，流量不够平滑

2、计数器（滑动窗口）

2.1、简介

计数器滑动窗口算法是计数器固定窗口算法的改进，解决了固定窗口切换时可能会产生两倍于阈值流量请求的缺点。
滑动窗口算法在固定窗口的基础上，将一个计时窗口分成了若干个小窗口，然后每个小窗口维护一个独立的计数器。当请求的时间大于当前窗口的最大时间时，则将计时窗口向前平移一个小窗口。平移时，将第一个小窗口的数据丢弃，然后将第二个小窗口设置为第一个小窗口，同时在最后面新增一个小窗口，将新的请求放在新增的小窗口中。同时要保证整个窗口中所有小窗口的请求数目之和不能超过设定的阈值。
在这里插入图片描述

滑动窗口算法其实就是对请求数进行了更细粒度的限流，窗口划分得越多，则限流越精准。

2.2、代码实现

public class CountSlideLimiter {
    // 窗口大小
    private int window;
    // 窗口内允许的请求次数
    private int limit;
    // 滑动窗口的个数
    private int splitNum;
    // 计数器
    private int[] counter;
    // 计数器索引
    private int index;
    // 开始时间
    private long startTime;

    public CountSlideLimiter() {
        this(1000, 20, 10);
    }

    public CountSlideLimiter(int window, int limit, int splitNum) {
        this.window = window;
        this.limit = limit;
        this.splitNum = splitNum;
        counter = new int[splitNum];
        index = 0;
        startTime = System.currentTimeMillis();
    }

    /**
     * 尝试获取令牌
     * @return 获取成功返回true，否则返回false
     */
    public synchronized boolean tryAcquire() {
        long curTime = System.currentTimeMillis();
        long windowNum = Math.max(curTime - window - startTime, 0) / (window / splitNum);
        slide(windowNum);
        int count = 0;
        for (int i = 0; i < splitNum; ++i) {
            count += counter[i];
        }
        if (count >= limit) {
            return false;
        } else {
            counter[index]++;
            return true;
        }
    }

    /**
     * 滑动窗口
     * @param windowNum 滑动窗口距离
     */
    private synchronized void slide(long windowNum) {
        if (windowNum == 0) {
            return;
        }
        long slideNum = Math.min(windowNum, splitNum);
        for (int i = 0; i < slideNum; ++i) {
            index = (index + 1) % splitNum;
            counter[index] = 0;
        }
        startTime = startTime + windowNum * (window / splitNum);
    }

    public static void main(String[] args) throws InterruptedException {
        int limit = 3;
        CountSlideLimiter limiter = new CountSlideLimiter(100, limit, 5);
        int count = 0;

        for (int i = 0; i < 10; ++i) {
            count = 0;
            for (int j = 0; j < 50; ++j) {
                if (limiter.tryAcquire()) {
                    count++;
                    System.out.println("请求通过, 当前时间" + System.currentTimeMillis());
                }
            }
            Thread.sleep(10);
            for (int j = 0; j < 50; ++j) {
                if (limiter.tryAcquire()) {
                    count++;
                    System.out.println("请求通过, 当前时间" + System.currentTimeMillis());
                }
            }
            if (count > limit) {
                System.out.println("限流失败，当前请求数：" + count);
            }
        }
    }
}

运行结果：

请求通过, 当前时间1709278379478
请求通过, 当前时间1709278379484
请求通过, 当前时间1709278379484

2.2、特点分析：

避免了计数器固定窗口算法在固定窗口切换时可能产生两倍于阈值流量的请求问题；
和漏斗算法相比，新来的请求也能够被处理到，避免了漏斗算法的饥饿问题。

3、漏桶算法

3.1、简介

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

3.2、代码实现

import java.util.LinkedList;
import java.util.List;

public class LeakyBucketLimiter {
    // 容量
    private int capacity;
    // 速率
    private int rate;
    // 剩余容量
    private int left;
    // 请求队列
    private List<Request> requests;
    // 是否运行
    private volatile boolean isRunning = true;

    public LeakyBucketLimiter(int capacity, int rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.left = capacity;
        this.requests = new LinkedList<>();
    }

    public void handelRequest(Request request) {
        request.setHandleTime(System.currentTimeMillis());
        left++;
        if (left > capacity) {
            left = capacity;
        }
        System.out.println(request.toString());
    }

    public synchronized boolean tryAcquire(Request request) {
        if (left <= 0) {
            return false;
        } else {
            left--;
            requests.add(request);
            return true;
        }
    }

    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(isRunning) {
                    if(!requests.isEmpty()) {
                        Request request = requests.remove(0);
                        handelRequest(request);
                    }
                    try {
                        Thread.sleep(1000 / rate);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();;
    }

    public void stop() {
        isRunning = false;
    }

    public static void main(String[] args) throws InterruptedException {
        LeakyBucketLimiter limiter = new LeakyBucketLimiter(3, 5);
        limiter.start();
        int count = 0;
        for(int i = 0; i < 10; i++) {
            Thread.sleep(50);
            Request request = new Request(i, System.currentTimeMillis(), 0L);
            if(limiter.tryAcquire(request)) {
                count++;
                System.out.println("请求被接受");
            } else {
                System.out.println("请求被拒绝");
            }
        }
        System.out.println("请求通过数：" + count);
        Thread.sleep(10000);
        limiter.stop();
    }

    public static class Request {
        private int code;
        private long lanchTime;
        private long handleTime;

        public Request() {
        }

        public Request(int code, long lanchTime, long handleTime) {
            this.code = code;
            this.lanchTime = lanchTime;
            this.handleTime = handleTime;
        }

        public int getCode() {
            return code;
        }
        public void setCode(int code) {
            this.code = code;
        }

        public long getLanchTime() {
            return lanchTime;
        }
        public void setLanchTime(long lanchTime) {
            this.lanchTime = lanchTime;
        }
        public long getHandleTime() {
            return handleTime;
        }
        public void setHandleTime(long handleTime) {
            this.handleTime = handleTime;
        }
        @Override
        public String toString() {
            return "Request{" +
                    "code=" + code +
                    ", lanchTime=" + lanchTime +
                    ", handleTime=" + handleTime +
                    '}';
        }
    }
}

运行结果：

请求被接受
请求被接受
请求被接受
请求被接受
Request{code=0, lanchTime=1709280004490, handleTime=1709280004640}
请求被拒绝
请求被拒绝
请求被拒绝
Request{code=1, lanchTime=1709280004546, handleTime=1709280004861}
请求被接受
请求被拒绝
请求被拒绝
请求通过数：5
Request{code=2, lanchTime=1709280004596, handleTime=1709280005066}
Request{code=3, lanchTime=1709280004651, handleTime=1709280005267}
Request{code=7, lanchTime=1709280004867, handleTime=1709280005471}

3.3、特点分析：

漏桶的漏出速率是固定的，对下游系统起到了保护的作用
不能解决流量突发情况。

4、令牌桶

4.1、简介

令牌桶算法同样是实现限流是一种常见的思路，最为常用的 Google 的 Java 开发工具包 Guava 中的限流工具类 RateLimiter 就是令牌桶的一个实现。令牌桶的实现思路类似于生产者和消费之间的关系。
系统服务作为生产者，按照指定频率向桶（容器）中添加令牌，如 QPS 为 2，每 500ms 向桶中添加一个令牌，如果桶中令牌数量达到阈值，则不再添加。
请求执行作为消费者，每个请求都需要去桶中拿取一个令牌，取到令牌则继续执行；如果桶中无令牌可取，就触发拒绝策略，可以是超时等待，也可以是直接拒绝本次请求，由此达到限流目的。
在这里插入图片描述

4.2、代码实现：

public class TokenBucketLimiter {
    private int capacity;
    private int rate;
    private int tokens;
    private volatile boolean isRunning = true;
    private static final int DEFAULT_CAPACITY = 10;
    private static final int DEFAULT_RATE = 5;

    public TokenBucketLimiter() {
        this(DEFAULT_CAPACITY, DEFAULT_RATE);
    }

    public TokenBucketLimiter(int capacity, int rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.tokens = capacity;
    }

    public synchronized boolean tryAcquire(Request request) {
        if (tokens > 0) {
            tokens--;
            handelRequest(request);
            return true;
        } else {
            return false;
        }
    }

    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    synchronized(this) {
                        tokens++;
                        if (tokens > capacity) {
                            tokens = capacity;
                        }
                    }
                    try {
                        Thread.sleep(1000 / rate);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();;
    }

    public void stop() {
        isRunning = false;
    }

    public void handelRequest(Request request) {
        request.setHandleTime(System.currentTimeMillis());
        System.out.println(request.toString());
    }

    public static void main(String[] args) throws InterruptedException {
        TokenBucketLimiter limiter = new TokenBucketLimiter(3, 5);
        limiter.start();
        int count = 0;
        for(int i = 0; i < 10; i++) {
            Thread.sleep(50);
            Request request = new Request(i, System.currentTimeMillis(), 0L);
            if(limiter.tryAcquire(request)) {
                count++;
                System.out.println("请求被接受，当前时间：" + System.currentTimeMillis());
            } else {
                System.out.println("请求被拒绝，当前时间：" + System.currentTimeMillis());
            }
        }
        System.out.println("请求通过数：" + count);
        limiter.stop();
    }

    public static class Request {
        private int code;
        private long lanchTime;
        private long handleTime;

        public Request() {
        }

        public Request(int code, long lanchTime, long handleTime) {
            this.code = code;
            this.lanchTime = lanchTime;
            this.handleTime = handleTime;
        }

        public int getCode() {
            return code;
        }
        public void setCode(int code) {
            this.code = code;
        }

        public long getLanchTime() {
            return lanchTime;
        }
        public void setLanchTime(long lanchTime) {
            this.lanchTime = lanchTime;
        }
        public long getHandleTime() {
            return handleTime;
        }
        public void setHandleTime(long handleTime) {
            this.handleTime = handleTime;
        }
        @Override
        public String toString() {
            return "Request{" +
                    "code=" + code +
                    ", lanchTime=" + lanchTime +
                    ", handleTime=" + handleTime +
                    '}';
        }
    }
}

运行结果：

Request{code=0, lanchTime=1709279960998, handleTime=1709279960998}
请求被接受，当前时间：1709279961008
Request{code=1, lanchTime=1709279961061, handleTime=1709279961061}
请求被接受，当前时间：1709279961061
Request{code=2, lanchTime=1709279961111, handleTime=1709279961111}
请求被接受，当前时间：1709279961111
Request{code=3, lanchTime=1709279961162, handleTime=1709279961162}
请求被接受，当前时间：1709279961162
请求被拒绝，当前时间：1709279961214
请求被拒绝，当前时间：1709279961268
请求被拒绝，当前时间：1709279961323
Request{code=7, lanchTime=1709279961378, handleTime=1709279961378}
请求被接受，当前时间：1709279961378
请求被拒绝，当前时间：1709279961433
请求被拒绝，当前时间：1709279961485
请求通过数：5