在Python的调用者线程中捕获线程的异常

Question

我对Python和多线程编程很新.基本上,我有一个脚本,将文件复制到另一个位置.我希望将它放在另一个线程中,以便我可以输出....以指示脚本仍在运行.

我遇到的问题是,如果文件无法复制,它将引发异常.如果在主线程中运行,这是可以的; 但是,具有以下代码不起作用:

try:
    threadClass = TheThread(param1, param2, etc.)
    threadClass.start()   ##### **Exception takes place here**
except:
    print "Caught an exception"

在线程类本身,我试图重新抛出异常,但它不起作用.我看到这里的人问类似的问题,但他们似乎都在做一些比我想做的更具体的事情(而且我不太了解提供的解决方案).我见过人们提到使用它sys.exc_info(),但我不知道在哪里或如何使用它.

非常感谢所有帮助!

编辑:线程类的代码如下:

class TheThread(threading.Thread):
    def __init__(self, sourceFolder, destFolder):
        threading.Thread.__init__(self)
        self.sourceFolder = sourceFolder
        self.destFolder = destFolder

    def run(self):
        try:
           shul.copytree(self.sourceFolder, self.destFolder)
        except:
           raise

Answer 1

问题是thread_obj.start()立即返回.您生成的子线程在其自己的上下文中执行,具有自己的堆栈.发生在那里的任何异常都存在于子线程的上下文中,并且它位于自己的堆栈中.我现在可以想到将这些信息传递给父线程的一种方法是使用某种消息传递,所以你可能会考虑这一点.

试试这个尺码:

import sys
import threading
import Queue


class ExcThread(threading.Thread):

    def __init__(self, bucket):
        threading.Thread.__init__(self)
        self.bucket = bucket

    def run(self):
        try:
            raise Exception('An error occured here.')
        except Exception:
            self.bucket.put(sys.exc_info())


def main():
    bucket = Queue.Queue()
    thread_obj = ExcThread(bucket)
    thread_obj.start()

    while True:
        try:
            exc = bucket.get(block=False)
        except Queue.Empty:
            pass
        else:
            exc_type, exc_obj, exc_trace = exc
            # deal with the exception
            print exc_type, exc_obj
            print exc_trace

        thread_obj.join(0.1)
        if thread_obj.isAlive():
            continue
        else:
            break


if __name__ == '__main__':
    main()

Answer 2

该concurrent.futures模块使在单独的线程(或进程)中工作变得简单,并处理任何产生的异常:

import concurrent.futures
import shutil

def copytree_with_dots(src_path, dst_path):
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        # Execute the copy on a separate thread,
        # creating a future object to track progress.
        future = executor.submit(shutil.copytree, src_path, dst_path)

        while future.running():
            # Print pretty dots here.
            pass

        # Return the value returned by shutil.copytree(), None.
        # Raise any exceptions raised during the copy process.
        return future.result()

concurrent.futures包含在Python 3.2中,可用作早期版本的反向移植futures模块.

Answer 3

虽然不可能直接捕获在不同线程中抛出的异常,但这里有一个代码可以非常透明地获得非常接近此功能的东西.您的子线程必须是子ExThread类而不是子类,threading.Thread并且父线程必须调用该child_thread.join_with_exception()方法,而不是child_thread.join()在等待线程完成其作业时.

这个实现的技术细节:当子线程抛出异常时,它通过a传递给Queue父线程并在父线程中再次抛出.请注意,这种方法没有繁忙的等待.

#!/usr/bin/env python

import sys
import threading
import Queue

class ExThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__status_queue = Queue.Queue()

    def run_with_exception(self):
        """This method should be overriden."""
        raise NotImplementedError

    def run(self):
        """This method should NOT be overriden."""
        try:
            self.run_with_exception()
        except BaseException:
            self.__status_queue.put(sys.exc_info())
        self.__status_queue.put(None)

    def wait_for_exc_info(self):
        return self.__status_queue.get()

    def join_with_exception(self):
        ex_info = self.wait_for_exc_info()
        if ex_info is None:
            return
        else:
            raise ex_info[1]

class MyException(Exception):
    pass

