在 Web 应用的生命周期中，「启动时初始化」和「关闭时清理」是两个绕不开的需求：数据库连接池需要在服务启动时创建、在关闭时释放；机器学习模型需要在启动时加载到内存；缓存需要在关闭时持久化到磁盘。FastAPI（底层 Starlette）提供了两套生命周期机制：传统的 @app.on_event 装饰器与现代的 lifespan 上下文管理器。本文带你彻底搞懂这两种方式的核心原理与最佳实践。

为什么需要生命周期管理

一个典型的 FastAPI 应用启动后就开始接收请求，但很多资源无法在请求级别按需创建——比如数据库连接池，如果每个请求都新建再销毁，性能会极差；又比如一份 2GB 的 ML 模型，不可能每次推理时重新加载。

生命周期事件要解决的核心问题是：把昂贵的初始化逻辑放在服务启动时执行一次，把资源释放逻辑放在服务关闭时执行一次，让所有请求共享这些预热的资源。

传统方式：on_event 装饰器

FastAPI 早期（Starlette 0.x 时代）使用 on_event 装饰器来注册启动和关闭事件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from fastapi import FastAPI

app = FastAPI()


@app.on_event("startup")
async def startup_db():
    """服务启动时：初始化数据库连接池"""
    import asyncpg
    app.state.db_pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/mydb",
        min_size=5,
        max_size=20,
    )
    print("数据库连接池已创建")


@app.on_event("shutdown")
async def shutdown_db():
    """服务关闭时：释放数据库连接池"""
    await app.state.db_pool.close()
    print("数据库连接池已关闭")


@app.get("/users")
async def list_users():
    async with app.state.db_pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users LIMIT 10")
    return [dict(r) for r in rows]

关键机制

  1. app.state 是跨请求共享状态的推荐载体——它是 Starlette 提供的全局状态字典，生命周期覆盖整个应用实例；
  2. 所有 startup 事件按注册顺序依次执行，所有 shutdown 事件按注册逆序执行；
  3. 事件函数可以是同步或异步的；
  4. 多个 on_event("startup") 可以分别注册，互不干扰。

on_event 的不足

虽然简单直观，但 on_event 有两个明显缺陷：

  • 资源与事件之间没有显式的生命周期绑定关系：如果 startup 中创建了连接池，你需要手动在 shutdown 中关闭它——两者在代码层面没有强制关联，容易遗漏；
  • 无法优雅地处理 startup 阶段的异常：一个 startup 失败，已注册的其他 startup 可能不会正确回滚已初始化的部分。

现代方式：lifespan 上下文管理器

Starlette 从 0.26 版本开始推荐使用 lifespan 上下文管理器，FastAPI 在 0.93+ 版本中正式支持。它通过 Python 原生的 async with 语法，在结构上天然地将初始化与清理绑定在一起：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # ===== 启动阶段：yield 之前的所有代码 =====
    import asyncpg
    app.state.db_pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/mydb",
        min_size=5,
        max_size=20,
    )
    print("数据库连接池已创建 (lifespan)")

    yield  # ← 服务运行期间停在这里

    # ===== 关闭阶段：yield 之后的所有代码 =====
    await app.state.db_pool.close()
    print("数据库连接池已关闭 (lifespan)")


app = FastAPI(lifespan=lifespan)

与 on_event 的核心区别

特性 on_event lifespan
资源生命周期绑定 分散在多个函数，靠约定 包裹在同一函数中，语法级绑定
异常处理 startup 异常可能导致部分资源泄漏 yield 前抛异常会自动跳过关闭逻辑，结构更安全
代码组织 多个装饰器分散在各处 集中管理所有初始化/清理逻辑
推荐版本 旧项目兼容保留 当前推荐方式

注意：lifespan 和 on_event 不能混用——如果传入了 lifespan 参数，on_event 装饰器注册的事件将被忽略。

更多实战场景

场景一：加载机器学习模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载模型（耗时操作只执行一次）
    from transformers import pipeline
    app.state.classifier = pipeline("sentiment-analysis")
    app.state.summarizer = pipeline("summarization")
    print("模型加载完成")

    yield

    # 关闭时释放 GPU 显存
    del app.state.classifier
    del app.state.summarizer
    print("模型资源已释放")


app = FastAPI(lifespan=lifespan)


@app.post("/analyze")
async def analyze(text: str):
    result = app.state.classifier(text)
    return {"sentiment": result[0]}

场景二：缓存预热 + 磁盘持久化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI

CACHE_FILE = "hot_cache.json"


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动：从磁盘加载热缓存
    try:
        with open(CACHE_FILE) as f:
            app.state.cache = json.load(f)
    except FileNotFoundError:
        app.state.cache = {}
    print(f"缓存已加载，{len(app.state.cache)} 条记录")

    yield

    # 关闭：持久化当前缓存到磁盘
    with open(CACHE_FILE, "w") as f:
        json.dump(app.state.cache, f, ensure_ascii=False, indent=2)
    print("缓存已持久化")


app = FastAPI(lifespan=lifespan)


@app.get("/cache/{key}")
async def get_cache(key: str):
    return {"value": app.state.cache.get(key)}

场景三：HTTP 客户端会话管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 创建全局复用的异步 HTTP 客户端
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(30.0),
        limits=httpx.Limits(max_connections=50),
    ) as client:
        app.state.http_client = client
        yield  # 服务运行期间使用这个 client
    # async with 退出时自动关闭


app = FastAPI(lifespan=lifespan)


@app.get("/proxy")
async def proxy_external_api():
    resp = await app.state.http_client.get("https://api.example.com/data")
    return resp.json()

这里使用 httpx.AsyncClient 的 async with 自带生命周期管理，嵌套在 lifespan 中实现双层保障。

注意事项

  1. 不要在 lifespan 中执行耗时过长的阻塞操作——这会导致服务启动慢、健康检查失败。如果初始化确实很慢（如下载模型），考虑使用后台线程或延迟加载策略；
  2. yield 之后的清理代码一定会执行，即使服务收到 SIGTERM 信号——Starlette 会等待清理完成后退出，这是优雅关闭的保障；
  3. 如果 lifespan 中抛出了未捕获异常，FastAPI 会终止启动过程，应用不会开始接收请求；
  4. 避免在 lifespan 中访问请求级别的依赖（如 Request 对象），这些在应用生命周期事件中不可用。

迁移建议

如果你的项目还在使用 @app.on_event，建议尽快迁移到 lifespan：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 迁移前（旧版）
@app.on_event("startup")
async def init(): ...

@app.on_event("shutdown")
async def cleanup(): ...

# 迁移后（推荐）
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 将 init 逻辑放在这里
    ...
    yield
    # 将 cleanup 逻辑放在这里
    ...

app = FastAPI(lifespan=lifespan)

迁移过程通常是机械的——把 startup 代码移到 yield 之前，shutdown 代码移到 yield 之后即可。

总结

要点 说明
推荐方式 lifespan 上下文管理器（FastAPI 0.93+）
共享状态容器 app.state
执行顺序 startup → 接收请求 → shutdown（逆序清理）
适用场景 数据库连接池、模型加载、缓存预热、HTTP 客户端复用
混用禁忌 lifespan 和 on_event 不可共存

掌握生命周期事件是构建「生产可用」FastAPI 应用的关键一步——它让你能优雅地管理全局资源，用最少的代码实现健壮的服务启动与关闭。