Python Watchdog Observer监控目录时函数触发两次的解决方法

常见原因及对应解决方案 #

1. Flask调试模式导致多进程重复初始化Observer #

Flask开启debug=True时会启动两个进程（主进程+重载进程），如果Observer的初始化代码未做进程判断，会导致两个进程同时监听目录，一次文件变更触发两次处理函数。

解决方法：
将Observer的初始化和启动代码放到if __name__ == '__main__':代码块中，确保仅主进程启动监听：

if __name__ == '__main__':
    path = "<directory_path>"
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    print("observer started running...")
    try:
        app.run(debug=True)
    finally:
        observer.stop()
        observer.join()

2. 系统/编辑器产生的多阶段事件 #

多数系统或编辑器执行文件操作时会触发多个事件：比如Windows下新建文件会先创建临时文件再重命名，触发added+modified；编辑器保存时可能先删除旧文件再写入新文件，触发deleted+added。

解决方法：

• 过滤目录事件：忽略目录本身的变更事件，仅处理文件事件：

  def on_any_event(self, event):
      if event.is_directory:
          return
      # 后续处理逻辑
  

• 添加防抖机制：记录上次处理事件的时间，短时间内的重复事件直接跳过，避免重复执行：

  class MyHandler(FileSystemEventHandler):
      def __init__(self):
          self.last_trigger_time = 0
  
      def on_any_event(self, event):
          if event.is_directory:
              return
          # 1秒内重复事件跳过
          current_time = time.time()
          if current_time - self.last_trigger_time < 1:
              return
          self.last_trigger_time = current_time
          
          print(f"Event: {event.event_type} - {event.src_path}")
          try:
              texts = load_data()
              function_to_trigger(texts)
          except Exception as err:
              print(err)
  

• 精确过滤事件类型：如果仅关注特定操作（比如文件新增），可以只实现对应的事件方法，并添加文件存在校验：

  def on_created(self, event):
      if event.is_directory:
          return
      # 确保文件确实存在（避免临时文件触发）
      if os.path.exists(event.src_path):
          print(f"File created: {event.src_path}")
          # 处理逻辑
  

修改后的完整示例代码 #

from flask import Flask
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
import os

app = Flask(__name__)

class MyHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_trigger_time = 0

    def on_any_event(self, event):
        # 忽略目录事件
        if event.is_directory:
            return
        # 防抖处理
        current_time = time.time()
        if current_time - self.last_trigger_time < 1:
            return
        self.last_trigger_time = current_time
        
        print(f"Event: {event.event_type} - {event.src_path}")
        try:
            texts = load_data()
            function_to_trigger(texts)
        except Exception as err:
            print(err)

def load_data():
    # 自定义加载数据逻辑
    return []

def function_to_trigger(texts):
    # 自定义业务处理逻辑
    pass

if __name__ == '__main__':
    path = "<directory_path>"
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    print("observer started running...")
    try:
        app.run(debug=True)
    finally:
        observer.stop()
        observer.join()

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