在 multiprocessing pool.map_async() 中处理 multiprocessing.TimeoutError答案

【问题标题】：Handle multiprocessing.TimeoutError in multiprocessing pool.map_async()在 multiprocessing pool.map_async() 中处理 multiprocessing.TimeoutError
【发布时间】：2016-05-30 13:36:43
【问题描述】：

到目前为止，我是这样做的：

rets=set(pool.map_async(my_callback, args.hosts).get(60*4))

如果超时，我会得到一个异常：

 File "/usr/lib/python2.7/multiprocessing/pool.py", line 524, in get
    raise TimeoutError
multiprocessing.TimeoutError

我想优雅地处理这个：

我可以访问的所有主机的输出都应该进入rets，所有超时的主机应该进入一个单独的列表。

这是怎么做到的？

【问题讨论】：

try-except 显然是解决方案，所以我想知道有什么问题？
hosts I could reach 进入try，all hosts which timed out 进入TimeoutError exception....这里的火箭科学在哪里？
@JohnMee 你说 try-except 是解决方案。但是怎么做？我 my_callback() 被并行调用 100 次。代码中有一行：调用 map_async()。如果引发异常，AFAIK rets 为空。如何获取两个列表：未超时调用的结果和超时主机列表？
啊，我明白了，“捕获”是一切都发生在单行中。您需要trawl through the docs 并弄清楚如何将其分解为多行问题，这样您就有一个语句调用一个进程，从而可以捕获该进程的超时，而不能捕获其他进程。

标签： python python-multiprocessing

【解决方案1】：

据我所知，您不能，或者至少不能使用map_async。 map_async 是一种方便的方法，可以解决特定用例的特定问题，但与您所拥有的不匹配，因为您想要更精细的控制。

但是，你仍然可以做到，你只需要使用多处理模块中更细粒度的方法。特别是，您可以使用apply_async 即时将作业添加到您的池中，这使您可以更好地控制如何处理单个任务的成功和失败。

下面是一个非常简单的示例，我认为它可以满足您的需求：

from multiprocessing.pool import Pool, TimeoutError
from time import sleep, time


def task_function(xx):
    print('Task %d running' % xx)
    sleep(xx)
    print('Task %d ended' % xx)
    return 'Result of task %d' % xx

pl = Pool()
results = [
    pl.apply_async(task_function, (_xx,)) 
    for _xx in range(10)]

start = time()
wait_until = start + 5

rets = []
timed_out_results = []

for res in results:
    timeout = wait_until - time()
    if timeout < 0:
        timeout = 0

    try:
        rets.append(res.get(timeout))
    except TimeoutError:
        timed_out_results.append(res)

print('%s ended' % (rets,))
print('%s timedout' % (timed_out_results,))

这会运行 10 个作业，打印一行，休眠，然后打印另一行。第一个休眠 0 秒，下一个休眠 1 秒，接下来是 2 秒，以此类推。我们在 5 秒后使池超时，因此我们预计 5 个任务已完成，5 个任务已超时。

请注意我没有停止仍在运行的任务，因此在现实世界中它们可能会继续并在打印结果所需的时间内完成。您必须弄清楚您对此的关心程度/如何处理。

【讨论】：