在 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-ed）与失败（引发异常且未在工作函数中捕获）。为此，您需要在每个Future 上调用.result()（这将导致它要么重新raise 来自任务的异常，要么产生return-ed 值）。还有其他方法可以在不实际引发主线程的情况下进行检查（例如.exception()），但.result() 是最直接的方法。

如果你想让它重新raise，最简单的方法就是将wait()调用替换为：

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

它将在Futures 完成时处理结果，并在发生时立即处理raise 和Exception。或者，您可以继续使用 wait，以便所有任务在检查任何异常之前完成，然后直接迭代 fs 并在每个任务上调用 .result()。

Answer 2

还有另一种方法可以使用 multiprocessing.Pool（用于进程）或 multiprocessing.pool.ThreadPool（用于线程）。据我所知，它会重新抛出任何捕获的异常。