Sentinel 限流算法

但是这种实现会有一个问题，举个例子:

假设我们设定 1 秒内允许通过的请求阈值是 99，如果有用户在时间窗口的最后几毫秒发送了 99 个请求，紧接着又在下一个时间窗口开始时发送了 200 个请求，那么这个用户其实在一秒内成功请求了 198 次，显然超过了阈值但并不会被限流。其实这就是临界值问题，那么临界值问题要怎么解决呢？

2、计数器滑动窗口算法

算法演示

计数器滑动窗口法就是为了解决上述固定窗口计数存在的问题而诞生。前面说了固定窗口存在临界值问题，要解决这种临界值问题，显然只用一个窗口是解决不了问题的。假设我们仍然设定 1 秒内允许通过的请求是 200 个，但是在这里我们需要把 1 秒的时间分成多格，假设分成 5 格（格数越多，流量过渡越平滑），每格窗口的时间大小是 200 毫秒，每过 200 毫秒，就将窗口向前移动一格。为了便于理解，可以看下图

图中将窗口划为 5 份，每个小窗口中的数字表示在这个窗口中请求数，所以通过观察上图，可知在当前时间快（200 毫秒）允许通过的请求数应该是 70 而不是 200（只要超过 70 就会被限流），因为我们最终统计请求数时是需要把当前窗口的值进行累加，进而得到当前请求数来判断是不是需要进行限流。

那么滑动窗口限流法是完美的吗？

细心观察的我们应该能马上发现问题，滑动窗口限流法其实就是计数器固定窗口算法的一个变种。流量的过渡是否平滑依赖于我们设置的窗口格数也就是统计时间间隔，格数越多，统计越精确，但是具体要分多少格我们也说不上来呀...

package com.on.sentinel;

import com.on.sentinel.exception.BlockException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class SlidingWindowRateLimit implements RateLimit, Runnable {

    /**
     * 阈值
     */
    private Integer limitCount;

    /**
     * 当前通过的请求数
     */
    private AtomicInteger passCount;

    /**
     * 窗口数
     */
    private Integer windowSize;

    /**
     * 每个窗口时间间隔大小
     */
    private long windowPeriod;
    private TimeUnit timeUnit;

    private Window[] windows;
    private volatile Integer windowIndex = 0;

    private Lock lock = new ReentrantLock();

    public SlidingWindowRateLimit(Integer limitCount) {
        // 默认统计 qps, 窗口大小 5
        this(limitCount, 5, 200, TimeUnit.MILLISECONDS);
    }

    /**
     * 统计总时间 = windowSize * windowPeriod
     */
    public SlidingWindowRateLimit(Integer limitCount, Integer windowSize, Integer windowPeriod, TimeUnit timeUnit) {
        this.limitCount = limitCount;
        this.windowSize = windowSize;
        this.windowPeriod = windowPeriod;
        this.timeUnit = timeUnit;
        this.passCount = new AtomicInteger(0);
        this.initWindows(windowSize);
        this.startResetTask();
    }

    @Override
    public boolean canPass() throws BlockException {
        lock.lock();
        if (passCount.get() > limitCount) {
            throw new BlockException();
        }
        windows[windowIndex].passCount.incrementAndGet();
        passCount.incrementAndGet();
        lock.unlock();
        return true;
    }

    private void initWindows(Integer windowSize) {
        windows = new Window[windowSize];
        for (int i = 0; i < windowSize; i++) {
            windows[i] = new Window();
        }
    }

    private ScheduledExecutorService scheduledExecutorService;
    private void startResetTask() {
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(this, windowPeriod, windowPeriod, timeUnit);
    }

    @Override
    public void run() {
        // 获取当前窗口索引
        Integer curIndex = (windowIndex + 1) % windowSize;
        // 重置当前窗口索引通过数量，并获取上一次通过数量
        Integer count = windows[curIndex].passCount.getAndSet(0);
        windowIndex = curIndex;
        // 总通过数量 减去 当前窗口上次通过数量
        passCount.addAndGet(-count);
    }

    @Data
    class Window {
        private AtomicInteger passCount;
        public Window() {
            this.passCount = new AtomicInteger(0);
        }
    }
}

3、漏桶算法

上面所介绍的两种算法都不能非常平滑的过渡，下面就是漏桶算法登场了

什么是漏桶算法？

漏桶算法以一个常量限制了出口流量速率，因此漏桶算法可以平滑突发的流量。其中漏桶作为流量容器我们可以看做一个 FIFO 的队列，当入口流量速率大于出口流量速率时，因为流量容器是有限的，当超出流量容器大小时，超出的流量会被丢弃。

