Redis-限流方案

在实际业务中，可能会遇到瞬时流量剧增的情况，大量的请求可能会导致服务器过载和宕机。为了保护系统自身和上下游服务，需要采用限流的方式，拒绝部分请求。
限流就是对请求的频率进行控制，迅速拒绝超过请求阈值的请求。
比如：Redis的QPS为10w，但是此时20w个请求同一时刻请求到Redis，造成Redis性能降低。

1. 什么是限流

限流就是在高并发场景下，通过限制系统处理请求的速率，迅速拒绝超过设定上限的请求(最大的QPS)，从而保证系统正常运行。
在限流技术中，主要是有两个注意点：当前系统请求的阈值和拒绝策略。
请求的阈值：阈值就是单位时间内允许请求的最大请求数。例如，将QPS(Queries Per Second)设置为1000，表明1s内最多接受1000个请求。通过设置适当的阈值，可以有效控制系统负载，避免系统请求过多而崩溃。（阈值是上线前通过压测或者评估得出的指标）

拒绝策略： 处理超过阈值的请求的方法，常见的拒绝策略包括直接拒绝和等待。

• 直接拒绝策略：直接拒绝超过阈值的请求，直接返回给客户端。（提示：当前抢购人数太多）
• 排队等待策略：将请求放入队列中，按照一定规则处理，避免瞬间拒绝大量请求。

2. 常见的限流算法

常见的限流算法分为：固定窗口(计数器限流)、滑动窗口、漏桶、令牌桶 四个方案

2.1. 固定窗口限流算法

在一段时间间隔内，处理请求的最大数量。

固定窗口其实就是时间窗口，原理是将时间划分成固定大小的窗口，并且在每个窗口内限制请求的数量和速率，如果某个窗口内的请求超过了阈值，就直接执行拒绝策略。

即：固定窗口限流算法是规定了单位时间处理的请求数量。

算法步骤：

1. 将时间划分为固定大小的窗口，例如每秒一个。
2. 在每一个时间窗口内，记录请求的数量。
3. 当新的请求到达时，当前窗口的请求计数器的值加一。
4. 如果窗口内请求计数器超过阈值，那么执行拒绝策略。
5. 当窗口结束后，重置计数器。
   

例如，每个时间窗口为1s，限流的阈值为3，窗口内超过阈值的请求都会触发拒绝策略。
优点：实现简单，易于理解。
缺点：

• 请求不够均匀：在固定窗口算法中，请求在窗口内的分布可能会不均匀。假如限制某个接口每分钟只能访问30次，那么前30秒就会有30个请求到达的话，那么后30秒就无法处理请求，导致请求不均匀。
• 无法保证限流速率，因而无法因对突增的流量。比如一个接口一分钟只能访问1000次，但是接口的QPS为500，前55秒这个接口没有收到请求，但是后一秒突然接收到1000个请求。然后，在当前场景下，这1000个请求在1s内是无法被处理的，系统可能就直接被大量的请求给击垮了。
• 存在明显的临界问题（请求突刺问题）：前一个窗口和后一个窗口之间的可能会存在大量的请求，无法应对无法流量。
  比如：一个窗口内请求的阈值为3，以下存在临界问题：
  

public class FixedWindowLimiter {
    private static final String FORMAT_TIME = "yyyy-MM-dd HH:mm:ss";

    private final long windowSize; // 窗口大小，默认是毫秒
    private final int maxRequests; // 最大请求数
    private int currentRequests; //当前窗口内的请求数
    private LocalDateTime lastRestTime; //上次窗口重置时间，即窗口开始时间
    private final Lock resetMutex; //重置锁

    public FixedWindowLimiter(Duration windowSize, int maxRequests) {
        this.windowSize = windowSize.toMillis();
        this.maxRequests = maxRequests;
        this.currentRequests = 0;
        this.lastRestTime = LocalDateTime.now();
        this.resetMutex = new ReentrantLock();
    }

    public boolean allow() {
        resetMutex.lock();
        try {
            LocalDateTime now = LocalDateTime.now();
            if (Duration.between(lastRestTime, now).toMillis() >= windowSize) {
                this.currentRequests = 0;
                this.lastRestTime = now;
            }
            if (currentRequests >= maxRequests) {
                return false;
            }
            currentRequests++;
            return true;
        }finally {
            resetMutex.unlock();
        }
    }
}


