Python collections.deque maxlen 实现滚动窗口的更新方案

处理“只保留最近几条数据”的需求时，直接使用 collections.deque(maxlen=N) 最稳妥。它把容量写进容器：用 append() 按时间顺序加入新记录，窗口已满时自动丢掉最旧的一端；如果数据从左侧加入，则从右端淘汰。关键是先约定新旧方向，再选择追加方法。

官方地址：https://docs.python.org/3.15/library/collections.html

先把滚动窗口的淘汰方向定下来

一次故障复盘中，窗口里明明设置了 3 条记录，结果展示出来的并不是“最近三次”，原因不是 maxlen 失效，而是生产代码混用了两种追加方向。建议把最旧记录放在左边、最新记录放在右边，这样阅读和排查都更直观。

from collections import deque

# maxlen 把窗口容量固定为 3，左端代表最旧记录
recent = deque(maxlen=3)
for event in ("A", "B", "C", "D"):
    # 新记录从右端进入，满容量时自动丢弃左端的 A
    recent.append(event)

print(list(recent))  # ['B', 'C', 'D']

这个行为适合最近日志、最近价格、最近几次状态变化等场景。它只保存当前窗口，旧记录被淘汰后不会自动归档；如果还要追溯完整历史，应把持久化写入单独的数据流。

append、appendleft 和读取边界怎么配套

如果上游数据已经按时间从旧到新到达，使用 append() 即可。若接收端总是先拿到最新值，使用 appendleft()，此时最新值在下标 0，最旧值在右端。

from collections import deque

def push_latest(window, value, newest_first=False):
    # 调用方明确顺序，避免同一窗口混用两种方向
    if newest_first:
        window.appendleft(value)
    else:
        window.append(value)
    # 返回快照，避免把内部 deque 直接暴露给修改方
    return tuple(window)

latest_first = deque(maxlen=3)
print(push_latest(latest_first, "C", True))
print(push_latest(latest_first, "B", True))
print(push_latest(latest_first, "A", True))  # ('A', 'B', 'C')

读取时优先使用 len(window) 判断当前是否有数据，再访问 window[0] 或 window[-1]。空 deque 直接取下标会抛出 IndexError，不要把它误判成容量配置错误。

别把 insert 当成有界窗口追加

maxlen 只约束两端追加和扩展的滚动语义。满容量时调用 insert(i, value)，Python 会抛出 IndexError，因为它无法替你决定应该从哪一端丢数据。需要“新值进来、旧值出去”时，统一封装 append 或 appendleft，不要在业务代码里随机插入。

容量参数也要在入口处校验：maxlen=0 会让追加后的 deque 仍保持为空；负数容量会直接报错。对外暴露时，可以只允许正整数，把“是否保留历史”的需求分流到数据库、文件或专门的消息队列。

上线前的四项核对

• 确认下标 0 和 -1 分别代表最新还是最旧，写在函数文档中。
• 只让一个封装函数负责追加，避免调用方混用方向。
• 为空时先判断长度，再读取首尾元素。
• 窗口用于短期状态；需要审计、回放或跨进程消费时，另设持久化边界。

常见问题

deque 的 maxlen 满了会主动报错吗？

使用 append、appendleft 或 extend 时通常会自动淘汰另一端；insert 在满容量时才会因为无法决定淘汰方向而抛出 IndexError。

为什么不用 list 加 pop(0)？

固定容量的两端进出是 deque 的直接语义，避免在列表头部移动大量元素；如果还需要频繁随机访问中间位置，应重新评估数据结构。

maxlen 能保证多进程之间的窗口一致吗？

不能。它只约束单个 deque 对象的内存状态，不提供跨进程同步、持久化或故障恢复。