Python多进程Queue传递本地函数/Lambda的Pickle错误解决

问题原因 #

Python的pickle模块无法序列化模块级别以外的函数（包括嵌套函数、lambda、本地定义函数），因为它依赖函数的完全限定命名空间路径完成反序列化，而本地函数仅存在于父函数的局部作用域中，没有全局可访问的路径，因此抛出AttributeError: Can't get local object 'task.<locals>.function'错误。

解决方案 #

使用cloudpickle（或dill）替代pickle，这两个库专门扩展了序列化能力，支持本地函数、lambda、嵌套函数等场景的序列化/反序列化。

步骤1：安装依赖库 #

pip install cloudpickle

步骤2：修改FilterBroker代码 #

替换原有的pickle调用为cloudpickle，调整消息队列的序列化逻辑：

import cloudpickle
from multiprocessing import Queue

class Filter:
    def __init__(self, func):
        self.func = func

class FilterBroker:
    def __init__(self):
        self.queue = Queue()
    
    def send_filter(self, filter_obj):
        # 用cloudpickle序列化Filter对象
        serialized = cloudpickle.dumps(filter_obj)
        self.queue.put(serialized)
    
    def receive_filter(self):
        serialized = self.queue.get()
        # 用cloudpickle反序列化
        return cloudpickle.loads(serialized)

def task():
    # 本地定义函数
    def local_func(x):
        return x * 2
    # Lambda函数
    lambda_func = lambda x: x + 3
    
    broker = FilterBroker()
    
    # 测试本地函数
    broker.send_filter(Filter(local_func))
    filter_obj = broker.receive_filter()
    print(f"本地函数执行结果: {filter_obj.func(5)}")
    
    # 测试Lambda函数
    broker.send_filter(Filter(lambda_func))
    filter_obj2 = broker.receive_filter()
    print(f"Lambda执行结果: {filter_obj2.func(5)}")

if __name__ == "__main__":
    task()

原理说明 #

cloudpickle不会依赖函数的命名空间路径，而是直接序列化函数的字节码、闭包变量等完整上下文信息，在反序列化时重建函数的执行环境，因此可以完美支持本地函数、lambda这类无法被pickle处理的可调用对象。

备选方案（无第三方库） #

如果无法引入第三方库，可以手动提取函数的源码文本，在接收端通过exec重建函数，但这种方法需要处理闭包变量、依赖导入等问题，复杂度高且稳定性差，仅建议在受限环境下使用。

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