我对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
使用裸例外并不是一个好的实践,因为您通常会获得比您讨价还价时更多的东西。
我建议修改except以只捕获您想要处理的异常。我不认为引发它有预期的效果,因为当你在外层try中实例化TheThread时,如果它引发一个异常,赋值永远不会发生。
相反,你可能只想提醒它,然后继续前进,比如:
def run(self):
try:
shul.copytree(self.sourceFolder, self.destFolder)
except OSError, err:
print err
然后,当异常被捕获时,您可以在那里处理它。然后,当外部try从TheThread捕获异常时,您知道它不是您已经处理过的异常,并将帮助您隔离流程流。
Pygolang提供同步功能。工作组,特别是将异常从派生的工作线程传播到主线程。例如:
#!/usr/bin/env python
"""This program demostrates how with sync.WorkGroup an exception raised in
spawned thread is propagated into main thread which spawned the worker."""
from __future__ import print_function
from golang import sync, context
def T1(ctx, *argv):
print('T1: run ... %r' % (argv,))
raise RuntimeError('T1: problem')
def T2(ctx):
print('T2: ran ok')
def main():
wg = sync.WorkGroup(context.background())
wg.go(T1, [1,2,3])
wg.go(T2)
try:
wg.wait()
except Exception as e:
print('Tmain: caught exception: %r\n' %e)
# reraising to see full traceback
raise
if __name__ == '__main__':
main()
在运行时给出以下结果:
T1: run ... ([1, 2, 3],)
T2: ran ok
Tmain: caught exception: RuntimeError('T1: problem',)
Traceback (most recent call last):
File "./x.py", line 28, in <module>
main()
File "./x.py", line 21, in main
wg.wait()
File "golang/_sync.pyx", line 198, in golang._sync.PyWorkGroup.wait
pyerr_reraise(pyerr)
File "golang/_sync.pyx", line 178, in golang._sync.PyWorkGroup.go.pyrunf
f(pywg._pyctx, *argv, **kw)
File "./x.py", line 10, in T1
raise RuntimeError('T1: problem')
RuntimeError: T1: problem
问题的原始代码将是:
wg = sync.WorkGroup(context.background())
def _(ctx):
shul.copytree(sourceFolder, destFolder)
wg.go(_)
# waits for spawned worker to complete and, on error, reraises
# its exception on the main thread.
wg.wait()
我知道我在这里有点晚了,但我有一个非常类似的问题,但它包括使用tkinter作为GUI,并且主循环使它不可能使用依赖于.join()的任何解决方案。因此,我调整了原问题EDIT中给出的解决方案,但使其更一般,以便于其他人更容易理解。
下面是运行中的新线程类:
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类,无需任何特殊修改。
我认为其他的解决方案有点复杂,如果你唯一想要的是真正看到某个异常,而不是完全无视和盲目。
解决方案是创建一个自定义线程,从主线程获取记录器并记录任何异常。
class ThreadWithLoggedException(threading.Thread):
"""
Similar to Thread but will log exceptions to passed logger.
Args:
logger: Logger instance used to log any exception in child thread
Exception is also reachable via <thread>.exception from the main thread.
"""
def __init__(self, *args, **kwargs):
try:
self.logger = kwargs.pop("logger")
except KeyError:
raise Exception("Missing 'logger' in kwargs")
super().__init__(*args, **kwargs)
self.exception = None
def run(self):
try:
if self._target is not None:
self._target(*self._args, **self._kwargs)
except Exception as exception:
thread = threading.current_thread()
self.exception = exception
self.logger.exception(f"Exception in child thread {thread}: {exception}")
finally:
del self._target, self._args, self._kwargs
例子:
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())
def serve():
raise Exception("Earth exploded.")
th = ThreadWithLoggedException(target=serve, logger=logger)
th.start()
主线程输出:
Exception in child thread <ThreadWithLoggedException(Thread-1, started 139922384414464)>: Earth exploded.
Traceback (most recent call last):
File "/core/utils.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/myapp.py", line 105, in serve
raise Exception("Earth exploded.")
Exception: Earth exploded.
如果在线程中发生异常,最好的方法是在连接期间在调用线程中重新引发它。您可以使用sys.exc_info()函数获取当前正在处理的异常的信息。此信息可以简单地存储为线程对象的属性,直到调用join,此时可以重新引发它。
注意,队列。队列(在其他回答中建议)在这个简单的情况下是不必要的,因为线程最多抛出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:
在Python 3中,raise的参数形式为3,因此将最后一行更改为:
raise new_exc.with_traceback(self.exc[2])
concurrent.futures.as_completed
https://docs.python.org/3.7/library/concurrent.futures.html#concurrent.futures.as_completed
解决方案如下:
当调用异常时,立即返回主线程
不需要额外的用户定义类,因为它不需要:
显式队列
在工作线程周围添加except 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:并发。如何使其可取消?
线程:有办法杀死一个线程吗?
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上测试。