Python中如何用双端队列实现限流滑动窗口算法？-Python教程-PHP中文网

双端队列（deque）是构建滑动窗口限流器最轻量可靠的数据结构，因其popleft()和append()均为o(1)，而list.pop(0)为o(n)；需用time.monotonic()防系统时间回拨，并加锁保障并发安全。

python中如何用双端队列实现限流滑动窗口算法？

双端队列（deque）本身不直接实现限流，但它是构建滑动窗口计数器最轻量、最可靠的数据结构——关键在于维护时间戳+计数的有序序列，而非简单存请求。

为什么不用 `list` 而必须用 `deque`？

滑动窗口需要频繁在头部淘汰过期请求、在尾部追加新请求。用 list 时，pop(0) 是 O(n) 操作，高并发下会拖垮性能；deque 的 popleft() 和 append() 都是 O(1)。

错误写法：requests = []; requests.pop(0) → 每次删头都遍历整个列表
正确做法：from collections import deque; window = deque()
注意：deque 不支持随机索引（如 window[0] 可读，但 window[-1] 在极老 Python 版本中可能慢），只应按序操作

如何用 `deque` 存储时间戳并判断是否超限？

核心逻辑是：把每个请求的 time.time()（或 time.monotonic()）追加到 deque 尾部，然后从头部持续弹出早于 now - window_size 的时间戳，最后检查剩余长度是否 ≤ 最大请求数。

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

from collections import deque
import time
<p>class SlidingWindowLimiter:
def <strong>init</strong>(self, max_requests: int, window_seconds: float):
self.max_requests = max_requests
self.window = deque()
self.window_seconds = window_seconds</p><pre class="brush:python;toolbar:false;">def allow(self) -> bool:
    now = time.monotonic()  # 更安全，不受系统时间调整影响
    # 清理过期时间戳
    while self.window and self.window[0] < now - self.window_seconds:
        self.window.popleft()
    # 判断是否还有容量
    if len(self.window) < self.max_requests:
        self.window.append(now)
        return True
    return False