Python FastAPI/Uvicorn技术问题：如何向uvicorn.run()传递Queue？

针对你遇到的问题，核心原因是Uvicorn的reload=True模式会启动额外的监控进程，导致普通的multiprocessing.Queue无法在子进程中复用；同时直接通过字符串导入app的方式会触发模块重新加载，全局变量失效。下面提供两种可行方案：

方案一：关闭热重载（reload），直接传递FastAPI实例 #

如果不需要开发时的热重载功能，这种方案最简单可靠：

from multiprocessing import Queue
from multiprocessing import Process
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

class WSServer:
    def run(self, queue):
        # 直接在run方法内创建FastAPI实例，并将队列绑定到app的state中
        app = FastAPI()
        app.state.queue = queue

        @app.websocket("/")
        async def websocket_endpoint(websocket: WebSocket):
            await websocket.accept()
            try:
                while True:
                    msg = await websocket.receive_text()
                    # 直接通过app.state访问队列
                    app.state.queue.put(f"收到WebSocket消息：{msg}")
                    # 也可以从队列取数据推送给客户端
                    # if not app.state.queue.empty():
                    #     await websocket.send_text(app.state.queue.get())
            except WebSocketDisconnect:
                pass

        # 直接传入app实例，关闭reload
        uvicorn.run(app, host="0.0.0.0", port=8081, reload=False, access_log=False) 

main_queue: Queue = Queue() 
proc = Process(target=WSServer().run, name="Process: Simple Server", args=(main_queue,))
proc.start()
proc.join()

原理 #

关闭reload后，Uvicorn不会启动监控进程，直接在当前进程内运行服务。我们直接创建FastAPI实例并绑定队列，避免了模块重新加载导致的全局变量失效问题。

方案二：保留热重载，使用Manager创建共享队列 #

如果必须保留reload功能，需要用multiprocessing.Manager创建跨进程共享的Queue——这种队列基于进程间通信实现，不受Uvicorn子进程的影响：

from multiprocessing import Manager, Process
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

# 必须在主进程创建Manager和共享队列，确保所有子进程都能访问
manager = Manager()
main_queue = manager.Queue()

app = FastAPI()
app.state.queue = main_queue

@app.websocket("/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            msg = await websocket.receive_text()
            app.state.queue.put(f"收到WebSocket消息：{msg}")
            # 示例：从队列取数据发送给客户端
            # if not app.state.queue.empty():
            #     await websocket.send_text(app.state.queue.get())
    except WebSocketDisconnect:
        pass

class WSServer:
    def run(self):
        # 开启reload模式，使用共享队列不受影响
        uvicorn.run("__main__:app", host="0.0.0.0", port=8081, reload=True, access_log=False) 

# 启动服务进程
proc = Process(target=WSServer().run, name="Process: Simple Server")
proc.start()
proc.join()

原理 #

multiprocessing.Manager会启动一个独立的管理进程，所有通过它创建的队列都会通过这个进程进行数据转发，不管Uvicorn启动多少子进程，都能访问到同一个队列实例。

为什么你之前的尝试失败 #

• 直接给uvicorn.run()传Queue：该函数没有接收自定义对象的参数，无法直接传递。
• 全局变量：reload模式下Uvicorn会重新导入模块，全局变量会被重新初始化，原来的Queue对象无法复用。
• 传递内存地址：进程间内存是隔离的，无法通过内存地址直接获取其他进程的对象。

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