Python多进程spawn模式下，FastAPI如何异步获取进程输出？

你的核心需求是让/generate端点异步等待后台Worker进程的计算结果，同时避免轮询output_store。原代码中直接访问output_store[chat_id]会立即抛出KeyError，因为Worker还没生成结果；而轮询会阻塞FastAPI的事件循环，影响并发性能。

解决方案思路 #

利用**多进程同步事件（multiprocessing.Event）**实现结果通知：

• 每个请求生成唯一chat_id时，同时创建一个Event对象，存入进程间共享的字典
• Worker进程处理完请求后，触发对应chat_id的Event
• FastAPI端点通过异步方式等待Event触发，再从output_store获取结果

修改后的完整代码 #

# server.py
import uuid
import asyncio
import multiprocessing as mp
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel


class RAGRequest(BaseModel):
    msg: str 

async def _bg_process(item):
    chat_id, msg = item['chat_id'], item['msg']
    print(f"{chat_id} is running...")
    await asyncio.sleep(10)     # 模拟耗时计算
    print(f"{chat_id} is done.")
    return f"{msg}_output"

# 后台处理函数：完成计算后触发事件
async def bg_process(item, output_store, event_store):
    output = await _bg_process(item)
    chat_id = item['chat_id']    
    # 存储结果
    output_store[chat_id] = {'chat_id': chat_id, "output": output}
    # 触发事件通知主进程
    event = event_store.get(chat_id)
    if event:
        event.set()

async def _worker(input_queue, output_store, event_store):
    async with asyncio.TaskGroup() as tg:
        while True:
            if not input_queue.empty():
                item = input_queue.get()
                cor = bg_process(item, output_store, event_store)
                tg.create_task(cor)
            else:
                await asyncio.sleep(0.1)

# 启动Worker进程的入口
def worker(input_queue, output_store, event_store):
    cor = _worker(input_queue, output_store, event_store)
    asyncio.run(cor)

app = FastAPI()

@app.post("/generate")
async def gen(request: RAGRequest):
    global input_queue
    global output_store
    global event_store
    
    chat_id = str(uuid.uuid1())
    # 创建进程间事件，用于结果通知
    result_event = mp.Event()
    event_store[chat_id] = result_event
    
    # 将请求放入队列
    input_queue.put({'chat_id': chat_id, "msg": request.msg})
    
    # 异步等待事件触发：用asyncio.to_thread避免阻塞事件循环
    await asyncio.to_thread(result_event.wait)
    
    # 获取结果并清理资源
    final_output = output_store[chat_id]['output']
    del output_store[chat_id]
    del event_store[chat_id]
    
    return final_output


if __name__ == "__main__":
    mp.set_start_method("spawn")
    input_queue = mp.Queue()          
    output_store = mp.Manager().dict()  # 进程间共享结果存储
    event_store = mp.Manager().dict()   # 进程间共享事件存储
    
    p = mp.Process(target=worker, args=(input_queue, output_store, event_store))
    p.start()

    uvicorn.run(app, host="0.0.0.0", port=3000)

关键修改点说明 #

1. 新增event_store：用mp.Manager().dict()存储每个chat_id对应的Event，实现进程间共享
2. 端点异步等待：使用asyncio.to_thread(result_event.wait)将阻塞的Event.wait()放到线程中执行，避免占用FastAPI的事件循环，保证并发能力
3. Worker触发事件：计算完成后，从event_store取出对应Event并调用set()，通知主进程结果已就绪
4. 资源清理：获取结果后删除output_store和event_store中的对应条目，避免内存泄漏

测试验证 #

使用你提供的send_requests.py发送请求，每个请求会在Worker完成计算后（约10秒）返回结果，且FastAPI能同时处理多个并发请求，不会被阻塞。

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