带有回调的python异步多处理编程

Question

我正在尝试在 python 中使用回调实现异步多处理编程。我阅读了有关多处理模块、进程/线程池和异步模块的材料，但无法指出。我有一个协程和一个回调函数，比如

async def func(x):
   value = await (other coroutine)
   return x 

def callback(x):
   print(x)

我想将 func(1), func(2), func(3), ... , func(100) 提交给给定数量的工作进程池，比如 5 个进程。然后我想让这些工作人员在任何返回值到达父进程时调用回调，而不是在每个返回值到达父进程之后。

multiprocessing.Pool 中的map/map_async/starmap/starmap_async 方法等待所有来自工作进程的返回到达父进程。我在 python 的 asyncio 模块中学习了异步编程的基本元素，但我仍然难以完成上述任务 - 带有回调的异步多处理编程

谁能给我这个任务的清晰示例代码？

提前致谢。

Answer 1

我认为你想要这个以一种 hacky 方式：

import asyncio
import random
from concurrent.futures import ProcessPoolExecutor


async def do_sth(future, n):
    await asyncio.sleep(random.randint(1, 5))
    future.set_result('{} is done!'.format(n))


def got_result(fut):
    print(fut.result())

if __name__ == '__main__':
    executor = ProcessPoolExecutor(5)

    task = []
    loop = asyncio.get_event_loop()
    for n in range(1, 100):
        future = asyncio.Future()
        task.append(
            asyncio.ensure_future(do_sth(future, n))
        )
        future.add_done_callback(got_result)
    try:
        loop.run_until_complete(asyncio.gather(*task))
    finally:
        loop.close()

但我建议你考虑一下从组合多进程和异步回调中实际获得什么，我认为你真正想要的可能是队列和多进程。