使用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)

这篇关于使用ProcessPoolExecutor在执行程序中运行AsyncIO的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！