Python Condition 条件变量详解_Python

threading.condition 是 python 多线程编程里用来做“条件同步”的工具，核心就是：‌让线程在条件不满足时等待（wait），条件满足后由其他线程通知（notify/notify_all）再继续跑‌。它内部自带一把锁（默认是 rlock），所以既能保证互斥，又能实现线程间的通信。‌‌

一、python condition 条件变量详

1、引言

在多线程编程中，线程之间经常需要协作：一个线程等待某个条件成立，另一个线程在条件满足时通知它继续执行。python 的 threading.condition 条件变量正是为此而生的同步原语。

本文基于 python 3.13 最新版本，深入讲解 condition 的原理、用法与实战场景，帮助你写出更健壮的多线程协作代码。

2、 condition 是什么

threading.condition 是 python 标准库 threading 模块提供的条件变量类。它本质上是对「锁 + 等待/通知机制」的封装，允许线程在条件不满足时释放锁并挂起等待，在条件满足时被唤醒并重新获取锁继续执行。

2.1 、核心概念

关联锁（lock）：condition 内部持有一个锁（默认是 rlock），用于保护共享资源。
等待（wait）：线程调用 wait() 后释放锁并阻塞，直到被通知。
通知（notify）：线程调用 notify() 唤醒一个等待中的线程；notify_all() 唤醒所有等待线程。

2.2、与 event、queue 的区别

同步原语	适用场景	特点
`event`	一次性事件触发	简单，但无法精确控制唤醒数量
`queue`	生产者-消费者	内部已封装锁与条件，开箱即用
`condition`	复杂条件协作	灵活，可精确控制等待与唤醒

3、基本用法

3.1、创建 condition

import threading
# 使用默认的 rlock
cond = threading.condition()
# 也可以传入自定义锁
lock = threading.lock()
cond = threading.condition(lock)

3.2、核心方法

方法	说明
`acquire()` / `release()`	获取/释放关联锁
`wait(timeout=none)`	释放锁并等待通知，被唤醒后重新获取锁
`notify(n=1)`	唤醒 n 个等待线程（默认 1 个）
`notify_all()`	唤醒所有等待线程
`wait_for(predicate, timeout=none)`	等待条件谓词为真，内部循环调用 `wait()`

3.3、标准使用模式

import threading
cond = threading.condition()
shared_data = []
def consumer():
    with cond:
        while not shared_data:
            cond.wait()
        item = shared_data.pop(0)
        print(f"消费: {item}")
def producer():
    with cond:
        shared_data.append("新数据")
        cond.notify()
t1 = threading.thread(target=consumer)
t2 = threading.thread(target=producer)
t1.start()
t2.start()
t1.join()
t2.join()

关键点：wait() 必须在持有锁的情况下调用，且判断条件必须用 while 循环而非 if，以防止虚假唤醒（spurious wakeup）。

4、实战案例：生产者-消费者模型

下面是一个更完整的生产者-消费者示例，使用 wait_for 简化条件判断：

import threading
import time
import random
class boundedbuffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = []
        self.cond = threading.condition()
    def put(self, item):
        with self.cond:
            # 缓冲区满则等待
            self.cond.wait_for(lambda: len(self.buffer) < self.capacity)
            self.buffer.append(item)
            print(f"生产: {item}, 缓冲区: {self.buffer}")
            self.cond.notify_all()
    def get(self):
        with self.cond:
            # 缓冲区空则等待
            self.cond.wait_for(lambda: len(self.buffer) > 0)
            item = self.buffer.pop(0)
            print(f"消费: {item}, 缓冲区: {self.buffer}")
            self.cond.notify_all()
            return item
def producer(buf, count):
    for i in range(count):
        buf.put(i)
        time.sleep(random.uniform(0.1, 0.3))
def consumer(buf, count):
    for _ in range(count):
        buf.get()
        time.sleep(random.uniform(0.1, 0.3))
