python - 在 multiprocessing pool.map_async() 中处理 multiprocessing.TimeoutError

Question

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

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，所有超时的主机应该进入一个单独的列表。

这是怎么做到的？

更新

六年后，我认为对于并发应用程序使用 go 而不是 Python 更有意义。

Answer 1

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

不过，你仍然可以做到，你只需要使用 multiprocessing 模块中更细粒度的方法。特别是，您可以使用 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 个任务已超时。

请注意我没有停止仍在运行的任务，因此在现实世界中它们可能会继续并在打印结果所需的时间内完成。你必须弄清楚你有多关心它/该怎么做。