Python多进程调用多类对象push方法遇序列化问题及并行日志需求

错误原因分析 #

你遇到的cannot pickle错误，是因为标准库multiprocessing.Pool使用Python默认的pickle模块序列化对象，而很多自定义类实例可能包含不可序列化的属性（比如打开的文件句柄、线程锁、第三方库对象），或者类本身未实现pickle接口，导致无法在进程间传递实例。

同时使用apply_async时脚本提前退出，是因为没有收集所有异步任务的结果并等待；而单独用job.wait()会串行等待任务，无法实现完成即日志的需求。

解决方案 #

一、解决Pickle序列化问题 #

方法1：改用pathos的ProcessingPool

pathos使用dill替代pickle，支持更多Python对象的序列化（包括大部分自定义类实例），无需大幅修改原有代码：

from pathos.multiprocessing import ProcessingPool as Pool

def push_wrapper(obj):
    obj.push()
    # 任务完成立即记录日志
    print(f"[{obj.__class__.__name__}] push方法执行完成")

class_objects = [class1_obj1, class2_obj2, class3_obj3]
max_thread_count = 3

with Pool(processes=max_thread_count) as pool:
    # imap_unordered会按任务完成顺序返回结果，配合循环实现实时日志
    for _ in pool.imap_unordered(push_wrapper, class_objects):
        pass
    pool.close()
    pool.join()

方法2：让自定义类支持Pickle

如果不想引入第三方库，可给类添加__getstate__和__setstate__方法，手动控制序列化逻辑，排除不可序列化属性：

class CustomClass:
    def __init__(self, data, non_picklable_obj):
        self.data = data
        self.non_picklable_obj = non_picklable_obj  # 比如文件句柄、网络连接

    def push(self):
        # 你的业务逻辑
        pass

    def __getstate__(self):
        # 只序列化可传递的属性
        state = self.__dict__.copy()
        del state['non_picklable_obj']
        return state

    def __setstate__(self, state):
        # 恢复属性，重新初始化不可序列化对象
        self.__dict__.update(state)
        self.non_picklable_obj = open('target_file.txt', 'r')  # 示例：重新打开文件

修改后，原multiprocessing.Pool代码即可正常运行。

方法3：在子进程中创建实例

如果实例初始化不依赖主进程资源，可传递类和初始化参数到子进程，在子进程内创建实例并调用方法：

from multiprocessing import Pool

def push_wrapper(args):
    cls, init_args = args
    obj = cls(*init_args)
    obj.push()
    print(f"[{cls.__name__}] push方法执行完成")

# 传递类+初始化参数，而非实例
class_args = [
    (Class1, [arg1, arg2]),
    (Class2, [arg3]),
    (Class3, [arg4, arg5])
]
max_thread_count = 3

with Pool(processes=max_thread_count) as pool:
    pool.map(push_wrapper, class_args)
    pool.close()
    pool.join()

二、实现任务完成立即日志+等待所有任务结束 #

使用apply_async时，通过收集所有异步任务对象、设置callback回调函数实现实时日志，最后用wait阻塞等待所有任务完成：

from multiprocessing import Pool, wait

def log_finish(result):
    # 回调函数：任务完成时立即执行
    print(f"任务完成，返回结果：{result}")

def push_wrapper(obj):
    return obj.push()

class_objects = [class1_obj1, class2_obj2, class3_obj3]
max_thread_count = 3

with Pool(processes=max_thread_count) as pool:
    jobs = []
    for obj in class_objects:
        # 提交异步任务并绑定回调
        job = pool.apply_async(push_wrapper, args=(obj,), callback=log_finish)
        jobs.append(job)
    
    # 等待所有任务完成
    wait(jobs)
    pool.close()
    pool.join()

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