if __name__ == "__main__":
    buf = boundedbuffer(capacity=3)
    threads = [
        threading.thread(target=producer, args=(buf, 5)),
        threading.thread(target=consumer, args=(buf, 5)),
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print("全部完成")

5、深入理解 wait 与 notify 的底层机制

5.1、 wait() 的执行流程

线程必须已持有关联锁。
wait() 释放锁，线程进入等待队列。
线程阻塞，直到被 notify() 唤醒。
唤醒后线程重新竞争获取锁。
获取锁后 wait() 返回，继续执行。

5.2 、为什么必须用 while 而不是 if

# 错误写法：可能因虚假唤醒而出错
with cond:
    if not shared_data:
        cond.wait()
    item = shared_data.pop(0)  # 可能 indexerror
# 正确写法：循环检查条件
with cond:
    while not shared_data:
        cond.wait()
    item = shared_data.pop(0)

虚假唤醒指线程在没有收到 notify() 的情况下被唤醒，虽然 cpython 中不常见，但跨平台实现中可能出现，while 循环是防御性编程的标准做法。

5.3、 notify 与 notify_all 的选择

notify()：只唤醒一个线程，适合「一个生产者对应一个消费者」的场景，开销更小。
notify_all()：唤醒所有线程，适合「多个消费者竞争」或「条件变化影响多个线程」的场景。

6、常见陷阱与最佳实践

6.1 、陷阱一：忘记持有锁

# 错误：wait() 前未获取锁
cond.wait()  # runtimeerror: cannot wait on un-acquired lock

6.2、陷阱二：在 wait 之后修改共享状态

wait() 返回后，线程虽然重新获取了锁，但条件可能已被其他线程改变，因此必须重新检查条件（即 while 循环）。

6.3、陷阱三：notify 后立即释放锁

with cond:
    shared_data.append(item)
    cond.notify()  # 推荐：在释放锁之前通知
# 锁在此处释放

在 with 块内调用 notify() 是推荐做法，因为唤醒的线程会立即尝试获取锁，若通知后仍持有锁，被唤醒线程会阻塞在锁获取上，但不会丢失通知。

6.4 、最佳实践清单

始终使用 with cond: 上下文管理器管理锁。
条件判断一律使用 while 循环。
优先使用 wait_for(predicate) 简化代码。
明确 notify() 与 notify_all() 的语义差异。
避免在持有锁时执行耗时操作。

7、性能与替代方案

7.1、性能考量

condition 基于锁实现，在高竞争场景下可能成为瓶颈。若只是简单的生产者-消费者，queue.queue 内部已优化，优先使用。

7.2、替代方案对比

方案	优点	缺点
`condition`	灵活、精确控制	需要手动管理锁与条件
`queue`	开箱即用、线程安全	灵活性较低
`event`	简单	无法精确控制唤醒数量
`asyncio.condition`	异步友好	仅适用于协程

8、总结

threading.condition 是 python 多线程协作的核心工具，掌握它的原理与正确用法，能让你写出高效且不易出错的多线程程序。核心要点回顾：

wait() 必须在持有锁时调用，且用 while 循环检查条件。
notify() 唤醒一个线程，notify_all() 唤醒全部。
优先使用 wait_for(predicate) 简化条件等待。
简单场景优先考虑 queue.queue。

希望本文能帮助你深入理解 condition 条件变量，在实际项目中灵活运用。

二、代码示例

import threading
import time
def demo_basic_notify():
    print("=" * 65)
    print("【1.基础演示：wait / notify 唤醒单个线程】")
    cond = threading.condition()
    def worker(name):
        with cond:
            print(f"{name}：进入临界区，开始等待条件")
            cond.wait()
            print(f"{name}：被唤醒，继续执行")
    t1 = threading.thread(target=worker, args=("线程1",))
    t2 = threading.thread(target=worker, args=("线程2",))
    t1.start()
    t2.start()
    time.sleep(1.5)
    with cond:
        print("\n>>>主线程 notify(1)，唤醒1个线程")
        cond.notify(n=1)
    time.sleep(1.5)
    with cond:
        print(">>>主线程 notify(1)，唤醒剩余线程")
        cond.notify(n=1)
    t1.join()
    t2.join()
    print()
def demo_notify_all():
    print("=" * 65)
    print("【2.notify_all() 一次性唤醒全部等待线程】")
    cond = threading.condition()
    def task(name):
        with cond:
            print(f"{name} 等待唤醒")
            cond.wait()
            print(f"{name} 收到广播唤醒")
    threads = [threading.thread(target=task, args=(f"t{i}",)) for i in range(3)]
    for t in threads:
        t.start()
    time.sleep(1.2)
    with cond:
        print("\n>>>执行 notify_all()")
        cond.notify_all()
    for t in threads:
        t.join()
    print()
def demo_wait_timeout():
    print("=" * 65)
    print("【3.wait(timeout) 超时等待，不被唤醒也自动返回】")
    cond = threading.condition()
    def timeout_task():
        with cond:
            print("子线程等待最多1秒")
            result = cond.wait(timeout=1.0)
            if result:
                print("被信号唤醒")
            else:
                print("等待超时返回，没有收到notify")
    t = threading.thread(target=timeout_task)
    t.start()
    t.join()
    print()
def demo_wait_for():
    print("=" * 65)
    print("【4.wait_for(predicate) 内置条件判断（替代while循环）】")
    cond = threading.condition()
    flag = false
    def predicate():
        return flag
    def wait_job():
        nonlocal flag
        with cond:
            ok = cond.wait_for(predicate, timeout=2)
            if ok:
                print("wait_for：条件变为true，继续运行")
            else:
                print("wait_for：超时，条件不满足")
    t = threading.thread(target=wait_job)
    t.start()
    time.sleep(1)
    with cond:
        flag = true
        cond.notify()
    t.join()
    print()
def demo_producer_consumer_standard():
    print("=" * 65)
    print("【5.标准生产者消费者模型（while防虚假唤醒）】")
    cond = threading.condition()
    buf = []
    max_cap = 3
    def producer(pid):
        for i in range(4):
            with cond:
                # while 判断！防止虚假唤醒
                while len(buf) >= max_cap:
                    print(f"生产者{pid}：缓冲区已满，等待消费")
                    cond.wait()
                item = f"item-{pid}-{i}"
                buf.append(item)
                print(f"生产 {item}  | buffer={buf}")
                cond.notify()
            time.sleep(0.3)
    def consumer(cid):
        for _ in range(4):
            with cond:
                while len(buf) == 0:
                    print(f"消费者{cid}：缓冲区为空，等待生产")
                    cond.wait()
                val = buf.pop(0)
                print(f"消费 {val}  | buffer={buf}")
                cond.notify()
            time.sleep(0.5)
    t_p = threading.thread(target=producer, args=(1,))
    t_c = threading.thread(target=consumer, args=(1,))
    t_p.start()
    t_c.start()
    t_p.join()
    t_c.join()
    print()
def demo_rlock_condition():
    print("=" * 65)
    print("【6.condition绑定rlock，支持嵌套上锁】")
    # 传入rlock实现可重入
    cond = threading.condition(lock=threading.rlock())
    def nested_func():
        with cond:
            print("第一层锁")
            with cond:
                print("第二层嵌套锁（rlock允许）")
    t = threading.thread(target=nested_func)
    t.start()
    t.join()
    print()
if __name__ == "__main__":
    demo_basic_notify()
    demo_notify_all()
    demo_wait_timeout()
    demo_wait_for()
    demo_producer_consumer_standard()
    demo_rlock_condition()

d:\user\01417804\桌面\pythonproject\.venv\scripts\python.exe d:\user\01417804\桌面\pythonproject\main.py 
=================================================================
【1.基础演示：wait / notify 唤醒单个线程】
线程1：进入临界区，开始等待条件
线程2：进入临界区，开始等待条件
>>>主线程 notify(1)，唤醒1个线程
线程1：被唤醒，继续执行
>>>主线程 notify(1)，唤醒剩余线程
线程2：被唤醒，继续执行
=================================================================
【2.notify_all() 一次性唤醒全部等待线程】
t0 等待唤醒
t1 等待唤醒
t2 等待唤醒
>>>执行 notify_all()
t1 收到广播唤醒
t2 收到广播唤醒
t0 收到广播唤醒
=================================================================
【3.wait(timeout) 超时等待，不被唤醒也自动返回】
子线程等待最多1秒
等待超时返回，没有收到notify
=================================================================
【4.wait_for(predicate) 内置条件判断（替代while循环）】
wait_for：条件变为true，继续运行
=================================================================
【5.标准生产者消费者模型（while防虚假唤醒）】
生产 item-1-0  | buffer=['item-1-0']
消费 item-1-0  | buffer=[]
生产 item-1-1  | buffer=['item-1-1']
消费 item-1-1  | buffer=[]
生产 item-1-2  | buffer=['item-1-2']
生产 item-1-3  | buffer=['item-1-2', 'item-1-3']
消费 item-1-2  | buffer=['item-1-3']
消费 item-1-3  | buffer=[]
=================================================================
【6.condition绑定rlock，支持嵌套上锁】
第一层锁
第二层嵌套锁（rlock允许）
进程已结束，退出代码为 0

到此这篇关于python condition 条件变量详解的文章就介绍到这了,更多相关python condition 条件变量内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

Python Condition 条件变量详解

2026年09月14日 • Python •我要评论

一、python condition 条件变量详

1、引言

2、 condition 是什么

2.1 、核心概念

2.2、与 event、queue 的区别

3、基本用法

3.1、创建 condition

3.2、核心方法

3.3、标准使用模式

4、实战案例：生产者-消费者模型

5、深入理解 wait 与 notify 的底层机制

5.1、 wait() 的执行流程

5.2 、为什么必须用 while 而不是 if

5.3、 notify 与 notify_all 的选择

6、常见陷阱与最佳实践

6.1 、陷阱一：忘记持有锁

6.2、陷阱二：在 wait 之后修改共享状态

6.3、陷阱三：notify 后立即释放锁

6.4 、最佳实践清单

7、性能与替代方案

7.1、性能考量

7.2、替代方案对比

8、总结

二、代码示例

相关文章:

发表评论


验证码：

Python Condition 条件变量详解

2026年09月14日 • Python •我要评论

一、python condition 条件变量详

1、引言

2、 condition 是什么

2.1 、核心概念

2.2、 与 event、queue 的区别

3、基本用法

3.1、 创建 condition

3.2、 核心方法

3.3、 标准使用模式

4、 实战案例：生产者-消费者模型

5、 深入理解 wait 与 notify 的底层机制

5.1、 wait() 的执行流程

5.2 、为什么必须用 while 而不是 if

5.3、 notify 与 notify_all 的选择

6、 常见陷阱与最佳实践

6.1 、陷阱一：忘记持有锁

6.2、 陷阱二：在 wait 之后修改共享状态

6.3、 陷阱三：notify 后立即释放锁

6.4 、最佳实践清单

7、 性能与替代方案

7.1、 性能考量

7.2、 替代方案对比

8、总结

二、代码示例

相关文章:

发表评论

2.2、与 event、queue 的区别

3.1、创建 condition

3.2、核心方法

3.3、标准使用模式

4、实战案例：生产者-消费者模型

5、深入理解 wait 与 notify 的底层机制

6、常见陷阱与最佳实践

6.2、陷阱二：在 wait 之后修改共享状态

6.3、陷阱三：notify 后立即释放锁

7、性能与替代方案

7.1、性能考量

7.2、替代方案对比