上周我的服务里有一个定时轮询文件夹的功能，每五秒用 os.listdir 扫一次，如果有新文件就送去处理。一开始文件不多还好，后面目录里堆了几万个小文件，listdir 一次就快两秒，扫描线程和处理线程互相掐架，最后还出现过一次重复处理事故。
后来决定改用 watchdog 监听文件系统事件。但是按 watchdog 默认写法，事件回调是在 watchdog 自己的线程里触发的，如果你在回调里直接跑耗时的处理逻辑，会拖垮整个观察器线程。所以最自然的搭档是：把事件扔进 asyncio.Queue，然后让主循环里面的协程慢慢取数据、慢慢处理。
下面我就把完整做法写出来，包括队列怎么搭、事件怎么去重、以及关闭时怎么才能不留下孤儿任务。
为什么不是 watchdog 官方示例里的那种 start()
watchdog 默认给了一个 Observer 类，你给它一个 EventHandler，然后 observer.start()。那个事件处理函数会在 observer 的线程里被调用，同一时刻只有一个线程在处理事件。如果你的 handler 是同步的，而且里面调用了 time.sleep() 或者 CPU 密集任务，你会发现后续的文件事件全都被堵在门口。
既然业务里本来就有异步循环，那就别让 watchdog 线程成为瓶颈。我们只需要让事件处理函数做一件事：往队列里塞一条消息，然后立刻返回。真正干活的是另一头的协程。
基本设计：生产者是事件线程，消费者是协程
先装依赖，不用多解释：
pip install watchdog接下来我直接写了一个可以独立运行的 Python 文件。它做的是：监控某个目录下新出现的 .txt 文件，读取每个文件的前几行并打印出来。但扩展成什么处理逻辑都行——因为关键点在队列，不在业务。
import asyncio
import os
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
class FileQueueHandler(FileSystemEventHandler):
"""watchdog 事件处理器：所有事件只往队列里放，绝不处理实质任务"""
def __init__(self, loop, queue):
self._loop = loop
self._queue = queue
def on_created(self, event):
if event.is_directory:
return
# 协程安全：loop.call_soon_threadsafe 是跨线程放消息的标准姿势
self._loop.call_soon_threadsafe(self._queue.put_nowait, ('created', Path(event.src_path)))
def on_modified(self, event):
if event.is_directory:
return
self._loop.call_soon_threadsafe(self._queue.put_nowait, ('modified', Path(event.src_path)))
def on_moved(self, event):
if event.is_directory:
return
self._loop.call_soon_threadsafe(
self._queue.put_nowait,
('moved', Path(event.dest_path), Path(event.src_path))
)注意，我没有用 on_any_event，而是单独监听 created / modified / moved。因为实际文件被保存时，通常会触发多次 modified，比如编辑器临时写入，或者生成 .tmp 文件然后又重命名。我们后面要用一个简单的技巧来合并重复消息，所以这里先不做过多的判断。
消费协程与去重小设计
在真正的业务里，我不希望几个毫秒内对一个文件连续处理三次。比如 IDE 按住 Ctrl+S 时会触发两次 modified。所以我在消费者那边搞了一个“最近事件”字典，记录每个文件最近一次事件类型和时间。如果看到同一路径在某段时间内再次出现，就只更新它，不再丢进真正的处理函数。
async def process_event(loop, queue, debounce_seconds=0.3):
recent = {}
while True:
try:
event = await queue.get()
except asyncio.CancelledError:
break
# 重构一下不同的事件数据结构
if event[0] == 'moved':
kind, dest, src = event
path = dest
else:
kind, path = event
now = loop.time()
last_time = recent.get(path)
if last_time and now - last_time < debounce_seconds:
# 离上次处理太近，跳过本次
queue.task_done()
continue
recent[path] = now
# 真正的业务逻辑就放这里
try:
if kind in ('created', 'modified', 'moved'):
await handle_file(path, kind)
except Exception as exc:
# 单独把错误打印出来，避免后台任务一下子就没了
print(f'处理 {path} 时出错: {exc}')
finally:
queue.task_done()最近一条事件记录我用普通的 dict 放着，没做定期清理。但如果文件很多，且时间间隔很长，这个 dict 会越占越大。所以更讲究一点可以在每次循环里检查最近记录是否太旧，然后删除——不过这个文章先不展开。
模拟真实业务逻辑
async def handle_file(path: Path, event_type: str):
# 模拟耗时操作，验证整个管道没有阻塞观察器线程
print(f'[{event_type}] 开始处理 {path.name}')
await asyncio.sleep(0.1)
if path.exists() and path.suffix == '.txt':
with open(path, encoding='utf-8') as f:
first_lines = [next(f, '').strip() for _ in range(2)]
print(f'文件 {path.name} 首两行内容: {first_lines}')
else:
print(f'{path.name} 不是 txt，或者已经被移走了')如何启动和优雅关闭
启动时有个小坑：需要一个正在运行的事件循环传给 watchdog 处理器。如果你直接 asyncio.run(main())，在 main 里很容易获得循环。但要注意，watchdog 的 observer 不是异步对象，它需要单独 start。
async def main():
watch_path = Path('./watch_folder')
watch_path.mkdir(exist_ok=True)
loop = asyncio.get_running_loop()
queue = asyncio.Queue(maxsize=200)
event_handler = FileQueueHandler(loop, queue)
observer = Observer()
observer.schedule(event_handler, str(watch_path), recursive=True)
observer.start()
print(f'正在监控目录: {watch_path.resolve()}')
# 启动消费者任务
consumer = asyncio.create_task(process_event(loop, queue))
# 为了保持程序一直运行，可以等待 Ctrl+C
try:
await asyncio.Event().wait()
except KeyboardInterrupt:
pass
finally:
# 一定要先停止 observer，不再产生新事件
observer.stop()
observer.join()
# 然后取消消费者并等待它真正结束
consumer.cancel()
try:
await consumer
except asyncio.CancelledError:
pass
# 如果队列里还有没处理完的消息，可以再给一点时间
if not queue.empty():
print(f'还有 {queue.qsize()} 条事件没处理，全部丢弃')
while not queue.empty():
queue.get_nowait()
queue.task_done()
await queue.join()
print('已关闭监控服务')asyncio.Queue，这里使用的是无限队列（没有设置 maxsize）。如果生产速度极快，内存可能会涨。所以我加了一个 maxsize=200，当队列满了，put_nowait 会抛异常。不过现在的事件处理器里我没有处理满队列的情况，因为普通文件操作远没达到这个量级。但你如果监控超大目录的批量复制文件，可以考虑在事件回调里使用 put 而不是 put_nowait，或者忽略部分事件。
调试时发现的一个易错点
我一开始写的是：
self._loop.call_soon_threadsafe(self._queue.put_nowait, ...)这里如果把 queue 换成别的非线程安全对象，很容易出问题。但 put_nowait 本身不是线程安全的？其实 asyncio.Queue 并不保证跨线程安全。我们用的是 call_soon_threadsafe，它让回调在事件循环线程内执行，所以在事件循环里调用 put_nowait 是安全的。如果不用 call_soon_threadsafe，直接在 watchdog 线程里调用 queue.put_nowait，会触发 runtime error：非事件循环线程去操作队列，偶尔会丢数据。
另一种做法是直接用 self._loop.call_soon_threadsafe(queue.put_nowait, event)，但为保险，我这里包一层 put_nowait。
如果你不需要 debounce，可以直接消费者全收
有时候你可能想处理每一个修改事件，尤其是日志文件。那就把 process_event 里的“距上次太近就跳过”这段去掉。保持同样的架构，照样不会阻塞 watchdog 线程。
更进一步，如果你想对同一文件多次修改后只执行一次，可以把 recent 字典改成 dict[path, asyncio.Task]，每次 put 时取消上一个未执行的任务，重新安排一个延迟任务——那又是更复杂的 throttle 实现。我这里先不塞进去，避免文章变味。
运行效果
在监控目录下随便放一个 demo.txt，终端输出类似：
正在监控目录: /tmp/watch_folder
[created] 开始处理 demo.txt
文件 demo.txt 首两行内容: ['第一行内容', '第二行内容']
[modified] 开始处理 demo.txt
文件 demo.txt 首两行内容: ['第一行内容', '第二行内容']
... 比较重要的是你观察不到任何卡顿。哪怕我用一个快速复制几百个文件的脚本去灌目录，observer 线程依旧能实时把所有事件塞进队列，处理协程在后面一个一个消费。这种感觉比原先用 time.sleep 硬等轮询优雅太多了。
结尾想说的是
Python 的 asyncio 和 watchdog 本身都没什么新鲜感，但把它们组合起来用队列解耦，确实能解决很多后台工具的“假死”问题。
这种模式也很容易迁移到别的地方——比如把 RabbitMQ 回调、Redis 订阅消息全部丢给 asyncio.Queue，然后在主循环里统一处理。只要记住一条线：任何非异步线程产生的消息，想进入异步世界，就一定要通过 call_soon_threadsafe 要么直接 loop.call_soon_threadsafe，要么让那个线程使用 loop.run_in_executor 包装。一旦跨过这条线，整个异步系统怎么玩都顺了。
