如何在FastAPI服务器启动后异步执行初始化函数且不阻塞服务启动?
如何在FastAPI服务器启动后异步执行初始化函数且不阻塞服务启动?
我完全懂你的困扰:把create_data()放在uvicorn.run(app)前面,服务得等初始化完才对外响应;放在后面又根本不会执行——因为uvicorn.run是阻塞式调用,会一直占着主线程。下面给你两个实用的解决办法:
方案一:用线程跑同步初始化逻辑
如果你的create_data是同步函数(比如包含time.sleep这种阻塞操作),最直接的方式是用Python的threading模块,在启动服务前开个后台线程执行初始化,让服务和初始化并行运行。
修改后的完整代码:
import time import threading import uvicorn from fastapi import FastAPI DATA = {"value": ""} # 可选:如果DATA会被多线程读写,加锁避免竞态问题 data_lock = threading.Lock() app = FastAPI() @app.get("/") def get_root(): return DATA def create_data(): time.sleep(2) # 操作共享数据时加锁,保证线程安全 with data_lock: DATA["value"] = "Hello World!" print("初始化完成:DATA已更新") if __name__ == "__main__": # 开启后台线程,daemon=True让线程随主进程结束自动退出 threading.Thread(target=create_data, daemon=True).start() # 启动FastAPI服务,此时主线程被uvicorn占用,但初始化线程在后台跑 uvicorn.run(app)
启动后你会发现:服务立刻就能响应/请求(初始返回{"value": ""}),2秒后初始化完成,再请求就会返回{"value": "Hello World!"}。
方案二:用异步任务实现(适合异步初始化逻辑)
如果你的初始化逻辑可以改成异步的(比如把time.sleep换成asyncio.sleep),可以用FastAPI的启动事件配合asyncio.create_task,把初始化逻辑放到后台异步执行,不会阻塞服务启动。
修改后的代码示例:
import asyncio import uvicorn from fastapi import FastAPI DATA = {"value": ""} app = FastAPI() @app.get("/") def get_root(): return DATA async def create_data(): await asyncio.sleep(2) DATA["value"] = "Hello World!" print("初始化完成:DATA已更新") @app.on_event("startup") async def startup_event(): # 把异步初始化函数包装成后台任务,启动事件会立刻完成,不阻塞服务 asyncio.create_task(create_data()) if __name__ == "__main__": uvicorn.run(app)
注意事项
- 如果初始化是CPU密集型操作,异步方案没性能优势,优先选线程方案;如果是IO密集型(比如等数据库连接、文件读写),异步方案更高效。
- 涉及共享数据(比如示例里的
DATA)时,一定要注意线程/协程安全,必要时加锁(线程用threading.Lock,异步用asyncio.Lock)。
备注:内容来源于stack exchange,提问作者pyjedy
相关产品推荐
相关产品推荐

