使用ProcessPoolExecutor在执行程序中运行AsyncIO

本文介绍了使用ProcessPoolExecutor在执行程序中运行AsyncIO的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我尝试使用ProcessPoolExecutor将阻止任务和非阻止(绑定I/O的)任务组合在一起，发现它的行为非常出乎意料.

I tried to combine blocking tasks and non-blocking (I/O bound) tasks using ProcessPoolExecutor and found it's behavior pretty unexpected.

class BlockingQueueListener(BaseBlockingListener): def run(self): # Continioulsy listening a queue blocking_listen() class NonBlockingListener(BaseNonBlocking): def non_blocking_listen(self): while True: await self.get_message() def run(blocking): blocking.run() if __name__ == "__main__": loop = asyncio.get_event_loop() executor = ProcessPoolExecutor() blocking = BlockingQueueListener() non_blocking = NonBlockingListener() future = loop.run_in_executor(executor, run(blocking)) loop.run_until_complete( asyncio.gather( non_blocking.main(), future ) )

我期望这两个任务将同时具有控制权，但是阻止任务以ProcessPoolExecutor块开始，并且永远不会返回控制权.怎么会这样在多处理执行器中，将常规协程和期货结合起来的正确方法是什么?

I was expecting that both tasks will have control concurrently, but blocking task started in ProcessPoolExecutor blocks and never return control. How could it happen? What the proper way to combine normal coroutines and futures started in multiprocessing executor?

推荐答案

此行:

future = loop.run_in_executor(executor, run(blocking))

实际上将运行阻塞函数并将其结果提供给执行程序.

Will actually run the blocking function and give its result to the executor.

根据文档，您需要显式地传递该函数以及其参数.

According to the documentation, you need to pass the function explicitly followed by its arguments.

future = loop.run_in_executor(executor, run, blocking)