实现线程安全的阻塞队列。

核心回答

用一把锁保护队列内容和状态，用条件变量让消费者在空队列时等待、生产者在有界队列满时等待。等待时释放锁，被唤醒后重新取得锁并用 while 检查条件，不能假设一次通知就一定满足自己的条件。

入队、出队和通知放在同一个临界区。关闭队列时设置关闭状态并唤醒所有等待者，明确“关闭后拒绝新元素，但已入队元素仍可被取完”的契约，避免线程永久等待。

Python 实现

下面实现有界队列，put() 在满时阻塞，get() 在空时阻塞；关闭后 put() 抛错，get() 排空已有元素后抛出 EOFError。

from collections import deque
from threading import Condition

class BoundedBlockingQueue:
    def __init__(self, capacity):
        if type(capacity) is not int or capacity <= 0:
            raise ValueError("capacity must be positive")
        self.capacity = capacity
        self.items = deque()
        self.closed = False
        self.condition = Condition()

    def put(self, value):
        with self.condition:
            while len(self.items) >= self.capacity and not self.closed:
                self.condition.wait()
            if self.closed:
                raise RuntimeError("queue closed")
            self.items.append(value)
            self.condition.notify_all()

    def get(self):
        with self.condition:
            while not self.items and not self.closed:
                self.condition.wait()
            if not self.items:
                raise EOFError("queue drained and closed")
            value = self.items.popleft()
            self.condition.notify_all()
            return value

    def close(self):
        with self.condition:
            self.closed = True
            self.condition.notify_all()

为什么使用 while 和 notify_all

虚假唤醒、多个消费者竞争同一个元素，以及被通知后其他线程先修改状态，都会让等待条件再次不满足，所以被唤醒后必须重新判断。

这个简化实现把“非空”和“非满”两类等待者放在同一个条件变量中，使用 notify_all() 让各方重新检查自己的谓词。更细的实现可以在同一把锁上建立两个条件变量，分别通知生产者和消费者，减少无效唤醒。

close() 不清空数据，也不强行终止正在处理已领取元素的线程。需要超时接口时，用单调时钟计算剩余等待时间，循环扣减，避免每次唤醒都重新等待一个完整超时。队列只保护元素的传递，不自动保证取出后的业务处理恰好一次。

参考资料