watchdog + asyncio 组合最稳：Observer 必须在独立线程中 start()，事件回调用 asyncio.create_task 或 asyncio.Queue 转交协程处理，切勿在回调中阻塞；需预检路径权限、捕获回调异常并监控 observer.is_alive()。
用 watchdog + asyncio 组合是当前最稳的路
Python 标准库没有原生异步文件监控能力，watchdog 本身是同步阻塞的，但直接扔进 loop.run_in_executor 就能安全接入异步流程。别碰 aiopath 或轮询方案——前者不监控事件，后者 CPU 拉满还漏事件。
关键判断：必须用 Observer 的 start() 启动独立线程，再用 asyncio.to_thread（3.9+）或 loop.run_in_executor（兼容 3.11）桥接事件回调，不能把 Observer 实例扔进协程里调 start()。
-
Observer必须在主线程外启动，否则会卡住 asyncio event loop - 事件回调函数（如
on_created）里不要做耗时操作，否则积压事件队列；转发到asyncio.create_task处理更安全 - Windows 下注意路径大小写敏感性差异，Linux/macOS 要留意 inotify 句柄数限制（
/proc/sys/fs/inotify/max_user_watches）
on_created 回调里怎么发异步通知？
回调函数本身是同步的，但可以触发异步动作。常见做法是把事件推到 asyncio.Queue，由单独的消费者协程消费；或者直接用 asyncio.create_task 启动处理逻辑（需确保 loop 已运行）。
示例片段（简化）：
立即学习“Python免费学习笔记（深入）”；
import asyncio from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler <p>queue = asyncio.Queue()</p><p>class AsyncHandler(FileSystemEventHandler): def on_created(self, event): if not event.is_directory:</p><h1>立即丢进异步队列，不阻塞 watchdog 线程</h1><pre class='brush:python;toolbar:false;'> asyncio.create_task(queue.put(event.src_path))
async def consume_events():
while True:
path = await queue.get()
print(f"New file: {path}")
这里可加 aiofiles 写日志、aiohttp 通知 webhook 等
queue.task_done()
启动 observer（在后台线程）
observer = Observer() observer.schedule(AsyncHandler(), path="/tmp/dir1", recursive=False) observer.schedule(AsyncHandler(), path="/tmp/dir2", recursive=True) observer.start()
启动异步消费者
asyncio.create_task(consume_events())
监控多个目录时，recursive 参数要按需分开设
同一个 Observer 实例可添加多个 schedule()，但每个路径的 recursive 是独立控制的——比如你只想递归监控 /logs，但只看 /uploads 顶层新文件，就必须调两次 schedule() 并传不同 recursive 值。
- 重复调用
observer.schedule()是合法且推荐的，不是 bug - 如果某目录权限不足或不存在，
schedule()不报错，但后续事件不会触发；建议先用os.path.isdir()和os.access(path, os.R_OK)预检 - 路径末尾斜杠不影响行为，但统一用
os.path.abspath()规范化，避免因软链接或相对路径导致重复监听同一物理位置
Python 3.11 下容易崩的点：异常没被 observer 线程捕获
Observer 在自己的线程里跑，如果 on_created 里抛出未捕获异常，整个 observer 会静默退出（observer.is_alive() 变 False），但主程序毫无感知。
- 务必在所有事件回调里包
try/except Exception，至少打日志 - 检查
observer.is_alive()状态，定期重连（比如每 30 秒if not observer.is_alive(): observer.start()） - 别依赖
observer.join(timeout=...)做健康检查——它只是阻塞等待线程结束，不是心跳 - 3.11 的
ExceptionGroup不会影响 watchdog，但如果你在回调里用了asyncio.gather(..., return_exceptions=True)，记得 unpack 异常再处理
真正麻烦的是跨线程异常传播不可见，得靠日志和存活检测双保险。
