Python多进程与多asyncio事件循环的用法是否正确？

问题背景 #

通过multiprocessing.Process启动3个进程，每个进程运行一组异步任务，期望利用多核CPU同时发挥asyncio的异步特性。但运行后日志显示，虽进程ID不同，却似乎共用同一线程/事件循环。

入口程序main.py #

import asyncio
from loguru import logger
from multiprocessing import Process
from app.events import type_a_tasks, type_b_tasks, type_c_tasks


def run_task(task):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(task())
    loop.run_forever()


def main():
    processes = list()
    processes.append(Process(target=run_task, args=(type_a_tasks,)))
    processes.append(Process(target=run_task, args=(type_b_tasks,)))
    processes.append(Process(target=run_task, args=(type_c_tasks,)))

    for process in processes:
        process.start()
        logger.info(f"Started process id={process.pid}, name={process.name}")

    for process in processes:
        process.join()


if __name__ == '__main__':
    main()

异步任务示例type_a_tasks #

import asyncio
from . import business_1, business_2, business_3, business_4, business_5, business_6


async def type_a_tasks():
    tasks = list()
    tasks.append(asyncio.create_task(business_1.main()))
    tasks.append(asyncio.create_task(business_2.main()))
    tasks.append(asyncio.create_task(business_3.main()))
    tasks.append(asyncio.create_task(business_4.main()))
    tasks.append(asyncio.create_task(business_5.main()))
    tasks.append(asyncio.create_task(business_6.main()))

    await asyncio.wait(tasks)
    return tasks

问题分析与修正 #

1. 代码中的冗余与逻辑问题 #

• loop.run_forever()完全多余：loop.run_until_complete(task())会一直阻塞到type_a_tasks执行完毕。如果business_*是长期运行的异步任务（比如持续监听消息），await asyncio.wait(tasks)会一直挂起，run_until_complete永远不会返回，run_forever()根本没机会执行；如果business_*是一次性任务，run_until_complete执行完后事件循环已无任务，run_forever()会让空循环一直跑，毫无意义。
• 事件循环创建方式不够规范：Python 3.7+推荐使用asyncio.run()替代手动管理事件循环，它会自动创建新循环、运行任务、执行后关闭循环，避免手动操作的潜在问题。

2. “同一线程/事件循环”的误解 #

不同进程的线程ID是进程内独立编号，不同进程的主线程ID可能完全相同，但它们属于不同进程，是完全独立的执行环境，各自拥有独立的事件循环。你看到的“同一线程ID”只是进程内的编号重复，并非真的共用线程/循环。可以通过完善日志格式确认：

# 在main.py开头添加日志格式配置
logger.add(sys.stderr, format="{time} | 进程ID:{process.id} | 线程ID:{thread.id} | {level} | {message}")

修正后的代码 #

修正run_task函数

def run_task(task):
    # 用asyncio.run自动管理事件循环，替代手动操作
    asyncio.run(task())

如果business_*是长期运行的任务（无需type_a_tasks返回），可以简化type_a_tasks：

async def type_a_tasks():
    tasks = [
        asyncio.create_task(business_1.main()),
        asyncio.create_task(business_2.main()),
        asyncio.create_task(business_3.main()),
        asyncio.create_task(business_4.main()),
        asyncio.create_task(business_5.main()),
        asyncio.create_task(business_6.main()),
    ]
    # 等待所有任务完成（如果是长期任务，此调用会一直阻塞）
    await asyncio.gather(*tasks)

最终结论 #

你的多进程+asyncio的核心思路是正确的——通过多进程利用多核，每个进程内用asyncio处理IO密集型异步任务，完全符合场景需求。只需修正代码中的冗余逻辑，并通过日志格式区分进程/线程ID即可消除误解。

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