题目重点：把 KV 正确性与调用计数分开，解释 300 秒环、时间注入、固定窗口/启动期分母及可变长窗口。
题意与假设
这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。
- 训练版 put/get 各算一次调用，get miss 也计数；操作在单进程、单线程且时间戳为单调递增整数秒。put 覆盖值，get 不存在返回 None。题目没有统一 get miss、并发和 timestamp 契约。
- 最近 W 秒定义为整数秒桶 [now-W+1, now]；固定窗口 QPS 分母 W，观测启动期 QPS 分母 min(W,now-start+1)。选择哪种必须与面试官确认。
解题路线
- KV 用 hash map 保存当前值；Put/Get 事件单独落入按 second mod W 寻址的环。每槽保存绝对秒数与两种操作计数，复用槽时重置，避免 300 秒前的旧值混入。
- 查询只加总秒数处于窗口内的桶，示例代码 O(W)；若固定 W 且要求高 QPS 查询，可维护两个滚动总数，并在时间前进时逐槽过期。若时间可能跳过大量秒，最多检查 W 个槽，不能按跳过秒数线性循环。
- 时钟作为构造依赖注入而非硬编码 time.time()，便于测试边界；若调用可能乱序，需要事件时间缓存/水位，不能直接用固定环覆盖。
- 窗口可变到一小时、一日甚至更久时，单个 300 秒环不足；用保留事件的时间序列加前缀和做精确任意窗口，或多级时间桶换取范围/边界近似，并说明精度。
- 若面试官要分布式全局速率，单机环只给本机数据；需按实例聚合并处理时间偏差、重复事件及重启归零。
手动走一遍
服务在 t=0 启动并 put 一次：固定 300 秒定义为 1/300 QPS，启动期定义为 1/1 QPS；两者都可能合理。到 t=300，t=0 的事件已退出 [1,300]，应为 0。t=299 时它仍在窗口。
Python 参考实现
下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。
代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。
"""Training contract for KV calls per second in the trailing 300 integer seconds.
Events at time t count in [t-299,t]. Clock timestamps are monotonic integers.
The rate denominator is selectable: fixed 300 seconds, or observed uptime
min(300, t-start+1). The posts disagree on whether warm-up uses uptime.
"""
from collections.abc import Callable
class KVLoad:
def __init__(self, clock: Callable[[], int], start: int, window: int = 300):
if window <= 0:
raise ValueError('window must be positive')
self.clock, self.start, self.window = clock, start, window
self.last_time = start
self.values: dict[str, str] = {}
self.seconds = [-1] * window
self.gets = [0] * window
self.puts = [0] * window
def _now(self) -> int:
now = self.clock()
if not isinstance(now, int) or now < self.start or now < self.last_time:
raise ValueError('clock must be monotonic integer seconds')
self.last_time = now
return now
def _record(self, now: int, operation: str) -> None:
index = now % self.window
if self.seconds[index] != now:
self.seconds[index] = now
self.gets[index] = self.puts[index] = 0
if operation == 'get':
self.gets[index] += 1
else:
self.puts[index] += 1
def put(self, key: str, value: str) -> None:
now = self._now()
self.values[key] = value
self._record(now, 'put')
def get(self, key: str) -> str | None:
now = self._now()
self._record(now, 'get')
return self.values.get(key)
def average(self, operation: str, denominator: str = 'fixed') -> float:
now = self._now()
if operation not in ('get', 'put') or denominator not in ('fixed', 'observed'):
raise ValueError('unknown operation or denominator')
buckets = self.gets if operation == 'get' else self.puts
earliest = now - self.window + 1
count = sum(buckets[i] for i, second in enumerate(self.seconds) if second >= earliest)
span = self.window if denominator == 'fixed' else min(self.window, now - self.start + 1)
return count / span
复杂度
每次 put/get 期望 O(1) 时间；示例 average 扫 W 槽为 O(W) 时间，KV 空间 O(K) 加 O(W) 统计空间。若维护滚动总数，固定窗口查询可 O(1)，但时钟推进和并发锁范围需正确实现。
易错点与追问
容易失分的边界
- 只维护一个总数，不保存桶时间导致环回卷错误。
- 把固定窗口分母 300 和启动期分母混为唯一真相。
- 把 get miss 排除计数而未问 API 语义。
- 直接使用真实时钟使边界测试不确定。
- 请求时间戳乱序或进程重启后沿用单调环。
继续深挖
- 查询 QPS 很高时怎样做到 O(1)？
- 300 天可变窗口怎样保精度？
- 稀疏事件和突发事件分别适合 deque 还是固定桶？
- 多线程 put/get 与计数要怎样保证一致？
- 重启后平均速率从何时计算？
