You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.20 12:28:05