题目重点：按秒聚合队列处理 hit/getHits，解释过期条件、单调时间要求与乱序事件替代方案。
题意与假设
这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。
- 训练版时间戳为非负整数秒，所有方法调用的时间戳全局非递减；同一秒可多次 hit。
- 定义窗口为 `(t−300,t]`，因此 t−300 的 hit 已过期，t−299 的 hit 仍有效。题目未明确端点例子，必须在面试中确认。
- 单进程内存类；并发写入要加锁或分片，分布式精确计数是另一题。
解题路线
- 维护按秒聚合的 deque `[timestamp,count]` 与窗口总数 total。
- 每次 hit/getHits 先推进时间，弹出 `timestamp<=t−300` 的桶并从 total 扣除。
- hit 若与末尾桶同秒则数量加一，否则新建桶；getHits 直接返回 total。
- 每桶最多入队出队一次，清理虽然单次可能弹多个桶，但对所有操作摊还 O(1)；8,399 次随机窗口查询已同原始 hit 全扫描核对。
手动走一遍
在 t=1 hit 两次、t=300 hit 一次，getHits(300)=3；getHits(301)=1，因为 t=1 正好落在边界 t−300，已过期。若在 t=301 再 hit，同秒新桶或已有桶加一，结果为 2。
Python 参考实现
下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。
代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。
from collections import deque
class HitCounter:
"""Monotonic integer-second timestamps, window (t-300, t]."""
def __init__(self) -> None:
self.seconds: deque[list[int]] = deque()
self.total = 0
self.latest_time: int | None = None
def _advance(self, timestamp: int) -> None:
if timestamp < 0 or (
self.latest_time is not None and timestamp < self.latest_time
):
raise ValueError("timestamps must be non-negative and monotonic")
self.latest_time = timestamp
while self.seconds and self.seconds[0][0] <= timestamp - 300:
self.total -= self.seconds.popleft()[1]
def hit(self, timestamp: int) -> None:
self._advance(timestamp)
if self.seconds and self.seconds[-1][0] == timestamp:
self.seconds[-1][1] += 1
else:
self.seconds.append([timestamp, 1])
self.total += 1
def get_hits(self, timestamp: int) -> int:
self._advance(timestamp)
return self.total
复杂度
每个 hit 桶只进出一次，摊还 O(1) 每调用；最坏单次查询可弹 O(b) 个桶，b≤300 个不同秒桶，空间 O(min(H,300))。循环数组也可固定 300 格，但须用秒标签防回绕误计。
易错点与追问
容易失分的边界
- 过期条件用 `<t−300` 而把 t−300 的 hit 错算进去。
- 每次 hit 存一条记录，高频流量下空间随 hit 数而非秒数增长。
- 时间戳乱序仍直接用队列，旧事件可能在错误时机加入。
- 只保存当前 total，无法在时间推进时知道应减去多少。
- 在多线程下把 total/队列更新当成原子操作。
继续深挖
- 如果事件可迟到 30 秒，如何定义 watermark 和结果更正？
- 为什么固定 300 格环形桶需要存真实秒标签？
- 若要按用户查询，状态如何分片、设置过期并控制热点用户？
- 统计集群中精确计数与近似低延迟计数分别怎么设计？
