Python中能否用await等待multiprocessing.Process.join()？

你的代码里await p.join()是行不通的——multiprocessing.Process.join()是同步阻塞方法，它不是协程，直接用await会报错；就算强行调用（不加await），也会卡住整个事件循环，让主进程没法处理其他任务。

要实现「等待独立进程的同时让事件循环处理其他任务」，有两种标准库的可行方案：

方案一：用ProcessPoolExecutor配合asyncio.run_in_executor #

这是最简洁的方式，利用进程池管理CPU密集任务，同时通过asyncio的接口实现异步等待：

import asyncio
from concurrent.futures import ProcessPoolExecutor
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class InputModel(BaseModel):
    data: int

def heavy_duty(input: InputModel) -> None:
    # 模拟耗时CPU密集计算
    result = 0
    for i in range(10**8):
        result += i
    print(f"计算完成：{result}")

# 全局初始化进程池，避免每次请求创建销毁进程的开销
process_pool = ProcessPoolExecutor()

@app.post("/process")
async def process_task(input: InputModel) -> dict:
    # 提交任务到进程池，await会让出事件循环，不阻塞其他请求
    await asyncio.get_event_loop().run_in_executor(process_pool, heavy_duty, input)
    return {"status": "任务完成"}

方案二：异步监听multiprocessing.Process状态 #

如果你需要更精细的进程控制（比如手动管理单个进程），可以通过循环检查进程状态+让出事件循环的方式实现异步等待：

import asyncio
import multiprocessing
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class InputModel(BaseModel):
    data: int

def heavy_duty(input: InputModel) -> None:
    result = 0
    for i in range(10**8):
        result += i
    print(f"计算完成：{result}")

@app.post("/process")
async def process_task(input: InputModel) -> dict:
    p = multiprocessing.Process(target=heavy_duty, args=(input,))
    p.start()
    
    # 循环检查进程状态，每次检查前让出事件循环
    while p.is_alive():
        await asyncio.sleep(0.1)
    
    return {"status": "任务完成"}

关键注意点 #

• 不要直接调用p.join()，它会阻塞事件循环，导致FastAPI无法处理其他请求
• CPU密集型任务必须用进程而不是线程，因为Python的GIL会限制线程的并行计算能力
• 进程池推荐全局初始化，避免每次请求创建进程带来的额外开销

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