经典限流算法设计与实现

public class FixedWindowRateLimiter implements RateLimiter{
    /**
     * 单位窗口时间内,最多可通信多少请求
     */
    private final int threshold = 3;
    /**
     * 窗口大小
     */
    private final long windowUnit = 1;
    /**
     * 当前窗口的初始创建时间
     */
    private long windowCreateTime;
    /**
     * 当前窗口已经通过的请求数量
     */
    private int counter;

    @Override
    public boolean tryAcquire() {
        // 1. 获取系统当前时间
        long currentTime = System.currentTimeMillis();
        // 2. 检查当前时间窗口是否过期
        if(currentTime - windowCreateTime > windowUnit){
            // 3. 重新开启一个时间窗口
            windowCreateTime = currentTime;
            counter = 0;
        }
        // 4. 判断当前窗口内放行的请求数量是否已经超过阈值
        if(counter > threshold){
            return false;
        }
        // 5. 放行请求
        counter++;
        return true;
    }
}

public class SlideWindowRateLimiter implements RateLimiter {
    /**
     * 循环队列，就是装多个窗口用，该数量是windowSize的2倍
     */
    private AtomicLong[] timeSlices;
    /**
     * 队列的总长度
     */
    private int timeSliceSize;
    /**
     * 每个时间片的时长，以毫秒为单位
     */
    private int timeMillisPerSlice;
    /**
     * 共有多少个时间片（即窗口长度）
     */
    private int windowSize;
    /**
     * 在一个完整窗口期内允许通过的最大阈值
     */
    private int threshold;
    /**
     * 该滑窗的起始创建时间，也就是第一个数据 (循环队列当前头元素对应时间戳)
     */
    private long beginTimestamp;
    /**
     * 最后一个数据的时间戳
     */
    private long lastAddTimestamp;

    public SlideWindowRateLimiter(int duration, int threshold) {
        //超过10分钟的按10分钟
        if (duration > 600) {
            duration = 600;
        }
        //要求5秒内探测出来的，
        if (duration <= 5) {
            this.windowSize = 5;
            this.timeMillisPerSlice = duration * 200;
        } else {
            this.windowSize = 10;
            this.timeMillisPerSlice = duration * 100;
        }
        this.threshold = threshold;
        // 保证存储在至少两个window
        this.timeSliceSize = windowSize * 2;

        reset();
    }

    @Override
    public boolean tryAcquire() {
        // 1. 当前自己所在的位置,是哪个小时间窗
        int index = locationIndex();
        // 2. 清空自己前面windowSize到2*windowSize之间的数据格的数据
        // 譬如1秒分4个窗口，那么数组共计8个窗口
        // 当前index为5时，就清空6、7、8、1。然后把2、3、4、5的加起来就是该窗口内的总和
        clearFromIndex(index);

        // 3. 在当前时间片里继续+1
        int sum = 0;
        sum += timeSlices[index].addAndGet(1);
        // 4. 加上前面几个时间片
        for (int i = 1; i < windowSize; i++) {
            sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();
        }

        lastAddTimestamp = System.currentTimeMillis();

        // 5. 判断是否超出阈值
        return sum >= threshold;
    }

    /**
     * 初始化
     */
    private void reset() {
        beginTimestamp = System.currentTimeMillis();
        AtomicLong[] localTimeSlices = new AtomicLong[timeSliceSize];
        for (int i = 0; i < timeSliceSize; i++) {
            localTimeSlices[i] = new AtomicLong(0);
        }
        timeSlices = localTimeSlices;
    }

    /**
     * 计算当前所在的时间片的位置
     */
    private int locationIndex() {
        long now = System.currentTimeMillis();
        // 1. 如果当前的key已经超出一整个时间片了,那么就直接初始化就行了,不用去计算了
        if (now - lastAddTimestamp > (long) timeMillisPerSlice * windowSize) {
            reset();
        }
        // 2. 计算当前时间所在队列中的下标 (beginTimestamp表示当前循环队列头元素对应时间戳)
        int index = (int) (((now - beginTimestamp) / timeMillisPerSlice) % timeSliceSize);
        return Math.max(index, 0);
    }

    private void clearFromIndex(int index) {
        for (int i = 1; i <= windowSize; i++) {
            int j = index + i;
            if (j >= windowSize * 2) {
                j -= windowSize * 2;
            }
            timeSlices[j].set(0);
        }
    }
}

public class LeakyBucketRateLimiter implements RateLimiter {
    /**
     * 每秒处理数（出水率）
     */
    private long rate;
    /**
     * 当前剩余水量
     */
    private long currentWater;
    /**
     * 最后刷新时间
     */
    private long refreshTime;
    /**
     * 桶容量
     */
    private long capacity;

    @Override
    public boolean tryAcquire() {
        // 1. 获取系统当前时间
        long currentTime = System.currentTimeMillis();
        // 2. 流出的水量 = (当前时间-上次刷新时间)* 出水率
        long outWater = (currentTime - refreshTime) / 1000 * rate;
        // 3. 当前水量 = 之前的桶内水量-流出的水量
        currentWater = Math.max(0, currentWater - outWater);
        // 4. 刷新时间
        refreshTime = currentTime; 

        // 5. 当前剩余水量还是小于桶的容量,则请求放行
        if (currentWater < capacity) {
            currentWater++;
            return true;
        }

        // 6. 当前剩余水量大于等于桶的容量,限流
        return false;
    }
}

public class TokenBucketRateLimiter implements RateLimiter {
    /**
     * 每秒处理数（放入令牌数量）
     */
    private long putTokenRate;
    /**
     * 最后刷新时间
     */
    private long refreshTime;
    /**
     * 令牌桶容量
     */
    private long capacity;
    /**
     * 当前桶内令牌数
     */
    private long currentToken = 0L;

    @Override
    public boolean tryAcquire() {
        // 1. 获取系统当前时间
        long currentTime = System.currentTimeMillis();
        // 2. 生成的令牌 =(当前时间-上次刷新时间)* 放入令牌的速率
        long generateToken = (currentTime - refreshTime) / 1000 * putTokenRate;
        // 3. 当前令牌数量 = 之前的桶内令牌数量+放入的令牌数量
        currentToken = Math.min(capacity, generateToken + currentToken);
        // 4. 刷新时间
        refreshTime = currentTime;

        // 5. 桶里面还有令牌，请求正常处理
        if (currentToken > 0) {
            currentToken--;
            return true;
        }

        return false;
    }
}