Redis 实现分布式限流：计数器、滑动窗口算法

各位观众，各位大爷大妈，各位靓仔靓女，晚上好！我是你们的老朋友，人称“Bug终结者”的程序猿老王。今天咱们不聊Bug，聊聊一个比Bug更让人头疼的东西——流量洪峰！🌊

想象一下，你的应用突然爆火，用户像潮水一样涌进来，服务器瞬间被淹没，响应慢如蜗牛，甚至直接崩溃！😱 这可不是闹着玩的，轻则用户体验极差，重则直接影响你的业务收入。

所以，今天老王就给大家带来一套独门秘籍，教你如何用Redis这把倚天剑，斩断流量洪峰，实现分布式限流，让你的应用在流量的海洋中也能稳如泰山！💪

一、流量如猛兽，限流是缰绳

在开始之前，我们先来聊聊为什么要限流。你想啊，你的服务器就像一辆小汽车，性能有限，如果一下子涌进来几百辆车，肯定会堵塞甚至抛锚。限流就像给这辆小汽车加了一个限速器，控制车流量，保证它在安全范围内运行。

限流的核心思想很简单：在单位时间内，限制访问某个接口或资源的请求数量。 超过限制的请求，要么拒绝，要么排队等待。

二、 Redis：限流的完美搭档

为什么要选择Redis来做限流呢？因为它有以下几个优点：

高性能： Redis基于内存操作，速度极快，可以承受高并发的请求。
原子性： Redis的操作是原子性的，可以保证并发场景下计数的准确性。
易用性： Redis提供了丰富的命令，方便我们实现各种限流算法。
分布式： Redis可以轻松实现分布式部署，满足大规模应用的限流需求。

简单来说，Redis就是限流界的“速度与激情”！🏎️

三、计数器算法：简单粗暴但有效

首先，我们来介绍一种最简单的限流算法——计数器算法。

原理：

为每个需要限流的接口或资源设置一个计数器。
每次有请求到达时，计数器加1。
如果计数器超过预设的阈值，则拒绝请求。
在每个时间窗口结束后，将计数器重置为0。

代码示例 (Python)：

import redis
import time

class CounterLimiter:
    def __init__(self, redis_host, redis_port, key_prefix, limit, period):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
        self.key_prefix = key_prefix
        self.limit = limit
        self.period = period

    def is_allowed(self, key):
        redis_key = f"{self.key_prefix}:{key}"
        now = int(time.time())
        with self.redis_client.pipeline() as pipe:
            pipe.incr(redis_key)
            pipe.expire(redis_key, self.period)
            count, _ = pipe.execute()

        if count > self.limit:
            return False
        return True

# 使用示例
limiter = CounterLimiter(redis_host='localhost', redis_port=6379, key_prefix='api_limit', limit=10, period=60) # 每分钟限制10次
user_id = "user_123"

if limiter.is_allowed(user_id):
    print(f"User {user_id}: Request allowed!")
    # 处理请求
else:
    print(f"User {user_id}: Request rejected (rate limit exceeded)!")

代码解释：

redis_key: 使用 key_prefix 和 key (例如用户ID) 拼接成唯一的 Redis 键。
redis_client.pipeline(): 使用 Redis 的 Pipeline 可以批量执行命令，提高效率。
pipe.incr(redis_key): 将 Redis 键对应的计数器加 1。如果键不存在，Redis 会自动创建并初始化为 0。
pipe.expire(redis_key, self.period): 设置 Redis 键的过期时间，保证计数器在每个时间窗口结束后自动重置。
pipe.execute(): 执行 Pipeline 中的所有命令，返回结果。
count: 从Redis返回的计数器值。

优缺点：

优点	缺点
实现简单，容易理解	存在临界问题，可能在时间窗口切换时允许超出限制的请求
性能高，适用于高并发场景	无法平滑限流，容易出现突发流量

临界问题：

假设我们的限流策略是每分钟限制10个请求。在第59秒的时候，已经有10个请求通过。如果在下一分钟的第1秒，又来了10个请求，那么这两秒内总共通过了20个请求，超过了限制。

适用场景：

计数器算法适用于对精度要求不高的场景，例如防止恶意刷单、限制用户访问频率等。

四、滑动窗口算法：更平滑，更精确

为了解决计数器算法的临界问题，我们引入了滑动窗口算法。

原理：

将时间窗口划分为多个小的时间片，每个时间片都有一个独立的计数器。
每次有请求到达时，找到请求所属的时间片，并将该时间片的计数器加1。
判断当前时间窗口内所有时间片的计数器之和是否超过预设的阈值。
随着时间的推移，窗口会向前滑动，移除旧的时间片，并添加新的时间片。

核心思想：

滑动窗口算法通过将时间窗口细分，可以更精确地控制请求速率，避免了计数器算法的临界问题。

代码示例 (Python)：

import redis
import time