class FixedWindowLimiterApp {
    private static final String FORMAT_TIME = "yyyy-MM-dd HH:mm:ss";
    public static void main(String[] args) {
        System.out.println(System.currentTimeMillis() / 1000);
        FixedWindowLimiter limiter = new FixedWindowLimiter(Duration.ofSeconds(1), 3);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern(FORMAT_TIME);
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);

        for (int i = 0; i < 20; i++) {
            Runnable task = () -> {
                String now = LocalDateTime.now().format(formatter);
                if (limiter.allow()) {
                    System.out.println(now + "请求通过");
                } else {
                    System.out.println(now + "请求被限流");
                }
            };
            pool.schedule(task, i * 100, TimeUnit.MILLISECONDS);
        }

        pool.shutdown();
        try {
            if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
                pool.shutdownNow();
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
        }

    }
}

2.2. 滑动窗口限流算法

滑动窗口限流算法是固定窗口限流算法的升级版本，限流的粒度更小。
滑动窗口和固定窗口相比：滑动窗口是将时间按照一定的比例分片。
假如接口限流每分钟处理60个请求，可以将1分钟分成60个小窗口，然后每隔一秒移动一次，每个窗口一秒只能处理不大于60(请求数) / 60(窗口数)，如果当前窗口的请求计数综合超过了限制的数量的话，就触发拒绝策略。

这样的话，可以处理[0.5秒 ~ 1.5秒]内的临界问题，红色部分的请求将会触发拒绝策略。
优点：

• 相比于固定窗口算法，滑动窗口计数器算法可以应对突发流量，解决一些临界问题。
• 限流粒度更小，可以提供更精确的限流控制。
  缺点：
• 和固定窗口一样，存在请求不够均匀的情况。

2.3. 漏桶限流算法

漏桶限流的核心思想就是将请求存储在一个漏桶中，无论我们以任意速率存入请求，漏桶始终以固定的速率流出。所以漏桶算法可以将突发流量均匀地处理，确保系统在稳定的负载下运行。

优点：

1. 实现简单，易于理解。
2. 可以控制限流速率，避免网络拥塞和系统过载。
   缺点：
3. 在突发情况请求过多时，会丢弃过多的请求。

2.4. 令牌桶限流算法

令牌桶算法也比较简单，和漏桶算法一样，主角还是桶。不过现在桶里面装的是令牌，请求在被处理之前需要拿到一个令牌，请求处理完毕之后直接丢弃令牌即可。我们可以根据限流大小，按照一定的速率往桶里添加令牌，如果桶装满了，就不能继续往桶里添加令牌了。

优点：

1. 可以应对突发流量：由于令牌可以在桶中堆积，当流量突然增大时，如果桶中有足够的令牌，系统可以快速响应这种突发流量，避免请求被立即拒绝。这使得令牌桶算法特别适合处理具有突发性流量的应用场景。
2. 灵活性强：通过适当调整流派的生成速率和桶的容量，可以灵活地控制流量。
   算法设计:

public class TokenBucketLimiter {
    private final int rate; // 令牌产生的速度，即每秒生成多少个令牌
    private final int capacity; // 令牌桶容量，最多可存储令牌数
    private int currentTokenNum; // 当前令牌数
    private LocalDateTime lastTime; // 上次请求时间
    private final Lock lock; // 请求锁

    public TokenBucketLimiter(int rate, int capacity) {
        this.rate = rate;
        this.capacity = capacity;
        this.currentTokenNum = 0;
        this.lastTime = LocalDateTime.now();
        this.lock = new ReentrantLock();
    }

    public int getCurrentTokenNum() {
        return currentTokenNum;
    }

    public int getRate() {
        return rate;
    }

    public int getCapacity() {
        return capacity;
    }

