请求限流算法与数据结构选型：高频接口滑动窗口限流需求

针对你这种每秒被高频调用的do_something(timestamp_in_ms)接口，要实现精确的滑动时间窗口限流（任意连续1秒窗口内仅处理前N个请求，其余直接丢弃），核心要兼顾「低延迟」和「线程安全」——毕竟调用频率极高，限流逻辑绝不能成为业务的瓶颈。

核心思路 #

滑动窗口的关键是实时计算「当前时间往前推1秒」这个窗口内的请求数。为了高性能，我们要避免全量遍历历史请求，而是利用有序时间戳集合+二分查找快速清理过期数据，同时严格控制当前窗口内的请求数不超过阈值N。

具体执行逻辑：

• 维护一个按时间戳递增的有序列表，保存最近1秒内已处理请求的时间戳
• 每次新请求到来时：
  1. 计算窗口左边界：current_window_start = timestamp_in_ms - 1000
  2. 从列表中移除所有早于current_window_start的时间戳（这些请求已脱离当前1秒窗口）
  3. 如果列表长度小于N，就执行do_something并将当前时间戳加入列表；否则直接丢弃请求

高性能优化点 #

1. 有序列表+二分查找：用二分查找快速定位过期时间戳的分界点，避免逐个遍历删除，时间复杂度从O(n)降到O(logn)（查找）+ O(k)（删除过期元素，k通常远小于n）
2. 极小锁粒度：仅在操作有序列表时加锁，业务逻辑do_something不持有锁，避免锁阻塞拖慢主业务
3. 内存可控：列表最多保存N个时间戳，不会出现内存溢出风险

Python 实现代码 #

import bisect
import threading
from typing import List

class SlidingWindowLimiter:
    def __init__(self, max_requests: int, window_ms: int = 1000):
        self.max_requests = max_requests
        self.window_ms = window_ms
        self.timestamps: List[int] = []
        self.lock = threading.Lock()

    def allow_request(self, timestamp_in_ms: int) -> bool:
        with self.lock:
            # 计算当前窗口的左边界
            window_start = timestamp_in_ms - self.window_ms
            # 用二分查找定位第一个有效时间戳的位置，截断过期数据
            idx = bisect.bisect_left(self.timestamps, window_start)
            del self.timestamps[:idx]

            # 检查当前窗口内请求数是否未达上限
            if len(self.timestamps) < self.max_requests:
                bisect.insort(self.timestamps, timestamp_in_ms)
                return True
            return False

# 用法示例
limiter = SlidingWindowLimiter(max_requests=100)  # 每秒最多处理100个请求

def do_something(timestamp_in_ms: int):
    # 这里是你的业务逻辑
    pass

def wrapped_do_something(timestamp_in_ms: int):
    if limiter.allow_request(timestamp_in_ms):
        do_something(timestamp_in_ms)
    else:
        # 丢弃请求，可按需记录日志或返回错误
        pass

极端场景补充说明 #

如果接口调用频率达到每秒百万级，Python的GIL可能成为瓶颈，此时可以考虑：

• 用多进程部署限流逻辑，每个进程维护独立的滑动窗口（适合无状态接口）
• 若为分布式场景，改用Redis ZSet实现分布式滑动窗口限流（但单机场景下本地实现性能最优）

内容的提问来源于stack exchange，提问作者Zizheng Tai