检测concurrent.futures中的失败任务

Question

我一直在使用concurrent.futures,因为它有一个简单的界面,让用户可以轻松控制线程/进程的最大数量.但是,似乎concurrent.futures隐藏了失败的任务,并在所有任务完成/失败后继续主线程.

import concurrent.futures

def f(i):
    return (i + 's')

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    fs = [executor.submit(f, i ) for i in range(10)]
    concurrent.futures.wait(fs)

在任何整数上调用f都会导致TypeError.但是,整个脚本运行得很好,并以代码0退出.有没有办法让它在任何线程失败时抛出异常/错误？

或者,有没有更好的方法来限制线程/进程数量而不使用concurrent.futures？

Answer 1

concurrent.futures.wait将确保完成所有任务,但它不会检查成功(某些内容return)与失败(引发异常并且未在工作人员函数中捕获).要做到这一点,你需要调用.result()每个Future(这将导致它raise从任务中重新获得异常,或者产生return-ed值).还有其他方法可以在没有实际提升主线程的情况下进行检查(例如.exception()),但这.result()是最简单的方法.

如果你想重新做raise,最简单的方法就是用以下方法替换wait()呼叫:

for fut in concurrent.futures.as_completed(fs):
    fut.result()

将处理结果Future必须填写,并及时raise的Exception,如果一次发生.或者,您继续使用wait所有任务在检查其中任何一个的异常之前完成,然后fs直接迭代并调用.result()每个任务.