Python等待条件成立再执行代码的最佳实践与高效实现方法

嘿，这个问题问到点子上了——刚上手的时候用while True:死循环轮询确实是最直观的，但那种“忙等”会把CPU拉满，完全是没必要的浪费。下面我给你分享几种行业里常用的高效实现方式，分场景给你讲清楚：

不管用哪种方案，核心都是让程序在等待时进入阻塞状态，而不是持续占用CPU循环检查。操作系统会把CPU资源让给其他进程，直到条件触发再唤醒你的程序，这才是高效的做法。

1. 监控文件系统变化：用watchdog库（首推方案） #

如果你是要检测文件/目录的修改、创建、删除这类操作，watchdog绝对是行业标准。它基于操作系统原生的文件事件通知机制（Linux上的inotify、macOS的FSEvents、Windows的ReadDirectoryChangesW），完全不需要轮询，只有当文件真的发生变化时才会触发回调，CPU占用几乎可以忽略。

示例代码： #

首先安装依赖：

pip install watchdog

然后写一个监控脚本：

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class MyFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 排除目录修改的情况，只处理文件
        if not event.is_directory:
            print(f"文件 {event.src_path} 被修改了！")
            # 这里写你要执行的特定代码
            # do_something()

if __name__ == "__main__":
    # 要监控的文件路径
    target_file = "/path/to/your/target/file.txt"
    event_handler = MyFileHandler()
    observer = Observer()
    # 监控文件所在的目录，recursive=False表示只监控当前目录
    observer.schedule(event_handler, path="/path/to/your/directory", recursive=False)
    observer.start()
    
    try:
        # 让脚本持续运行，直到手动中断（Ctrl+C）
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

这个脚本会一直运行，只有当目标文件被修改时才会执行你的业务代码，平时几乎不占CPU。

2. 通用条件等待：用threading.Event（适合自定义业务逻辑） #

如果你的等待条件不是文件系统事件，而是自定义的业务逻辑（比如等待某个变量变为True、等待某个异步任务完成、等待API返回特定结果），那么threading.Event是绝佳选择。它的wait()方法会让线程进入阻塞状态，直到被set()唤醒，完全不占用CPU。

示例代码： #

import threading
import time

# 创建一个事件对象
condition_event = threading.Event()

def check_condition():
    # 这里模拟持续检查条件的逻辑，比如每隔一段时间检查一次
    while True:
        # 假设这里是你的条件判断，比如检查某个变量、API状态等
        # 替换成你实际的条件判断逻辑
        some_custom_condition_is_true = lambda: time.time() % 10 == 0
        if some_custom_condition_is_true():
            # 条件满足，唤醒等待的线程
            condition_event.set()
            # 重置事件，方便下次等待
            condition_event.clear()
        time.sleep(5)  # 合理设置检查间隔，避免太频繁

def do_when_condition_met():
    while True:
        # 阻塞等待条件触发
        condition_event.wait()
        print("条件满足！执行特定代码...")
        # 这里写你要执行的业务代码
        # run_your_task()

if __name__ == "__main__":
    # 启动检查条件的线程
    checker_thread = threading.Thread(target=check_condition, daemon=True)
    checker_thread.start()
    
    # 启动执行任务的线程
    worker_thread = threading.Thread(target=do_when_condition_met, daemon=True)
    worker_thread.start()
    
    # 让主线程持续运行，直到手动中断
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("脚本已终止")

这种方式把条件检查和任务执行分开，主线程只负责保持脚本运行，其他线程各司其职，CPU占用极低。

3. 简单场景：带合理间隔的轮询（无依赖方案） #

如果你的场景非常简单，而且不想引入任何第三方库，那可以优化你现有的while True:循环——关键是加上合适的time.sleep()间隔，并且只在必要时检查条件。

比如检测文件修改的示例：

import os
import time

def wait_for_file_modify(file_path):
    # 获取初始修改时间
    last_modified = os.path.getmtime(file_path)
    while True:
        current_modified = os.path.getmtime(file_path)
        if current_modified != last_modified:
            print("文件被修改了！")
            last_modified = current_modified
            # 执行你的代码
            # do_something()
        # 每隔2秒检查一次，平衡响应速度和CPU占用
        time.sleep(2)

if __name__ == "__main__":
    wait_for_file_modify("/path/to/your/file.txt")

这种方案虽然还是轮询，但因为有sleep，CPU占用会非常低（几乎可以忽略），适合快速实现的简单场景。

• 文件/目录监控：优先用watchdog，原生事件驱动，最高效，也是行业标准
• 自定义业务条件：用threading.Event，阻塞等待不占CPU，适合复杂逻辑
• 简单无依赖场景：带合理sleep的轮询，快速实现，CPU占用可控

内容的提问来源于stack exchange，提问作者superx