使用ProcessPoolExecutor异步批量转docx为PDF未并行执行问题求助

问题现象 #

使用watchdog监控目录，提交docx转PDF任务到ProcessPoolExecutor后，任务卡住不执行，直到按下回车键才批量处理；尝试ThreadPoolExecutor后，ProcessPoolExecutor表现相对最优，但仍无法实现预期的并行转换效果。

核心排查方向 #

1. 主线程阻塞导致任务调度挂起 #

watchdog的事件循环默认是单线程运行，如果主线程被observer.join()这类操作阻塞，会直接导致ProcessPoolExecutor的任务调度无法正常执行。

• 必须将watchdog的监控逻辑放到独立线程中启动，避免占用主线程的调度资源
• 禁止在事件回调函数中执行同步IO或耗时操作，回调只做任务提交动作

2. ProcessPoolExecutor初始化时机错误 #

如果在watchdog事件回调中动态创建Executor实例，首次提交任务时的进程创建过程会被主线程阻塞打断。

• 在程序启动阶段就初始化ProcessPoolExecutor实例，提前完成子进程创建
• 确保任务函数是可序列化的：ProcessPool依赖pickle传递任务，不能将watchdog的事件对象直接传入任务，要提取出文件路径等基础数据

3. 标准IO缓冲引发的进程挂起 #

子进程的stdout/stderr如果没有及时flush，会因为缓冲满而挂起，直到主线程读取输出。

• 在任务函数的print操作中添加flush=True参数
• 或者将子进程的IO重定向到文件，避免依赖主线程的控制台输出

4. 任务依赖共享资源导致串行执行 #

如果多个任务复用同一个docx转PDF工具实例（比如win32com.client或libreoffice进程），会因为资源锁导致任务串行。

• 每个任务单独初始化转换工具实例，保证进程间资源隔离
• 避免在任务间共享不可进程安全的对象

5. 典型错误代码与修正 #

错误写法（主线程被watchdog阻塞）：

from watchdog.observers import Observer
from concurrent.futures import ProcessPoolExecutor

def convert_task(file_path):
    # 转PDF逻辑
    pass

def on_created(event):
    executor.submit(convert_task, event.src_path)

if __name__ == "__main__":
    executor = ProcessPoolExecutor(max_workers=4)
    observer = Observer()
    observer.schedule(MyHandler(), path="./watch_dir")
    observer.start()
    observer.join()  # 阻塞主线程，Executor无法调度任务

修正写法（独立线程启动watchdog）：

import threading
from watchdog.observers import Observer
from concurrent.futures import ProcessPoolExecutor

def convert_task(file_path):
    # 转PDF逻辑
    pass

def on_created(event):
    executor.submit(convert_task, event.src_path)

def start_watchdog():
    observer = Observer()
    observer.schedule(MyHandler(), path="./watch_dir")
    observer.start()
    observer.join()

if __name__ == "__main__":
    executor = ProcessPoolExecutor(max_workers=4)
    # 独立线程启动监控，释放主线程用于任务调度
    threading.Thread(target=start_watchdog, daemon=True).start()
    # 保持主线程运行（服务端场景可替换为其他阻塞逻辑）
    while True:
        try:
            input()
        except KeyboardInterrupt:
            executor.shutdown()
            break

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