Python并发编程：为什么传入进程池的目标函数不执行，也没有报错？

转载：Python并发编程：为什么传入进程池的目标函数不执行，也没有报错？ - 知乎 (zhihu.com)

python初学者使用进程池时，很容易掉坑里！

python并发编程中，这个问题是新手经常容易犯的错，十个人，大概有九个都会掉入其中。

借此机会，对该问题的前因后果做个记录，分享于此！

一、错误代码复现

我把错误代码抽象出来，大概就是下面这个样子：

"""
进程池
"""
from multiprocessing import Pool


class Func(object):
    def __init__(self):
        # 利用匿名函数模拟一个不可序列化象
        # 更常见的错误写法是，在这里初始化一个数据库的长链接
        self.num = lambda: None

    def work(self, num=None):
        self.num = num
        return self.num

    @staticmethod
    def call_back(res):
        print(f'Hello,World! {res}')


if __name__ == '__main__':
    func = Func()
    pool = Pool(3)
    for k in range(3):
        pool.apply_async(
            func.work,
            args=(k,),
            callback=func.call_back,
        )
    pool.close()
    pool.join()

按照程序设计的初衷，是想输出三遍“Hello,World!”

可是，运行代码后，没有输出任何东西，也无报错！

F:\python3\env\Scripts\python.exe F:/python_code/tips/demo.py

Process finished with exit code 0

二、现象分析

上面的代码，运行后输出了“Process finished with exit code 0”。

返回“code 0 ”，给人的第一感觉就是程序正常结束，没有报错。

其实不然，程序是有报错的。只是这个错误默认被pass掉了。

上面代码的关键函数是：apply_async()

进程池中，这是大家使用最多的一个函数。

在上面的代码中，为 apply_async() 函数指定了一个可执行的函数对象、函数对象所需参数，以及一个处理结果的回调函数。

pool.apply_async(
            func.work,
            args=(k,),
            callback=func.call_back,
        )

这也是我们使用 apply_async() 最常见写法。这样写存在一个隐患，很容易出现指定函数不执行，也无报错的现象。

对于编程新手，这是非常难排查的bug！

查看 python 官方文档，我们可以看到 apply_async() 的定义如下：

apply_async(func[, args[, kwds[, callback[, error_callback]]]])

重点是最后的 “error_callback”参数。

如果指定了 callback , 它必须是一个接受单个参数的可调用对象。当执行成功时， callback 会被用于处理执行后的返回结果，否则，调用 error_callback 。
如果指定了 error_callback , 它必须是一个接受单个参数的可调用对象。当目标函数执行失败时， 会将抛出的异常对象作为参数传递给 error_callback 执行。

比较坑的是，error_callback 是一个可选参数。

如果不传入error_callback，那么执行失败时， 抛出的异常就会被pass掉，让人误以为没有报错！

三、利用 error_callback 查看报错信息

我们以最开始的代码为基础，来添加一个 error_callback 函数。

现在，整体代码如下：

"""
进程池
"""
from multiprocessing import Pool


class Func(object):
    def __init__(self):
        # 利用匿名函数模拟一个不可序列化象
        # 更常见的错误写法是，在这里初始化一个数据库的长链接
        self.num = lambda: None

    def work(self, num=None):
        self.num = num
        return self.num

    @staticmethod
    def call_back(res):
        print(f'Hello,World! {res}')

    @staticmethod
    def err_call_back(err):
        print(f'出错啦~ error：{str(err)}')


if __name__ == '__main__':
    func = Func()
    pool = Pool(3)
    for k in range(3):
        pool.apply_async(
            func.work,
            args=(k,),
            callback=func.call_back,
            error_callback=func.err_call_back
        )
    pool.close()
    pool.join()

再次运行程序，可以看到错误输出信息：

F:\python3\env\Scripts\python.exe F:/python_code/tips/demo.py
出错啦~ error：Can't pickle local object 'Func.__init__.<locals>.<lambda>'
出错啦~ error：Can't pickle local object 'Func.__init__.<locals>.<lambda>'
出错啦~ error：Can't pickle local object 'Func.__init__.<locals>.<lambda>'

Process finished with exit code 0

以上结果证明：最开始的代码是有报错的！

建议使用 apply_async() 时，都传入一个 error_callback！

四、错误分析

通过 error_callback 的输出，我们可以看到，程序报错信息是：

Can't pickle local object 'Func.__init__.<locals>.<lambda>'

意思是：不能序列化 Func 类中 初始化函数_init_() 中的局部对象‘<locals>.<lambda>’。

值得一提的是：这个问题，并不是由进程池引起的。（只是在进程池中比较难排查）

引起该问题的根本原因是“多进程”。

注意区分进程池与多进程。

在 multiprocessing 模块中，Process.__init__() 的所有参数都必须可序列化。

当继承 Process 时，也需要保证当调用 Process.start 方法时，实例可以被序列化。

在多进程中，该问题很容易被发现。

例如，以下代码：

"""
多进程
"""
import multiprocessing


class Func(object):
    def __init__(self):
        # 利用匿名函数模拟一个不可序列化象
        # 更常见的错误写法是，在这里初始化一个数据库的长链接
        self.num = lambda: None

    def work(self, num=None):
        self.num = num
        return self.num


if __name__ == '__main__':
    func = Func()
    pro = multiprocessing.Process(
        target=func.work,
        args=('123',)
    )
    pro.start()
    pro.join()

运行程序，可以直接看到报错信息：

AttributeError: Can't pickle local object 'Func.__init__.<locals>.<lambda>'

另外，值得注意的是：该问题在 windows 和 linux 平台上的表现，各不相同！

• linux下，通过 os.fork() 方式来创建子进程，不存在该问题。
• windows下，通过 spawn 方式来创建子进程，必须保证 Process.__init__() 的所有参数都必须可序列化。