Python 捕获多进程 OR 多线程 error 的最佳实现是什么？

def test():
    a=[2]
    print(a[3])

def run():
    test()

def err(error):
    logger.exception(error)

if __name__ == '__main__':
    pool = Pool(2)
    t1 = pool.apply_async(run,error_callback=err)
    pool.close()
    pool.join()

日志记录 :

list index out of range

显然并没有拿到堆栈信息,在 callback 里再加上 try

def test():
    a=[2]
    print(a[3])

def run():
    test()

def err(error):
    #这边已经拿到 error 了但是还要 try，要不然日志只记录 list index out of range
    try:
        raise error
    except Exception as e:
        logger.exception(e)

if __name__ == '__main__':
    pool = Pool(2)
    t1 = pool.apply_async(run,error_callback=err)
    pool.close()
    pool.join()

日志输出 :

list index out of range
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/home/work_dir/test/timer.py", line 13, in run
    test()
  File "/home/work_dir/test/timer.py", line 10, in test
    print(a[3])
IndexError: list index out of range
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/work_dir/test/timer.py", line 17, in err
    raise error
IndexError: list index out of range

难道要在回调函数里再次捕获输出？

第 1 条附言 · 2020-01-07 17:32:29 +08:00

用了个简单的办法

from multiprocessing.pool import rebuild_exc
from multiprocessing import pool

class My_Except(pool.ExceptionWithTraceback):
    def __init__(self, exc, tb):
        logger.exception(tb)
        return super().__init__(exc,tb)

pool.ExceptionWithTraceback = My_Except

在包的__init__.py中打一个猴子补丁，这样就可以抓到堆栈，然后写日志，或刷新状态.

然后在代码中导入__init__.py,也不需要回调函数直接就拿到堆栈了

from multiprocessing import Pool
import py_mutilprocessing

def test():
    a=[2]
    print(a[3])

def run():
    test()

if __name__ == '__main__':
    pool = Pool(2)
    t1 = pool.apply_async(run,)
    pool.close()
    pool.join()

日志:

<traceback object at 0x7ff050695988>
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/home/tarena/work_dir/py_mutilprocessing/timer.py", line 9, in run
    test()
  File "/home/tarena/work_dir/py_mutilprocessing/timer.py", line 6, in test
    print(a[3])
IndexError: list index out of range

但是觉得这么写怪怪的~~~~

10 条回复 • 2020-05-19 16:29:07 +08:00

BingoXuan

2020-01-07 14:58:58 +08:00

多进程和多线程通讯的问题，内置有 queue 之类的数据类型可以使用。

wuwukai007

2020-01-07 15:01:43 +08:00

@BingoXuan 不是通信，是抓到子进程子线程错误，输出到日志或者写状态等，但是拿不到堆栈信息。我现在是再次 try 或者是用装饰器来做

zyqzyq08

2020-01-07 15:07:23 +08:00

可以考虑直接重写 sys.stderr

BingoXuan

2020-01-07 15:17:04 +08:00

你试图捕抓子进程或子线程信息本质就是通讯。为什么一定要主进程去干着活呢？你完全可以让子进程或子线程运行时候去捕捉啊。装饰器的其中一个广泛用途就是捕获并处理错误信息。如果子进程是一个 worker，那么子进程 worker 接受主进程的请求运行指定函数，记录运行结果或捕获运行错误，并通过 socket 或文件通知或保存信息。

Vegetable

2020-01-07 15:18:59 +08:00

traceback.format_exc()是你想要的?

oahebky

2020-01-07 15:19:26 +08:00

了解一下 traceback 库或许你就有答案了。

wuwukai007

2020-01-07 15:22:47 +08:00

@Vegetable traceback.format_exc() 是主进程中错误会抛出，用的进程池或线程池，除非用装饰器包起来，要不然子进程错误不会抛出来的，比如我现在用的回调，用这个直接输出 NoneType

todd7zhang

2020-01-07 17:42:25 +08:00

子进程的错误, 就让子进程抛呗

wuwukai007

2020-01-07 17:49:30 +08:00

@todd7zhang 子进程要捕获，之前用的装饰器把子进程整个包起来可以拿到 error，但是这样感觉很浪费，应为进程池本身就能拿到 error，这样多此一举了

hiyoi

2020-05-19 16:29:07 +08:00

最近刚好碰到这个情况，我是这么解决的，把要子线程和日志对象封装到一起，使用 logger.exception()来 traceback 错误

```
from concurrent.futures import ThreadPoolExecutor
import logging

class Main(object):
def __init__(self, logger=None):
self.pool = ThreadPoolExecutor
self.logger = logger or self._get_logger(__name__)

def _get_logger(self, name):
logger = logging.getLogger(name)
handler = logging.StreamHandler()
formatter = logging.Formatter(
"%(asctime)s - %(levelname)s - %(threadName)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
return logger

def do_something(self):
try:
a = 1 / 0
except Exception as e:
self.logger.exception(e)

def async_task(self):
with self.pool() as p:
for i in range(3):
p.submit(self.do_something)

if __name__ == '__main__':
m = Main()
m.async_task()
```

日志：
```
2020-05-19 16:23:38,378 - ERROR - ThreadPoolExecutor-0_0 - division by zero
Traceback (most recent call last):
File "thread.py", line 22, in do_something
a = 1 / 0
ZeroDivisionError: division by zero
2020-05-19 16:23:38,378 - ERROR - ThreadPoolExecutor-0_0 - division by zero
Traceback (most recent call last):
File "thread.py", line 22, in do_something
a = 1 / 0
ZeroDivisionError: division by zero
2020-05-19 16:23:38,378 - ERROR - ThreadPoolExecutor-0_2 - division by zero
Traceback (most recent call last):
File "thread.py", line 22, in do_something
a = 1 / 0
ZeroDivisionError: division by zero
```