在AsyncIO中用ProcessPoolExecutor结合阻塞与非阻塞任务遇异常行为

从你给出的代码片段来看，核心问题应该是阻塞任务的提交方式不对，或者主事件循环被阻塞，导致非阻塞I/O任务无法正常运行。ProcessPoolExecutor确实适合处理CPU密集型的阻塞任务，但需要注意和asyncio事件循环的配合，不能让阻塞操作卡住主循环。

先明确几个关键问题点： #

• BlockingQueueListener.run() 是一个无限循环的阻塞方法，如果直接在主进程调用，会完全卡住主线程，asyncio的事件循环根本跑不起来。
• 非阻塞的NonBlockingListener.non_blocking_listen()是协程，必须交给asyncio事件循环调度，不能和阻塞任务在同一个执行流里。

修正后的完整代码示例 #

import asyncio
from concurrent.futures import ProcessPoolExecutor

class BaseBlockingListener:
    def blocking_listen(self):
        # 模拟阻塞的CPU密集型/同步任务，比如持续监听队列
        while True:
            print("Blocking task running...")
            # 这里替换成实际的阻塞逻辑
            import time
            time.sleep(2)

class BlockingQueueListener(BaseBlockingListener):
    def run(self):
        self.blocking_listen()

class BaseNonBlocking:
    async def get_message(self):
        # 模拟非阻塞I/O任务，比如从网络/消息队列获取数据
        print("Non-blocking task fetching message...")
        await asyncio.sleep(1)
        return "fake message"

class NonBlockingListener(BaseNonBlocking):
    async def non_blocking_listen(self):
        while True:
            msg = await self.get_message()
            print(f"Received: {msg}")

def run_blocking_task(blocking_listener):
    # 这个函数会被提交到ProcessPoolExecutor的子进程中运行
    blocking_listener.run()

async def main():
    # 初始化阻塞任务实例
    blocking_listener = BlockingQueueListener()
    # 初始化非阻塞任务实例
    non_blocking_listener = NonBlockingListener()

    # 创建进程池，建议指定max_workers，避免创建过多进程
    executor = ProcessPoolExecutor(max_workers=1)
    
    # 把阻塞任务提交到进程池，注意这里不要用result()，否则会阻塞主循环
    executor.submit(run_blocking_task, blocking_listener)

    # 运行非阻塞的协程任务
    await non_blocking_listener.non_blocking_listen()

if __name__ == "__main__":
    asyncio.run(main())

关键修正说明： #

1. 阻塞任务的提交方式：

   • 把blocking.run()包装在run_blocking_task函数里，通过executor.submit()提交到子进程，这样主进程的事件循环不会被阻塞。
   • 绝对不要调用future.result()（除非你明确要等待任务结束），否则会卡住asyncio循环。

2. 非阻塞任务的运行：

   • 使用asyncio.run()（Python 3.7+）来启动主协程，或者用loop.run_until_complete()，确保协程被事件循环正确调度。
   • 非阻塞任务的while True循环里用await让出CPU，保证事件循环能处理其他任务。

3. 进程池的注意事项：

   • 提交到ProcessPoolExecutor的函数和参数必须是可序列化的（pickle兼容），所以你的BlockingQueueListener实例要确保能被pickle序列化，如果有不可序列化的属性，需要调整。
   • 根据阻塞任务的CPU密集程度设置max_workers，一般不超过CPU核心数。

可能的其他问题排查： #

• 如果你的阻塞任务是I/O密集型而非CPU密集型，其实用ThreadPoolExecutor更合适，进程池的开销比线程池大很多。
• 如果发现非阻塞任务还是没运行，检查是否在主循环里有其他阻塞调用（比如time.sleep()），必须全部替换成await asyncio.sleep()。

内容的提问来源于stack exchange，提问作者I159