python - 如何捕获线程中发生的异常？

Question

这个问题已经有答案了:

Catch a thread's exception in the caller thread? (20 个回答)

已关闭 3 年前。

注意:有 one exact answer对于我的重复问题(另请参阅下面我的答案以了解修改后的代码)。谢谢@quamrana 的指点

上下文:我在一个类中有一个方法列表，这些方法都是在线程中启动的。其中一些方法预计会引发异常，并且这些异常必须在主程序中处理(=而不是在方法本身中)。

问题:未捕获异常，并且解释(成功/失败)错误，因为所有线程都“成功”。

我认为可行的:a try/except其中线程实际上是 start() 。

请注意，在回溯中，两个答案都是 (...) was successful就好像try只处理启动线程( .start() )的事实，而不处理线程本身发生的事情。

import threading

class Checks:

    @staticmethod
    def isok():
        print("OK")

    @staticmethod
    def isko():
        raise Exception("KO")

# db will keep a map of method names in Check with the actual (executable) method
db = {}

# getting all the methods from Checks, without the built_in ones
for check in [k for k in dir(Checks) if not k.startswith('_')]:
    # create a thread for the method
    db[check] = threading.Thread(target=getattr(Checks, check))
    try:
        # start the thread
        db[check].start()
    except Exception:
        print(f"{check} raised an exception")
    else:
        print(f"{check} was successful")

# wait for all the threads to be finished
for thread in db.keys():
    db[thread].join()

# all the threads are finished at that point
print("all threads are done")

输出:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "C:\Users\yop\AppData\Local\Programs\Python\Python37-32\lib\threading.py", line 926, in _bootstrap_inner
    self.run()
  File "C:\Users\yop\AppData\Local\Programs\Python\Python37-32\lib\threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "C:/Users/yop/.PyCharm2019.2/config/scratches/scratch_6.py", line 11, in isko
    raise Exception("KO")
Exception: KO

isko was successful
OK
isok was successful
all threads are done

(由于线程的原因，回溯将与程序的实际输出混合，但顺序始终相同)

编辑:根据评论，我想再次强调异常将在方法中发生，但必须在主程序中捕获(=不在方法本身中处理)。

在非线程方法中，这很简单:try/exception与上面类似的代码中的子句会在它们冒泡时捕获它们。

Answer 1

创建一个队列来捕获异常。

错误=queue.Queue()

def threaded_func():
  try:
    # perform some task
  except:
    errors.put(  
      # push into queue as required
    )

def main():
  while True and threads_running:
   if errors.__len__():
     error_in_main = errors.pop()
     # handle the error as required.

使用这种方式，您几乎可以立即捕获主线程中的错误并根据需要执行操作。