class MyThread(ExThread):
    def __init__(self):
        ExThread.__init__(self)

    def run_with_exception(self):
        thread_name = threading.current_thread().name
        raise MyException("An error in thread '{}'.".format(thread_name))

def main():
    t = MyThread()
    t.start()
    try:
        t.join_with_exception()
    except MyException as ex:
        thread_name = threading.current_thread().name
        print "Caught a MyException in thread '{}': {}".format(thread_name, ex)

if __name__ == '__main__':
    main()

Answer 4

这个问题有很多非常复杂的答案.我是否过度简化了这一点,因为这对我来说似乎已经足够了.

from threading import Thread

class PropagatingThread(Thread):
    def run(self):
        self.exc = None
        try:
            if hasattr(self, '_Thread__target'):
                # Thread uses name mangling prior to Python 3.
                self.ret = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
            else:
                self.ret = self._target(*self._args, **self._kwargs)
        except BaseException as e:
            self.exc = e

    def join(self):
        super(PropagatingThread, self).join()
        if self.exc:
            raise self.exc
        return self.ret

如果你确定你只会在一个或另一个版本的Python上运行,那么你可以将run()方法减少到仅受损的版本(如果你只在3之前的Python版本上运行),或者只是干净的版本(如果你只在从3开始的Python版本上运行).

用法示例:

def f(*args, **kwargs)
    print(args)
    print(kwargs)
    raise Exception('I suck')

t = PropagatingThread(target=f, args=(5,), kwargs={'hello':'world'})
t.start()
t.join()

当你加入时,你会看到另一个线程上引发的异常.

Answer 5

如果线程中发生异常,最好的方法是在调用程序线程中重新引发它join.您可以使用该sys.exc_info()函数获取有关当前正在处理的异常的信息.这个信息可以简单地存储为线程对象的属性,直到join被调用,此时可以重新引发它.

请注意,Queue.Queue在这种简单的情况下,(在其他答案中建议)不需要线程,其中线程抛出最多1个异常并在抛出异常后立即完成.我们通过简单地等待线程完成来避免竞争条件.

例如,扩展ExcThread(下面),覆盖excRun(而不是run).

Python 2.x:

import threading

class ExcThread(threading.Thread):
  def excRun(self):
    pass

  def run(self):
    self.exc = None
    try:
      # Possibly throws an exception
      self.excRun()
    except:
      import sys
      self.exc = sys.exc_info()
      # Save details of the exception thrown but don't rethrow,
      # just complete the function

  def join(self):
    threading.Thread.join(self)
    if self.exc:
      msg = "Thread '%s' threw an exception: %s" % (self.getName(), self.exc[1])
      new_exc = Exception(msg)
      raise new_exc.__class__, new_exc, self.exc[2]

Python 3.x:

3参数形式raise在Python 3中消失了,所以将最后一行更改为:

raise new_exc.with_traceback(self.exc[2])

Answer 6

我正在做的是，简单地重写线程的 join 和 run 方法：

class RaisingThread(threading.Thread):
  def run(self):
    self._exc = None
    try:
      super().run()
    except Exception as e:
      self._exc = e

  def join(self, timeout=None):
    super().join(timeout=timeout)
    if self._exc:
      raise self._exc

使用如下：

def foo():
  time.sleep(2)
  print('hi, from foo!')
  raise Exception('exception from foo')    

t = RaisingThread(target=foo)
t.start()
try:
  t.join()
except Exception as e:
  print(e)

结果：

hi, from foo!
exception from foo!

Answer 7

concurrent.futures.as_completed

https://docs.python.org/3.7/library/concurrent.futures.html#concurrent.futures.as_completed

以下解决方案：

• 调用异常时立即返回主线程
• 不需要额外的用户定义类，因为它不需要：
  • 明确的 Queue
  • 在您的工作线程周围添加一个除了 else

来源：

#!/usr/bin/env python3

import concurrent.futures
import time

def func_that_raises(do_raise):
    for i in range(3):
        print(i)
        time.sleep(0.1)
    if do_raise:
        raise Exception()
    for i in range(3):
        print(i)
        time.sleep(0.1)
    
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    futures = []
    futures.append(executor.submit(func_that_raises, False))
    futures.append(executor.submit(func_that_raises, True))
    for future in concurrent.futures.as_completed(futures):
        print(repr(future.exception()))

