如何从池中获取第一个完成的异步结果

Question

假设我有 3 个任务。

tasks = [task1, task2, task3]

我想并行运行这 3 个任务，所以我将它们放在一个线程池中。

from multiprocessing.pool import ThreadPool
worker_pool = ThreadPool(len(tasks))
async_results = [worker_pool.apply_async(task) for task in tasks]

然后我需要get() 这些异步结果。这是我的工作：

[async_result.get() for async_result in async_results]

但是，此方法尝试将这些异步结果一一get()。 task2 有可能实际上在进程中间失败了，我想通过get() 相应的异步结果来了解这一点。在这种情况下，这里的方法实际上会在第一个任务处被阻塞。我只能在 task1 完成后注意到 task2 失败。

我想知道是否可以通过某种方式获得首先​​准备好的异步结果。

Answer 1

我不认为多处理的 Pool 能够做到这一点。你真的应该试试concurrent.futures，它提供了比 Pool 更多的功能，并且更易于使用。

concurrent.futures.as_completed 只为你的任务：

import concurrent.futures

with concurrent.futures.ProcessPoolExecutor(max_workers=5) as worker_pool: 
    futures = [worker_pool.submit(task) for task in tasks]
    for future in concurrent.futures.as_completed(futures):
        # completed task is yielded first 
        result = future.result()

就是这样。

Answer 2

您可以查看Pool.imap_unordered:

async_results = worker_pool.imap_unordered(apply, tasks)
results = [result for result in async_results]