Netflix Blocking Queue：Offer、Poll、Peek、Size 与条件变量

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 训练版容量固定且大于 0；offer(item,timeout) 满时阻塞，poll(timeout) 空时阻塞；timeout=None 无限等待，超时 offer 返回 False、poll 返回 None。
• 不允许将 None 作为 item，否则 poll 的空/超时返回会歧义；示例未实现 close/shutdown。
• peek/size 给调用瞬间受同一 mutex 保护的快照，但返回后队列可立刻变化。

解题路线

1. Condition 内封装一把锁与 deque。offer 在 `while size>=capacity` 中 wait，醒后必须重查，因为可能虚假唤醒或其他生产者先占空位；成功 append 后通知等待消费者。
2. poll 同样在 `while empty` 中 wait，成功 popleft 后通知等待生产者。有限 timeout 一次计算 monotonic deadline，每次醒后计算剩余时间，避免反复等待导致超过总超时。
3. peek 与 size 也持同一锁，确保读一致；notify_all 简单正确但可能造成惊群，生产优化可用两个 Condition（notFull/notEmpty）配合同一锁，并基于压测选 notify 方式。

手动走一遍

训练例容量 1，生产者 offer A 后队列满，另一 offer B 阻塞；消费者 poll 得 A，通知生产者，B 入队。两个消费者同时等待空队列时，只加一个 item，醒来的两个线程中只有一个能取到，另一个必须重新等待。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from collections import deque
from threading import Condition
from time import monotonic


class BlockingQueue:
    def __init__(self, capacity):
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        self._capacity = capacity
        self._items = deque()
        self._condition = Condition()

    def offer(self, item, timeout=None):
        if item is None:
            raise ValueError("None is reserved for empty poll")
        if timeout is not None and timeout < 0:
            raise ValueError("negative timeout")
        deadline = None if timeout is None else monotonic() + timeout
        with self._condition:
            while len(self._items) >= self._capacity:
                remaining = None if deadline is None else deadline - monotonic()
                if remaining is not None and remaining <= 0:
                    return False
                self._condition.wait(remaining)
            self._items.append(item)
            self._condition.notify_all()
            return True

    def poll(self, timeout=None):
        if timeout is not None and timeout < 0:
            raise ValueError("negative timeout")
        deadline = None if timeout is None else monotonic() + timeout
        with self._condition:
            while not self._items:
                remaining = None if deadline is None else deadline - monotonic()
                if remaining is not None and remaining <= 0:
                    return None
                self._condition.wait(remaining)
            item = self._items.popleft()
            self._condition.notify_all()
            return item

    def peek(self):
        with self._condition:
            return self._items[0] if self._items else None

    def size(self):
        with self._condition:
            return len(self._items)

复杂度

offer/poll/peek/size 在锁内都是 O(1)，队列空间 O(C)。阻塞等待时间不由 Big-O 给出，取决于生产/消费速率；notify_all 的唤醒成本随等待线程数增加。

易错点与追问

容易失分的边界

• 用 if 而非 while 检查等待条件，虚假唤醒或竞争后越界。
• 在锁外读 size 或 peek，使状态判断和操作不一致。
• 每次 wait 都用原始 timeout，导致总等待可无限延长。
• 忘记 poll 后通知满队列生产者或 offer 后通知空队列消费者。
• 允许 None 入队却也用 None 代表超时。

继续深挖

• 如何支持 close 后唤醒所有阻塞线程？
• 公平性是否需要保证，Condition 能否提供严格 FIFO？
• 如果容量可动态调整，等待中的生产者如何重新判断？
• 与无锁队列相比在哪些负载下值得复杂化？