FastAPI中使用multiprocessing Pipe实现Keras模型推理遇通信问题

当你用uvicorn启动FastAPI服务时，默认会根据CPU核心数创建多个工作进程处理请求。你在全局定义的pipe_p会在每个工作进程中被重新初始化，而子进程只和启动uvicorn的主进程的Pipe端建立了连接。API请求是在uvicorn的工作进程中处理的，这里的pipe_p和子进程监听的不是同一个管道，所以子进程收不到数据。

方案一：强制uvicorn单进程启动 #

如果不需要多进程处理API请求，最简单的方式是指定uvicorn只启动1个工作进程，这样全局Pipe能和子进程正常通信：

修改主服务的启动代码：

if __name__ == "__main__":
    p = Process(target=main, args=(pipe_c, model_type, model_name))
    p.start()
    uvicorn.run("main:app", port=port, host=host, workers=1)  # 新增workers=1参数

方案二：使用Manager共享管道（多进程场景推荐） #

如果需要多进程处理API请求，用multiprocessing.Manager创建可跨进程共享的Pipe，确保所有uvicorn工作进程都能访问同一个通信管道：

主服务代码调整：

from multiprocessing import Process, Manager
from multiprocessing.connection import PipeConnection

app = FastAPI()
pipe_p = None  # 全局变量，后续通过Manager初始化

@app.post("/api/test", response_model=MyResponse)
async def test(req:MyDataModel= Body(example=c.INFOEXAMPLE)):
    global pipe_p
    pipe_p.send(req)
    result = pipe_p.recv()
    return {
        "data": f"inference result: {result}"
    }

if __name__ == "__main__":
    with Manager() as manager:
        pipe_p, pipe_c = manager.Pipe()
        p = Process(target=main, args=(pipe_c, model_type, model_name))
        p.start()
        uvicorn.run("main:app", port=port, host=host)

方案三：改用Queue替代Pipe #

multiprocessing.Queue是线程和进程安全的，更适合多生产者（API请求）多消费者（推理子进程）的场景，代码调整更直观：

主服务代码 #

from multiprocessing import Process, Queue

app = FastAPI()
request_queue = None
response_queue = None

@app.post("/api/test", response_model=MyResponse)
async def test(req:MyDataModel= Body(example=c.INFOEXAMPLE)):
    global request_queue, response_queue
    request_queue.put(req)
    result = response_queue.get()
    return {
        "data": f"inference result: {result}"
    }

if __name__ == "__main__":
    request_queue = Queue()
    response_queue = Queue()
    p = Process(target=main, args=(request_queue, response_queue, model_type, model_name))
    p.start()
    uvicorn.run("main:app", port=port, host=host)

子进程代码 #

def main(request_queue, response_queue, model_type:str, model_name:str):
    model = load_model(model_type)(model_name)

    while True:
        print("waiting...")
        data = request_queue.get()
        print(f"received {data}")
        if data == "kill":
            return
        if data == "test":
            print(data)
            continue
        result = model(data)
        response_queue.put(result)

• 确保传递给子进程的请求数据是可序列化的（multiprocessing依赖pickle序列化，Keras模型的输入数据要符合序列化要求）
• 模型必须在子进程内部加载，不要在主进程加载后传递给子进程（Keras模型跨进程传递会出现兼容性问题）
• 多进程场景下，避免用普通全局变量存储通信管道/队列，必须用进程间共享机制

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