    public boolean allow() {
        lock.lock();
        try {
            // 计算当前时间和上一次更新时间的时间间隔
            long timeDiff = Duration.between(lastTime, LocalDateTime.now()).toMillis();
            // 计算时间间隔内生成的令牌数量
            int tokenCount = (int) (timeDiff / 1000.0 * rate);
            if (tokenCount > 0) {
                currentTokenNum += tokenCount;
                lastTime = LocalDateTime.now();
            }
            // 令牌的数量不能超过桶的大小
            if (currentTokenNum > capacity) {
                currentTokenNum = capacity;
            }

            // 获取令牌
            if (currentTokenNum > 0) {
                currentTokenNum--;
                return true;
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

}

class TokenBucketLimiterApp {
    public static void mockRequest(int n, Duration d, TokenBucketLimiter limiter) {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

        for (int i = 0; i < n; i++) {
            int requestId = i + 1;
            executor.schedule(() -> {
                if (limiter.allow()) {
                    System.out.printf("第%d个请求通过\n", requestId);
                } else {
                    System.out.printf("第%d个请求被限流\n", requestId);
                }
            }, i * d.toMillis(), TimeUnit.MILLISECONDS);
        }

        executor.shutdown();
        try {
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }

    public static void main(String[] args) {
        System.out.println("=================令牌桶算法=================");
        // 创建一个令牌桶，令牌的生成速度为每秒4个，桶容量为5个请求
        TokenBucketLimiter limiter = new TokenBucketLimiter(4, 5);
        // 发送10个请求，每50ms发送一个
        mockRequest(10, Duration.ofMillis(50), limiter);
        System.out.println("------------------------------------------");
    }
}

3. 针对什么进行限流？

实际项目中，针对的先流对象是什么？也就是说针对什么来进行限流？常见的限流对象如下：

1. 针对IP限流：比较常见，比如通过X-forwarded-For或TCP options字段中真实源IP信息。
2. 业务ID：挑选唯一的业务ID实现限流。比如，根据用户的ID进行限流。（写leetcode的时候多次提交，提醒提交太快）。
3. 个性化：根据用户的属性或者行为。进行不同的限流策略。例如，VIP用户不限流，普通用户限流。

4. 单机限流

单机限流是指在单台服务器上，通过限制其在单位时间内处理的请求数量来防止过载。
单机限流可以使用Google Guava自带的限流工具类 RateLimiter。RateLimiter是基于令牌桶算法实现的，可以应对突发流量。

5. 分布式限流

单机限流只能保证一个节点限流，但是无法保证分布式限流。
分布式限流和分布式锁的思想是一样的，就是将原本设计实现在本地服务的限流操作抽离出来，做成一个中心化的限流器，实现全局共享。

分布式限流常见方案：可以借助Sentinel或者使用Redis来自己实现对应的限流逻辑。
基于Redis的实现步骤如下：(Redis + Lua)

1. 选取一个Redis中心化组件
2. 设定限流规则，比如每秒允许的最大请求数(QPS)，并且将该值存储在Redis中。
3. 每当有请求到达时，服务器首先会向Redis请求令牌。
4. 如果获得令牌，请求可以继续处理；如果未获得令牌，则表示请求被限流，执行拒绝策略。

local key = KEYS[1] -- hashmap的key
local rate = tonumber(ARGV[1]) -- 令牌生成速率
local capacity = tonumber(ARGV[2]) -- 桶的容量
local now = tonumber(ARGV[3]) -- 当前时间(毫秒)
local requested = tonumber(ARGV[4]) -- 请求的令牌数

-- 获取上次更新时间和当前令牌数
local last_time = redis.call('HGET', key, 'last_time')
local current_tokens = redis.call('HGET', key, 'tokens')

-- 如果是首次访问，那么就将last_time初始化为当前时间，当前令牌数为桶容量
if last_time == false then
    last_time = now
    current_tokens = capacity
else
    -- 不是第一次访问，那么从redis获取上次访问时间和当前令牌数
    last_time = tonumber(last_time)
    current_tokens = tonumber(current_tokens)
end

-- 计算时间差
local time_diff = now - last_time
-- 计算新的令牌数
local new_tokens = math.min(capacity, current_tokens + (time_diff * rate / 1000))

local allowed = new_tokens >= requested

if allowed then
    new_tokens = new_tokens - requested
end

redis.call('HSET', key, 'last_time', now)
redis.call('HSET', key, 'tokens', new_tokens)

return allowed