题目重点:用每个 key 的已接受请求时间队列实现精确滑动窗口,明确窗口开闭、拒绝请求是否计数、时间单调与并发原子性。
题意与假设
这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。
- 训练版每个 key 在任意半开时间窗 (now−W,now] 最多放行 L 次,L≥1、W>0。
- 只记录已放行请求,拒绝不占未来额度;若产品按所有尝试计数,需改入队时机。
- 同 key 的 now 单调不减,可由调用者传入单调时钟;代码不是线程安全,生产环境需按 key 锁或原子存储。
解题路线
- 为每个 key 维护队列,按时间升序只存放行的时间戳。
- 收到请求时删除队头所有 t≤now−W 的历史,这些不属于当前半开窗口。队列长度若已达 L 则拒绝,否则 append(now) 并放行。
- 队列正确性:每次请求前队列恰保存当前窗口内已放行的时间,故长度判定与限额一致;每个时间戳只入队/出队一次。
- 如果要求拒绝请求也计入尝试阈值或多实例全局严格限额,必须重新定义状态;q55 讨论分布式配额与热点,不可把本地 deque 直接搬到多实例。
手动走一遍
L=1、W=5,t=0 放行;t=4.999 被拒;t=5 时旧 t=0 满足 t≤now−W,被移除,新请求放行。若 L=2 且两次同 t=0,第三次也在 t=0 则拒绝;相等时间戳可共存。
Python 参考实现
下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。
代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。
"""Exact per-key sliding-window limiter over accepted requests."""
from collections import defaultdict, deque
class SlidingWindowLimiter:
def __init__(self, max_requests: int, window_seconds: float):
if max_requests < 1 or window_seconds <= 0:
raise ValueError("positive limit and window required")
self.max_requests = max_requests
self.window_seconds = window_seconds
self.accepted = defaultdict(deque)
self.last_time = {}
def allow(self, key: str, now: float) -> bool:
previous = self.last_time.get(key, float("-inf"))
if now < previous:
raise ValueError("timestamps must be nondecreasing per key")
self.last_time[key] = now
times = self.accepted[key]
threshold = now - self.window_seconds
while times and times[0] <= threshold:
times.popleft()
if len(times) >= self.max_requests:
return False
times.append(now)
return True
复杂度
每次检查摊销 O(1),偶有 O(L) 连续过期删除;每个活跃 key 最多保存 L 个已接受时间,总空间 O(KL),另有 key 元数据。若 key 基数持续增长,需要空闲 key TTL 清理。
易错点与追问
容易失分的边界
- 把窗口误写成 [now−W,now] 而在边界多计一次。
- 拒绝也入队却又声称只限制已放行次数。
- 同 key 并发读取长度后都 append,导致超额。
- 进程多副本各自一个队列却声称全局限额。
- 无限保留从不再使用的 key。
继续深挖
- 多实例精确限流如何实现且代价是什么?
- 如果必须计数所有尝试以防暴力请求,状态怎样变化?
- 时间戳乱序或机器时钟回拨怎么办?
- 与 fixed window 和 token bucket 的突发行为有何不同?