class SlidingWindowLimiter:
    def __init__(self, redis_host, redis_port, key_prefix, limit, window_size, slot_size):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
        self.key_prefix = key_prefix
        self.limit = limit
        self.window_size = window_size #窗口大小，秒
        self.slot_size = slot_size #每个时间片的大小，秒
        self.slot_count = window_size // slot_size #时间片数量

    def is_allowed(self, key):
        redis_key = f"{self.key_prefix}:{key}"
        now = int(time.time())
        slot_index = now % self.slot_count #计算当前时间片索引

        with self.redis_client.pipeline() as pipe:
            # 删除最老的时间片
            pipe.zremrangebyscore(redis_key, 0, now - self.window_size)
            # 增加当前时间片的计数
            pipe.zadd(redis_key, {now: now})
            # 获取当前窗口内的总请求数量
            pipe.zcard(redis_key)
            pipe.expire(redis_key, self.window_size * 2) #设置过期时间，防止数据无限增长

            _, _, count = pipe.execute()

        if count > self.limit:
            return False
        return True

# 使用示例
limiter = SlidingWindowLimiter(redis_host='localhost', redis_port=6379, key_prefix='api_limit', limit=10, window_size=60, slot_size=1) # 每分钟限制10次，窗口大小60秒，每个时间片1秒
user_id = "user_123"

if limiter.is_allowed(user_id):
    print(f"User {user_id}: Request allowed!")
    # 处理请求
else:
    print(f"User {user_id}: Request rejected (rate limit exceeded)!")

代码解释：

window_size: 滑动窗口的大小，单位为秒。
slot_size: 每个时间片的大小，单位为秒。
slot_count: 时间片的数量，等于 window_size / slot_size。
redis_key: 使用 key_prefix 和 key (例如用户ID) 拼接成唯一的 Redis 键。
redis_client.zremrangebyscore(redis_key, 0, now - self.window_size)：删除窗口之外的旧数据。使用 Redis 的有序集合（Sorted Set）存储时间戳，通过 score 范围删除旧数据。
redis_client.zadd(redis_key, {now: now})：将当前时间戳添加到有序集合中，score 和 member 都设置为当前时间戳。
redis_client.zcard(redis_key)：获取有序集合中元素的数量，即当前窗口内的请求数量。
redis_client.expire(redis_key, self.window_size * 2)：设置过期时间，防止数据无限增长。设置为窗口大小的两倍，可以保证即使程序出现异常，旧数据也能被自动删除。

优缺点：

优点	缺点
可以更精确地控制请求速率，避免临界问题	实现相对复杂
可以平滑限流，减少突发流量的影响	需要维护时间片，占用更多的存储空间

适用场景：

滑动窗口算法适用于对精度要求较高的场景，例如API接口限流、防止恶意攻击等。

五、选择合适的算法：没有最好，只有最合适

那么，在实际应用中，我们应该选择哪种限流算法呢？

算法	优点	缺点	适用场景
计数器算法	实现简单，性能高	存在临界问题，无法平滑限流	对精度要求不高的场景，例如防止恶意刷单、限制用户访问频率等
滑动窗口算法	精度高，可以平滑限流	实现相对复杂，占用更多存储空间	对精度要求较高的场景，例如API接口限流、防止恶意攻击等

总而言之，没有最好的算法，只有最合适的算法。我们需要根据具体的业务场景和需求，选择最合适的限流算法。就像选择伴侣一样，适合自己的才是最好的！😉

六、分布式限流：单机不行，集群来凑

上面我们讲的都是单机限流，如果我们的应用部署在多个服务器上，就需要实现分布式限流。

原理：

将限流的逻辑放在一个中心化的组件中，例如Redis。
所有服务器都连接到这个中心化组件，共享同一个计数器或滑动窗口。
每次有请求到达时，服务器都向中心化组件请求，判断是否允许通过。

代码示例 (Python)：

上面的代码示例本身就已经是分布式限流了，因为我们使用了 Redis 作为中心化的计数器。只需要确保所有的服务器都连接到同一个 Redis 集群即可。

需要注意的点：

Redis集群： 为了保证高可用和高性能，建议使用Redis集群。
网络延迟： 分布式限流会引入网络延迟，需要根据实际情况调整限流策略。
并发竞争： 在高并发场景下，需要注意Redis的并发竞争问题，可以使用Lua脚本来保证操作的原子性。

七、总结：限流并非万能，监控才是王道

今天我们学习了如何使用Redis实现分布式限流，包括计数器算法和滑动窗口算法。但是，限流并非万能的，它只能解决部分问题。

我们还需要对系统进行全面的监控，及时发现和解决问题。就像医生一样，不仅要开药，还要定期检查，才能保证病人的健康。🩺

监控指标：

请求总量
成功请求数量
失败请求数量
平均响应时间
错误率

监控工具：

Prometheus
Grafana
ELK Stack

最后，老王想说：

限流就像一把双刃剑，用得好可以保护系统，用得不好可能会影响用户体验。我们需要 carefully 地选择合适的算法，并进行充分的测试和监控，才能让限流真正发挥作用。

希望今天的分享对大家有所帮助！如果大家还有什么问题，欢迎在评论区留言，老王会尽力解答。

谢谢大家！🙏

发表回复 取消回复

发表回复取消回复