据我所知,你不能使用
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个作业,每个作业都会先打印一行文字,然后sleep,再打印另一行文字。第一个作业sleep 0秒,第二个作业sleep 1秒,第三个作业sleep 2秒,以此类推。我们设置池子在5秒后超时,因此我们期望有5个任务完成且5个任务超时。
请注意,我并没有停止仍在运行的任务,在现实世界中它们可能会继续运行并在打印结果所需的时间内完成。您需要确定对此有多在意并决定如何处理。
rets
将为空。如何获取两个列表:未超时调用的结果列表和超时主机列表? - guettli