使用 ProcessPoolExecutor 在执行器中运行 AsyncIO答案

【问题标题】：AsyncIO run in executor using ProcessPoolExecutor使用 ProcessPoolExecutor 在执行器中运行 AsyncIO
【发布时间】：2018-04-23 10:02:48
【问题描述】：

我尝试使用 ProcessPoolExecutor 组合阻塞任务和非阻塞（I/O 绑定）任务，发现它的行为非常出乎意料。

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 块中开始并且永远不会返回控制权。怎么会发生？在多处理执行器中开始结合普通协程和期货的正确方法是什么？

【问题讨论】：

标签： python concurrency multiprocessing

【解决方案1】：

这一行：

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

将实际运行阻塞函数并将其结果提供给执行器。

根据documentation，您需要显式传递函数，后跟其参数。

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

【讨论】：