可能的输出：

0
0
1
1
2
2
0
Exception()
1
2
None

不幸的是，不可能杀死期货来取消其他期货，因为其中一个失败了：

• concurrent.futures; Python：concurrent.futures 如何使它可取消？
• threading:有没有办法杀死一个线程？
• C pthreads：杀死Pthread库中的线程

如果您执行以下操作：

for future in concurrent.futures.as_completed(futures):
    if future.exception() is not None:
        raise future.exception()

然后with捕获它，并在继续之前等待第二个线程完成。以下行为类似：

for future in concurrent.futures.as_completed(futures):
    future.result()

因为future.result()如果发生异常，则会重新引发异常。

如果您想退出整个 Python 进程，您可以使用os._exit(0)，但这可能意味着您需要重构。

具有完美异常语义的自定义类

我最终为自己编写了完美的界面：限制一次运行的最大线程数的正确方法？“带有错误处理的队列示例”部分。该课程旨在既方便又让您完全控制提交和结果/错误处理。

在 Python 3.6.7、Ubuntu 18.04 上测试。

Answer 8

在 Python 3.8 中，我们可以使用threading.excepthook来钩住所有子线程中未捕获的异常！例如，

threading.excepthook = thread_exception_handler

推荐人：https : //stackoverflow.com/a/60002752/5093308

Answer 9

这是一个令人讨厌的小问题，我想提出我的解决方案。我发现的其他一些解决方案（例如 async.io）看起来很有希望，但也呈现出一些黑匣子。队列/事件循环方法将您与某个实现联系起来。但是并发期货源代码只有 1000 行左右，易于理解。它让我可以轻松解决我的问题：无需太多设置即可创建临时工作线程，并且能够在主线程中捕获异常。

我的解决方案使用并发期货 API 和线程 API。它允许您创建一个工人，为您提供线程和未来。这样就可以加入线程等待结果了：

worker = Worker(test)
thread = worker.start()
thread.join()
print(worker.future.result())

...或者您可以让工作人员在完成后发送回调：

worker = Worker(test)
thread = worker.start(lambda x: print('callback', x))

...或者你可以循环直到事件完成：

worker = Worker(test)
thread = worker.start()

while True:
    print("waiting")
    if worker.future.done():
        exc = worker.future.exception()
        print('exception?', exc)
        result = worker.future.result()
        print('result', result)           
        break
    time.sleep(0.25)

这是代码：

from concurrent.futures import Future
import threading
import time

class Worker(object):
    def __init__(self, fn, args=()):
        self.future = Future()
        self._fn = fn
        self._args = args

    def start(self, cb=None):
        self._cb = cb
        self.future.set_running_or_notify_cancel()
        thread = threading.Thread(target=self.run, args=())
        thread.daemon = True #this will continue thread execution after the main thread runs out of code - you can still ctrl + c or kill the process
        thread.start()
        return thread

    def run(self):
        try:
            self.future.set_result(self._fn(*self._args))
        except BaseException as e:
            self.future.set_exception(e)

        if(self._cb):
            self._cb(self.future.result())

...和测试功能：

def test(*args):
    print('args are', args)
    time.sleep(2)
    raise Exception('foo')

Answer 10

我知道我来晚了一点，但我遇到了一个非常相似的问题，但它包括使用 tkinter 作为 GUI，并且主循环使得无法使用任何依赖于 .join() 的解决方案。因此，我调整了原始问题的编辑中给出的解决方案，但使其更加通用，以便其他人更容易理解。

这是正在运行的新线程类：

import threading
import traceback
import logging


class ExceptionThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except Exception:
            logging.error(traceback.format_exc())


def test_function_1(input):
    raise IndexError(input)


if __name__ == "__main__":
    input = 'useful'

    t1 = ExceptionThread(target=test_function_1, args=[input])
    t1.start()

当然，您始终可以让它通过日志记录以其他方式处理异常，例如将其打印出来，或将其输出到控制台。

这允许您像使用 Thread 类一样使用 ExceptionThread 类，无需任何特殊修改。