下图比较形象的说明了漏桶算法的原理，其中水龙头是入口流量，漏桶是流量容器，匀速流出的水是出口流量。

漏桶算法的特点

• 漏桶具有固定容量，出口流量速率是固定常量（流出请求）
• 入口流量可以以任意速率流入到漏桶中（流入请求）
• 如果入口流量超出了桶的容量，则流入流量会溢出（新请求被拒绝）

package com.on.sentinel;

import com.on.sentinel.exception.BlockException;
import lombok.extern.slf4j.Slf4j;

import javax.swing.*;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

@Slf4j
public class LeakyBucketRateLimit implements RateLimit, Runnable {

    /**
     * 出口限制 qps
     */
    private Integer limitSecond;
    /**
     * 漏桶队列
     */
    private BlockingQueue<Thread> leakyBucket;

    private ScheduledExecutorService scheduledExecutorService;

    public LeakyBucketRateLimit(Integer bucketSize, Integer limitSecond) {
        this.limitSecond = limitSecond;
        this.leakyBucket = new LinkedBlockingDeque<>(bucketSize);

        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        //1 秒=1000000000 纳秒
        // 计算出一个请求需要多少纳秒
        long interval = (1000 * 1000 * 1000) / limitSecond;
        scheduledExecutorService.scheduleAtFixedRate(this, 0, interval, TimeUnit.NANOSECONDS);
    }

    @Override
    public boolean canPass() throws BlockException {
        if (leakyBucket.remainingCapacity() == 0) {
            throw new BlockException();
        }
        //如果立即可行且不违反容量限制，则将指定的元素插入此双端队列表示的队列中（即此双端队列的尾部），
        // 并在成功时返回 true；如果当前没有空间可用，则返回 false。
        leakyBucket.offer(Thread.currentThread());
        //所有请求进来，只要有容量，就会停止到这里
        // 如果没有容量，就停止在上行代码
        LockSupport.park();
        return true;
    }

    @Override
    public void run() {
        // 以固定的速率去唤醒他，这样执行的速度就固定值
        Thread thread = leakyBucket.poll();
        if (Objects.nonNull(thread)) {
            LockSupport.unpark(thread);
        }
    }
}

不过因为漏桶算法限制了流出速率是一个固定常量值，所以漏桶算法不支持出现突发流出流量。但是在实际情况下，流量往往是突发的。

4、令牌桶算法

令牌桶算法是漏桶算法的改进版，可以支持突发流量。不过与漏桶算法不同的是，令牌桶算法的漏桶中存放的是令牌而不是流量。

那么令牌桶算法是怎么突发流量的呢？

最开始，令牌桶是空的，我们以恒定速率往令牌桶里加入令牌，令牌桶被装满时，多余的令牌会被丢弃。当请求到来时，会先尝试从令牌桶获取令牌（相当于从令牌桶移除一个令牌），获取成功则请求被放行，获取失败则阻塞活拒绝请求。

令牌桶算法的特点

• 最多可以存发 b 个令牌。如果令牌到达时令牌桶已经满了，那么这个令牌会被丢弃
• 每当一个请求过来时，就会尝试从桶里移除一个令牌，如果没有令牌的话，请求无法通过。

令牌桶算法限制的是平均流量，因此其允许突发流量（只要令牌桶中有令牌，就不会被限流）

package com.on.sentinel;
import com.on.sentinel.exception.BlockException;
import org.apache.commons.lang3.StringUtils;
import java.util.concurrent.*;

public class TokenBucketRateLimit implements RateLimit, Runnable {
    /** * token 生成 速率 （每秒） */
    private Integer tokenLimitSecond;
    /** * 令牌桶队列 */
    private BlockingQueue<String /* token */> tokenBucket;
    private static final String TOKEN = "__token__";
    private ScheduledExecutorService scheduledExecutorService;

    public TokenBucketRateLimit(Integer bucketSize, Integer tokenLimitSecond) {
        this.tokenLimitSecond = tokenLimitSecond;
        this.tokenBucket = new LinkedBlockingDeque<>(bucketSize);
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        long interval = (1000 * 1000 * 1000) / tokenLimitSecond;
        scheduledExecutorService.scheduleAtFixedRate(this, 0, interval, TimeUnit.NANOSECONDS);
    }

    @Override
    public boolean canPass() throws BlockException {
        String token = tokenBucket.poll();
        if (StringUtils.isEmpty(token)) {
            throw new BlockException();
        }
        return true;
    }

    @Override
    public void run() {
        if (tokenBucket.remainingCapacity() == 0) {
            return;
        }
        tokenBucket.offer(TOKEN);
    }
}