如何在主asyncio循环的同步子进程中运行多个独立asyncio事件循环

核心原理 #

子进程会继承父进程的地址空间，但asyncio事件循环是进程/线程绑定的，父进程的事件循环在子进程中处于失效状态，因此只需在每个子进程的独立执行逻辑中，初始化新的事件循环即可实现完全隔离的异步运行环境。

完整实现示例 #

你可以配合ProcessPoolExecutor进程池来实现需求，示例代码如下：

import asyncio
from concurrent.futures import ProcessPoolExecutor
import os

# 子进程内要执行的异步业务逻辑
async def sub_async_task(task_id: int):
    print(f"子进程 {os.getpid()} 开始处理异步任务 {task_id}")
    await asyncio.sleep(2) # 模拟异步IO操作
    return f"任务{task_id}处理完成，所属进程ID：{os.getpid()}"

# 提交到进程池的同步包装函数，每个子进程会独立运行该函数
def sub_process_blocking_wrapper(task_id: int):
    # 方式1：手动创建、管理事件循环（兼容所有Python 3版本）
    sub_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(sub_loop)
    try:
        return sub_loop.run_until_complete(sub_async_task(task_id))
    finally:
        sub_loop.close()

    # 方式2：Python 3.7+ 可以直接用asyncio.run简化代码，会自动处理循环的创建、绑定、关闭逻辑
    # return asyncio.run(sub_async_task(task_id))

# 主进程的异步主函数
async def main():
    # 初始化进程池，max_workers可按需调整
    with ProcessPoolExecutor(max_workers=4) as executor:
        main_loop = asyncio.get_running_loop()
        # 批量提交任务到子进程执行
        tasks = [
            main_loop.run_in_executor(executor, sub_process_blocking_wrapper, task_id)
            for task_id in range(6)
        ]
        # 批量获取所有任务执行结果
        results = await asyncio.gather(*tasks)
        for res in results:
            print(res)

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

注意事项 #

• 必须使用ProcessPoolExecutor进程池，不要使用ThreadPoolExecutor线程池：只有进程池的执行单元是独立子进程，线程池所有线程共享同一个进程的内存空间，不存在独立子进程的运行环境
• 不要尝试在子进程中复用父进程的事件循环，强制调用会直接抛出异常
• 如果子进程需要长期运行异步服务，把run_until_complete替换为run_forever即可，进程退出前记得调用loop.stop()和loop.close()完成资源清理

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