FastAPI 項(xiàng)目啟動/關(guān)閉事件實(shí)戰(zhàn)：數(shù)據(jù)庫連接、緩存預(yù)熱、模型加載一站式搞定！
作者：Ss肥魚
FastAPI 原生就支持啟動與關(guān)閉事件機(jī)制（startup/shutdown），而且使用起來異常優(yōu)雅！
在生產(chǎn)環(huán)境中，我們經(jīng)常會遇到這樣的需求：
- 項(xiàng)目啟動時，需要連接數(shù)據(jù)庫、加載機(jī)器學(xué)習(xí)模型或預(yù)熱緩存
- 項(xiàng)目關(guān)閉時，需要優(yōu)雅釋放連接、清理資源，確保不會“僵尸占用”
好消息是：FastAPI 原生就支持啟動與關(guān)閉事件機(jī)制（startup/shutdown），而且使用起來異常優(yōu)雅！
一、什么是生命周期事件？
FastAPI 提供兩種方式注冊生命周期事件：
方法一：使用 @app.on_event() 裝飾器（經(jīng)典方式）
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def startup_event():
print("應(yīng)用啟動，初始化資源...")
@app.on_event("shutdown")
async def shutdown_event():
print("應(yīng)用關(guān)閉，釋放資源...")方法二：使用 lifespan() 上下文函數(shù)（推薦方式）
FastAPI 1.0 后推薦使用 lifespan 函數(shù)進(jìn)行統(tǒng)一管理，功能更清晰：
from fastapi import FastAPI
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
print("? 應(yīng)用啟動 - startup")
# 初始化資源
yield
print("?? 應(yīng)用關(guān)閉 - shutdown")
# 清理資源
app = FastAPI(lifespan=lifespan)二、實(shí)戰(zhàn)應(yīng)用場景演示
我們將模擬如下生命周期管理需求：
場景 | 操作 |
啟動時 | 連接數(shù)據(jù)庫、預(yù)熱緩存、加載模型 |
關(guān)閉時 | 關(guān)閉數(shù)據(jù)庫連接、清空緩存 |
示例結(jié)構(gòu)：
project/
├── main.py
├── utils/
│ ├── database.py
│ ├── cache.py
│ └── model_loader.py1. 模擬數(shù)據(jù)庫連接模塊：database.py
# utils/database.py
class DBClient:
def __init__(self):
self.connected = False
async def connect(self):
print("?? 正在連接數(shù)據(jù)庫...")
self.connected = True
async def disconnect(self):
print("? 正在關(guān)閉數(shù)據(jù)庫連接...")
self.connected = False
db_client = DBClient()2. 模擬緩存模塊：cache.py
# utils/cache.py
cache = {}
async def preload_cache():
print("? 預(yù)熱緩存中...")
cache["hot_data"] = [1, 2, 3, 4]
async def clear_cache():
print("?? 清理緩存...")
cache.clear()3. 模擬模型加載模塊：model_loader.py
# utils/model_loader.py
model = None
async def load_model():
global model
print("?? 加載機(jī)器學(xué)習(xí)模型...")
model = "MyModel"
async def unload_model():
global model
print("?? 卸載模型...")
model = None4. 項(xiàng)目入口：main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from utils.database import db_client
from utils.cache import preload_cache, clear_cache
from utils.model_loader import load_model, unload_model
@asynccontextmanager
async def lifespan(app: FastAPI):
# 應(yīng)用啟動
await db_client.connect()
await preload_cache()
await load_model()
yield
# 應(yīng)用關(guān)閉
await db_client.disconnect()
await clear_cache()
await unload_model()
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Hello, FastAPI 生命周期！"}三、啟動效果（控制臺日志）
啟動應(yīng)用后你會看到類似輸出：
?? 正在連接數(shù)據(jù)庫...
? 預(yù)熱緩存中...
?? 加載機(jī)器學(xué)習(xí)模型...關(guān)閉應(yīng)用時：
? 正在關(guān)閉數(shù)據(jù)庫連接...
?? 清理緩存...
?? 卸載模型...四、總結(jié)
FastAPI 的生命周期事件非常適合處理啟動初始化與資源釋放的場景：
- lifespan() 是更現(xiàn)代、推薦的方式
- 支持 異步資源初始化，天然適配 async 框架
- 非常適合用于加載模型、連接 Redis、數(shù)據(jù)庫、Kafka、RabbitMQ 等
責(zé)任編輯：趙寧寧
來源：
Ssoul肥魚
