Python多进程Event跨进程通信失效求助：文件夹监控与定时器联动异常

问题背景 #

用Python watchdog监控文件夹变化，需求是：文件夹有修改时重置30秒定时器，若超时无变化则终止所有进程。但观察者进程触发recount.set()后，定时器进程无法捕获事件，重置逻辑失效。

原始代码 #

import multiprocessing
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from threading import Timer

# 自定义可重置定时器类
class TimerReset(Timer):
    def reset(self):
        self.finished.set()
        self.finished.clear()

DIRECTORY_TO_WATCH = "./test_dir"

recount = multiprocessing.Event()

def run():
    event_handler = Handler()
    observer = Observer()
    observer.schedule(event_handler, DIRECTORY_TO_WATCH, recursive=True)
    observer.start()
    observer.join()

class Handler(FileSystemEventHandler):

    @staticmethod
    def on_any_event(event):
        global recount
        if event.is_directory:
            return None
        else :
            recount.set()

def timer(finish_state):
    global recount
    t = TimerReset(30.0)
    t.start()
    while not t.finished.is_set():
        if recount.is_set():
            print(recount.is_set())
        if recount.is_set() :
            t.reset()
            t.start
            recount.clear()
    finish_state.set()

if __name__ == '__main__':
    finish_state = multiprocessing.Event()
    p1 = multiprocessing.Process(target=run)
    p2 = multiprocessing.Process(target=timer, args=(finish_state,))
    p1.start()
    p2.start()
    finish_state.wait()
    p1.terminate()
    p2.terminate()

核心错误点 #

1. 全局Event无法跨进程共享
   多进程中，子进程会复制主进程的全局变量副本，而非共享同一个对象。观察者进程修改的是自己的recount副本，定时器进程的recount完全不受影响，自然看不到事件触发。

2. 定时器代码语法错误
   timer函数里的t.start缺少括号，写成t.start()才会实际启动定时器，否则重置后定时器不会重新运行。

3. 轮询方式低效且易漏事件
   用while循环轮询recount.is_set()会浪费CPU资源，且可能因循环间隙错过事件触发。

修复后的代码 #

import multiprocessing
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from threading import Timer

class TimerReset(Timer):
    def reset(self):
        # 终止当前定时器并重置状态
        self.finished.set()
        self.join()
        self.finished.clear()

DIRECTORY_TO_WATCH = "./test_dir"

def run(recount):
    event_handler = Handler(recount)
    observer = Observer()
    observer.schedule(event_handler, DIRECTORY_TO_WATCH, recursive=True)
    observer.start()
    observer.join()

class Handler(FileSystemEventHandler):
    def __init__(self, recount_event):
        self.recount = recount_event

    def on_any_event(self, event):
        if event.is_directory:
            return None
        # 触发跨进程事件
        self.recount.set()

def timer(finish_state, recount):
    # 定时器超时直接触发结束信号
    t = TimerReset(30.0, finish_state.set)
    t.start()
    
    while not finish_state.is_set():
        # 阻塞等待事件，超时1秒后检查是否结束
        if recount.wait(timeout=1):
            print("检测到文件夹变化，重置定时器")
            t.reset()
            t.start()
            recount.clear()
    
    print("定时器超时，终止进程")

if __name__ == '__main__':
    finish_state = multiprocessing.Event()
    # 主进程创建共享Event，传递给子进程
    recount = multiprocessing.Event()
    
    p1 = multiprocessing.Process(target=run, args=(recount,))
    p2 = multiprocessing.Process(target=timer, args=(finish_state, recount))
    
    p1.start()
    p2.start()
    
    finish_state.wait()
    p1.terminate()
    p2.terminate()

关键修复说明 #

• 共享Event传递：主进程创建recount后作为参数传给两个子进程，确保两者操作的是同一个跨进程同步对象。
• 移除全局变量依赖：Handler类通过构造函数接收recount，避免全局变量带来的跨进程问题。
• 阻塞等待替代轮询：用recount.wait(timeout=1)减少CPU占用，同时保证及时响应事件。
• 定时器回调优化：将finish_state.set()设为定时器的回调，超时自动触发结束逻辑，代码更简洁。

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