Python Condition 被唤醒以后仍要检查条件：写一个可关闭的有界缓冲区

消费者从Condition.wait返回，不代表队列里已经有数据。它只意味着等待过程结束、重新拿到了锁；在这之前，其他消费者可能先取走数据，通知方也可能根本没有增加数据。

正确的等待条件必须由共享状态决定，在持锁时反复检查。if not items: wait()把一次醒来当成数据许可；while not items: wait()或wait_for则把醒来当成重新判断的机会。区别不在循环写法漂亮，而在锁释放期间状态可能已经变化。

下面使用Python 3.12，先做一次明确的“只有通知、没有数据”，再写一个支持关闭的有界缓冲区。关闭的语义也要事先确定：不再接收新条目，已有条目允许排空，最后让消费者退出。

只有通知的那次等待，不能返回业务结果

保存condition_buffer.py。开头的小实验不用随机睡眠安排谁先执行，用Event确认消费者已经进入持锁等待区域。

from collections import deque
from threading import Condition, Event, Thread


def notification_is_not_data():
    condition = Condition()
    values = deque()
    waiting = Event()
    completed = Event()
    result = []
    def consumer():
        with condition:
            waiting.set()
            condition.wait_for(lambda: bool(values))
            result.append(values.popleft())
        completed.set()
    worker = Thread(target=consumer)
    worker.start()
    try:
        if not waiting.wait(2):
            raise RuntimeError('consumer did not start')
        with condition:
            condition.notify_all()  # Deliberately does not change the predicate.
        assert not completed.wait(0.05), 'notification must not fabricate data'
    finally:
        with condition:
            values.append('real item')
            condition.notify_all()
        worker.join(2)
    if worker.is_alive():
        raise RuntimeError('consumer failed to finish')
    assert result == ['real item']
    print('notification alone did not complete; real item delivered')


class ClosedBuffer(Exception):
    pass


class Buffer:
    def __init__(self, capacity):
        if type(capacity) is not int or capacity < 1:
            raise ValueError('positive integer capacity required')
        self._capacity = capacity
        self._items = deque()
        self._closed = False
        self._condition = Condition()

    def put(self, item):
        with self._condition:
            self._condition.wait_for(
                lambda: self._closed or len(self._items) < self._capacity
            )
            if self._closed:
                raise ClosedBuffer('buffer is closed for producers')
            self._items.append(item)
            self._condition.notify_all()

    def get(self):
        with self._condition:
            self._condition.wait_for(lambda: self._items or self._closed)
            if not self._items:
                raise ClosedBuffer('buffer is drained')
            item = self._items.popleft()
            self._condition.notify_all()
            return item

    def close(self):
        with self._condition:
            self._closed = True
            self._condition.notify_all()


def main():
    notification_is_not_data()
    buffer = Buffer(2)
    received = []
    def consume():
        while True:
            try:
                item = buffer.get()
            except ClosedBuffer:
                return
            received.append(item)
    worker = Thread(target=consume)
    worker.start()
    try:
        for item in range(5):
            buffer.put(item)
    finally:
        buffer.close()
    worker.join(2)
    if worker.is_alive():
        raise RuntimeError('consumer did not stop after draining')
    assert received == list(range(5))
    try:
        buffer.put(99)
        raise AssertionError('closed producer should be rejected')
    except ClosedBuffer:
        print('put after close: rejected')
    print('drained:', received)

if __name__ == '__main__':
    main()

执行 python3 condition_buffer.py。通知实验先确认没有业务结果，再追加真实条目，消费者才完成；缓冲区最终按顺序交付0到4，关闭后写入被拒绝。

0.05秒只用于确认受控失败路径没有提前完成，不是线程唤醒延迟基准。程序中的两秒join同样是实验故障探测，不能解释成Buffer每次put/get都支持两秒超时。方法本身可以一直等待，取消与时间预算若属于接口，应另外设计。

状态检查与修改必须共用同一把锁

容量检查如果放在with外面，两个生产者可能都观察到还有位置，随后各自入队，超过上限。Condition帮助等待，不会替分散在不同锁中的状态动作建立原子性。

本例把谓词、入队、出队和关闭全部放在同一Condition锁内。wait_for等待时释放锁，返回前重新取得锁，谓词也在这个同步边界内判断。调用方拿到item以后才做业务处理，避免把慢工作放在共享锁里。

notify并不释放当前持有的锁。通知方在with内完成状态更新，等它退出，等待方才有机会重新获得锁。若在持锁期间执行远程调用，即使已经notify，其他线程也可能继续堵在锁上。

关闭必须唤醒两种等待者

消费者可能在等非空，生产者可能在等非满。close把_closed设为True，并notify_all，让两类等待都能重新判断退出条件。只唤醒消费者，满队列里的生产者仍可能永远等不到可以写入的状态。

消费者检查“有条目就先取”，因此关闭后可以排空。生产者则在唤醒后先检查关闭，哪怕此时刚好有空位，也不再接受新输入。这两条顺序共同表达先前声明的结束合同。

若业务要求关闭后立即丢弃剩余项，应明确清空与资源回收规则，并调整测试。不要把当前排空实现的某一行顺手改掉，却仍沿用“全部条目都处理完成”的日志。

以后增加超时参数时，也要处理wait_for返回假值的路径。等待预算到期不代表谓词成立，不能随后照常popleft或append；应返回明确超时，并保持共享状态未被错误修改。截止时间最好覆盖整个等待过程，而不是每次被唤醒以后重新获得一份完整预算。

有些接口还要区分“等待超时”与“缓冲区已经关闭”。二者都可能结束等待，却对应不同调用方动作：超时可能继续等待或放弃，关闭则表示这份输入生命周期已经结束。把它们统一返回None，还会和本来允许入队的None载荷发生混淆，所以示例用单独异常表达关闭状态。

notify_all 简单，但不承诺公平

一个Condition同时服务非空和非满两个谓词，notify_all让所有等待者重新竞争，逻辑容易核对。数据量与等待者很多时，频繁全体唤醒可能增加竞争；标准Queue等成熟工具已有更完整的并发协议，普通应用不必为这份演示重写基础容器。

通知也不指定下一条由哪个消费者领取，不承诺先进等待者先获得锁。FIFO条目顺序与线程调度公平性不是同一保证。需要按租户轮转或任务优先级处理时，应在调度模型里表达，不能期待Condition自动提供。

这里没有任务完成计数，get成功只表示领取，不表示业务持久成功。处理失败的重试、确认和死信另有协议；不要因为消费者最终排空队列，就把整个批次标记为业务成功。

演示之外，异常退出也要有结束路径

main用finally关闭缓冲区，避免生产循环出错后让消费者永远等新数据。如果消费者可能自己抛业务异常，真实程序还应将失败传回协调者，关闭输入并等待其他线程结束，不能让生产者继续堵在已无消费者的满队列上。

因此这份Buffer是同步机制实验，正常路径把领取结果收集到列表，故意没有模拟外部业务失败。加入外部处理以后，任务所有者需要建立错误传播和收尾；线程函数抛异常不会自动把异常抛回启动它的main。

重新检查一个Condition等待点时，可以直接问三件事：当前谓词是什么，谁在同一锁里改变它，关闭或失败时谁负责唤醒。这三项都明确，wait返回以后才不会被误当成一张不需要再次验证的通行证。