滑动窗口限流是接口限流场景中常用的算法，它通过将固定时间窗口拆分为多个更小的时间片段，统计当前时间范围内已处理的请求数量，当请求数达到阈值时拒绝后续请求，相比固定窗口限流能更平滑地控制流量，减少临界时间点的流量突刺问题。
滑动窗口限流核心原理
滑动窗口限流的核心逻辑是将一个完整的时间窗口（比如1秒）拆分为多个等份的小时间片（比如10个100毫秒的片），每个时间片记录该时间段内的请求数量。当新的请求到来时，先滑动窗口，移除已经过期的时间片，再统计当前窗口内所有时间片的请求总数，若总数小于限流阈值则允许请求，否则拒绝请求。
假设限流规则是1秒内最多处理10个请求，窗口拆分为10个100毫秒的片，当前时间在第350毫秒，那么有效窗口是第250毫秒到第350毫秒对应的时间片，统计这些片的请求总和即可判断是否触发限流。
基于时间片的简单实现
这种实现方式适合单机场景下的简单限流需求，使用列表存储每个时间片的请求计数，通过时间戳计算当前所属的时间片索引。
import time
class SlidingWindowLimiter:
def __init__(self, max_requests, window_seconds, slice_count):
"""
:param max_requests: 窗口内最大请求数
:param window_seconds: 窗口总时长（秒）
:param slice_count: 窗口拆分的时间片数量
"""
self.max_requests = max_requests
self.window_seconds = window_seconds
self.slice_count = slice_count
self.slice_duration = window_seconds / slice_count # 单个时间片时长
self.request_counts = [0] * slice_count # 存储每个时间片的请求数
self.last_update_time = time.time() # 上次更新时间
def _get_current_slice_index(self):
"""获取当前时间对应的时间片索引"""
current_time = time.time()
# 计算当前时间距离起始时间的偏移，取模得到时间片索引
return int((current_time % self.window_seconds) // self.slice_duration)
def _slide_window(self):
"""滑动窗口，重置过期的时间片（这里简化为每次请求时重新计算，实际可根据需求优化）"""
current_time = time.time()
# 计算距离上次更新过去了多少个时间片
passed_slices = int((current_time - self.last_update_time) // self.slice_duration)
if passed_slices > 0:
# 重置过去的时间片计数
for i in range(passed_slices):
index = (self._get_current_slice_index() - i - 1) % self.slice_count
self.request_counts[index] = 0
self.last_update_time = current_time
def allow_request(self):
"""判断是否允许请求"""
self._slide_window()
current_slice = self._get_current_slice_index()
# 统计当前窗口内所有时间片的请求总数
total_requests = sum(self.request_counts)
if total_requests < self.max_requests:
self.request_counts[current_slice] += 1
return True
return False
# 使用示例：1秒窗口，最多10个请求，拆分为10个时间片
limiter = SlidingWindowLimiter(max_requests=10, window_seconds=1, slice_count=10)
for i in range(15):
if limiter.allow_request():
print(f"请求{i+1}：允许通过")
else:
print(f"请求{i+1}：被限流拒绝")
time.sleep(0.05)
基于有序集合的分布式实现
如果是分布式场景，需要多个服务节点共享限流状态，可以使用Redis的有序集合（ZSET）实现，利用ZSET的分数存储请求时间戳，通过范围查询统计窗口内的请求数量。
import time
import redis
class RedisSlidingWindowLimiter:
def __init__(self, redis_client, key, max_requests, window_seconds):
"""
:param redis_client: Redis连接客户端
:param key: 限流键名
:param max_requests: 窗口内最大请求数
:param window_seconds: 窗口时长（秒）
"""
self.redis_client = redis_client
self.key = key
self.max_requests = max_requests
self.window_seconds = window_seconds
def allow_request(self):
"""判断是否允许请求"""
current_time = time.time()
window_start = current_time - self.window_seconds
# 开启Redis管道，保证操作原子性
pipe = self.redis_client.pipeline()
# 移除窗口外的过期请求记录
pipe.zremrangebyscore(self.key, 0, window_start)
# 统计当前窗口内的请求数量
pipe.zcard(self.key)
# 添加当前请求的时间戳
pipe.zadd(self.key, {str(current_time): current_time})
# 设置键过期时间，避免无效数据堆积
pipe.expire(self.key, self.window_seconds + 1)
results = pipe.execute()
# results[1]是zcard返回的当前窗口请求数
current_count = results[1]
if current_count < self.max_requests:
return True
else:
# 如果请求被拒绝，需要删除刚才添加的当前请求记录
self.redis_client.zrem(self.key, str(current_time))
return False
# 使用示例，假设Redis连接本地默认端口
# redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0)
# limiter = RedisSlidingWindowLimiter(redis_client, 'api_limit', max_requests=10, window_seconds=1)
# print(limiter.allow_request())
实现注意事项
- 时间片拆分的数量需要根据业务场景调整，拆分越细限流越精准，但内存和计算开销也会越大。
- 分布式实现中要注意Redis操作的原子性，避免并发场景下统计错误，建议使用管道或者Lua脚本实现。
- 单机实现的时间戳精度依赖系统时间，如果系统时间发生跳变可能会导致限流逻辑异常，生产环境需要做好时间同步。
- 限流阈值和窗口时长需要根据接口的实际处理能力设置，避免阈值过小导致正常请求被拒绝，或者阈值过大失去限流作用。
两种实现方案对比
| 实现方案 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 基于时间片的单机实现 | 单机应用、简单限流需求 | 实现简单，无需额外依赖，性能高 | 不支持分布式，进程重启后状态丢失 |
| 基于Redis的分布式实现 | 分布式系统、多节点共享限流 | 支持分布式，状态持久化，可跨节点共享 | 依赖Redis组件，网络开销略高 |