如何捕获使用 multiprocessing.Process() (python) 执行的函数引发的异常

Question

如何从使用multiprocessing.Process() 执行的进程中捕获异常？

考虑以下使用mulitprocessing.Process()在子进程内部执行简单failFunction()（立即引发运行时错误）的python脚本

#!/usr/bin/env python3
import multiprocessing, time

# this function will be executed in a child process asynchronously
def failFunction():
   raise RuntimeError('trust fall, catch me!')

# execute the helloWorld() function in a child process in the background
process = multiprocessing.Process(
 target = failFunction,
)
process.start()

# <this is where async stuff would happen>
time.sleep(1)

# try (and fail) to catch the exception
try:
    process.join()
except Exception as e:
    print( "This won't catch the exception" )

从以下执行中可以看出，尝试包装 .join() 确实不实际上捕获了异常

user@host:~$ python3 example.py 
Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/usr/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "example4.py", line 6, in failFunction
    raise RuntimeError('trust fall, catch me!')
RuntimeError: trust fall, catch me!
user@host:~$ 

如何更新上述脚本以实际捕获使用multiprocessing.Process() 在子进程内部执行的函数的异常？

Answer 1

这可以通过使用try..except 语句重载multiprocessing.Proccess() 类中的run() 方法并设置Pipe() 来获取并存储来自子进程的任何引发的异常到命名的实例字段中来实现exception:

#!/usr/bin/env python3
import multiprocessing, traceback, time

class Process(multiprocessing.Process):

    def __init__(self, *args, **kwargs):
        multiprocessing.Process.__init__(self, *args, **kwargs)
        self._pconn, self._cconn = multiprocessing.Pipe()
        self._exception = None

    def run(self):
        try:
            multiprocessing.Process.run(self)
            self._cconn.send(None)
        except Exception as e:
            tb = traceback.format_exc()
            self._cconn.send((e, tb))
            #raise e  # You can still rise this exception if you need to

    @property
    def exception(self):
        if self._pconn.poll():
            self._exception = self._pconn.recv()
        return self._exception


# this function will be executed in a child process asynchronously
def failFunction():
   raise RuntimeError('trust fall, catch me!')

# execute the helloWorld() function in a child process in the background
process = Process(
 target = failFunction,
)
process.start()

# <this is where async stuff would happen>
time.sleep(1)

# catch the child process' exception
try:
    process.join()
    if process.exception:
        raise process.exception
except Exception as e:
    print( "Exception caught!" )

示例执行：

user@host:~$ python3 example.py 
Exception caught!
user@host:~$ 

从这个答案中得到的解决方案：

• https://stackoverflow.com/a/33599967/1174102

Answer 2

此解决方案不需要目标函数必须捕获自己的异常。

这看起来有点矫枉过正，但您可以使用模块concurrent.futures 中的类ProcessPoolExecutor 创建一个大小为1 的进程池，这就是您所需的全部。当您向执行程序提交“作业”时，会创建一个 Future 实例，表示流程的执行状态。当您在 Future 实例上调用 result() 时，您会阻塞直到进程终止并返回结果（目标函数返回）。如果目标函数抛出异常，可以在调用result()时捕获：

import concurrent.futures

def failFunction():
   raise RuntimeError('trust fall, catch me!')

def main():
    with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
        future = executor.submit(failFunction)
        try:
            result = future.result()
        except Exception as e:
            print('exception = ', e)
        else:
            print('result = ', result)

if __name__ == '__main__':
    main()

打印：

exception =  trust fall, catch me!

使用进程池的好处是，如果您有需要在子进程中调用的其他功能，那么您已经创建了一个现成的进程。