Python使用joblib并行时如何实现异常文件写入仅单次执行

使用joblib并行运行函数时，函数内包含捕获CustomError异常并写入文件的逻辑。希望异常发生时，文件写入步骤仅在单个CPU上执行一次，而非所有CPU同时执行，之后继续并行处理剩余代码，如何实现这种多线程/多进程与单线程/单进程切换的需求？

代码示例：

from joblib import Parallel, delayed

def process(x):
    try:
        ...some code
    except CustomError as e:
        # 需要这部分代码仅在单个CPU上执行
        with open("foo.txt") as f:
            f.write(e)

    ...other code

if __name__ == '__main__':
    Parallel(n_jobs=-1)(delayed(process)(i) for i in range(100))

核心思路是通过进程锁控制文件写入操作的唯一性，确保同一时间只有一个进程能执行写入逻辑，不影响其他进程的并行处理。具体实现步骤如下：

• 初始化一个全局进程锁对象，让所有并行子进程共享该锁
• 捕获到CustomError时，通过锁的上下文管理器获取锁，执行写入后自动释放锁
• 注意joblib默认采用多进程模式，需使用进程间有效的锁（而非线程锁）

修改后的完整代码：

from joblib import Parallel, delayed
from multiprocessing import Lock
import traceback

# 自定义异常类示例
class CustomError(Exception):
    pass

# 全局进程锁，所有子进程共享
write_lock = Lock()

def process(x):
    try:
        # 模拟可能抛出CustomError的业务代码
        if x % 10 == 0:
            raise CustomError(f"任务{x}执行出错")
        # 正常业务处理逻辑
        result = x * 2
    except CustomError as e:
        # 加锁确保单进程执行文件写入
        with write_lock:
            # 用追加模式打开文件，避免覆盖已有内容
            with open("foo.txt", "a", encoding="utf-8") as f:
                f.write(f"{str(e)}\n")
                # 可选：写入异常栈信息便于排查
                f.write(traceback.format_exc() + "\n")
        # 异常处理后继续执行剩余代码
        result = f"已处理任务{x}的异常"
    # 剩余通用处理代码
    return result

if __name__ == '__main__':
    # 启动并行任务
    results = Parallel(n_jobs=-1)(delayed(process)(i) for i in range(100))
    print("所有任务执行完成")

关键细节说明：

• 使用with write_lock:上下文管理器，会自动完成锁的获取与释放，避免手动操作遗漏释放导致死锁
• 文件打开采用"a"追加模式，防止多进程写入时覆盖内容，同时确保写入内容为字符串类型
• 异常处理完成后，函数可继续执行剩余逻辑，不中断其他并行任务的运行

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