在自动化构建、日志采集和配置热更新等场景中,及时感知目录内文件的增删改是核心需求。Python生态里最成熟的方案是watchdog库,它封装了不同系统的原生文件监控接口,让开发者用统一API拿到精准事件。

一、为什么不用轮询而用事件驱动
早期做法是用os.listdir加上time.sleep每隔几秒扫一遍目录,比对文件大小和修改时间。这种方式实现简单,但缺陷明显:扫描间隔设太短会吃掉CPU,设太长又无法实时响应;而且每次全量遍历在文件多时非常慢。
现代操作系统提供了inotify(Linux)、kqueue(macOS)、ReadDirectoryChangesW(Windows)等机制,内核在文件变更时主动通知应用。watchdog正是这些能力的Python封装,事件延迟通常在毫秒级,且几乎不占计算资源。理解这一点,是写出高效监控程序的基础。
二、watchdog基本用法与事件类型
watchdog提供Observer和FileSystemEventHandler两个核心类。Observer负责在后台线程接收系统事件,EventHandler里的方法会在对应事件发生时被回调。主要事件包括on_created、on_modified、on_deleted和on_moved。
下面是一段最小可用监控代码,监控指定路径下所有变更并打印日志:
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
if not event.is_directory:
print("文件被修改:" + event.src_path)
def on_created(self, event):
if not event.is_directory:
print("新文件出现:" + event.src_path)
if __name__ == "__main__":
path = "./watch_dir"
observer = Observer()
observer.schedule(MyHandler(), path, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
这段代码中recursive=True表示递归监控子目录。需要注意,某些编辑器保存文件时会先删后建,可能同时触发deleted和created,而不是modified,业务层要做兼容。
三、变更检测的可靠性设计
仅依赖on_modified事件并不够,因为事件可能重复投递,且临时文件(如vim的.swp)也会干扰。推荐在事件触发后做一次轻量校验:先比对mtime,若变化再计算内容hash,确认真有改动才执行业务。
以下函数演示如何用hashlib做内容指纹,配合os.path.getmtime避免重复处理:
import hashlib
import os
_cache = {}
def file_changed(path):
try:
mtime = os.path.getmtime(path)
if path in _cache and _cache[path][0] == mtime:
return False
with open(path, "rb") as f:
data = f.read()
h = hashlib.md5(data).hexdigest()
if path in _cache and _cache[path][1] == h:
_cache[path] = (mtime, h)
return False
_cache[path] = (mtime, h)
return True
except FileNotFoundError:
if path in _cache:
del _cache[path]
return True
这种双重校验能挡掉绝大多数虚假变更。对于初次启动时的全量扫描,可以用os.scandir高效遍历,为每个文件建立初始指纹,避免启动后把历史文件当成新增。
四、防抖与忽略规则
像IDE保存操作可能连续抛出多个modified事件,直接触发同步会浪费IO。常见做法是收到事件后丢进队列,用0.5秒防抖窗口合并同路径操作。
同时应通过正则忽略临时文件与缓存目录,例如排除以~结尾或位于node_modules内的路径。watchdog的PatternMatchingEventHandler可声明ignore_patterns,减少无效回调:
from watchdog.events import PatternMatchingEventHandler
handler = PatternMatchingEventHandler(
patterns=["*.py"],
ignore_patterns=["*.swp", "*~", "__pycache__/*"],
ignore_directories=True
)
合理设置忽略规则不仅降低噪声,也避免监控程序自身产生的文件又触发自身,形成死循环。结合前面的指纹缓存,整套机制在长时间运行下依然稳定。
五、小结与落地建议
用watchdog做事件驱动监控,再辅以mtime加hash的变更确认和防抖队列,就能在Python中构建出资源占用低、准确性高的文件感知模块。对于跨机器同步,可将变更事件推到消息队列;对于本地热加载,直接在回调里重载模块即可。
实际部署时记得处理Observer异常退出,并给监控目录加上权限检查,防止因为无权访问子目录而导致整个观察器崩溃。把这些细节补齐,文件监控就不再是一个脆弱的脚本,而是可依赖的基础设施。