FastAPI Lifespan 事件详解：用 lifespan 上下文管理器管理应用启动与关闭逻辑

本文基于 FastAPI 官方文档 docs/ja/docs/advanced/events.md（Lifespan イベント章节）整理撰写，系统讲解如何在 FastAPI 应用中定义“启动前只执行一次”和“关闭时执行一次”的生命周期逻辑。读完后，你将掌握推荐的 lifespan 参数 + 异步上下文管理器的完整写法，理解其与已弃用的 startup/shutdown 事件处理器的区别，并能结合源码看清 FastAPI 内部是如何解析和调度这些事件的。

一、什么是 Lifespan（生命周期）事件

你可以定义一段在应用启动前执行的逻辑——这段代码只执行一次，且发生在应用开始接收请求之前；同样，你也可以定义一段在应用关闭时执行的逻辑——它会在（可能已经处理了）大量请求之后，只执行一次。

由于这段代码在应用开始接收请求之前、以及结束处理请求之后各执行一次，它恰好覆盖了应用的整个生命周期（Lifespan）。这在需要为整个应用设置、并在请求之间共享、最后还需要清理的“资源”场景下非常实用，例如：

• 数据库连接池的创建与释放；
• 共享的机器学习模型的加载与卸载；
• 其他任何需要“获取一次、共享使用、最终清理”的昂贵资源。

二、典型使用场景：昂贵资源的加载

先看一个文档中给出的具体场景：假设你有若干用于处理请求的机器学习模型（🤖）。

关键在于：

1. 模型是共享的——同一个模型被所有请求复用，而不是每个请求或每个用户各加载一份；
2. 加载成本高——模型需要从磁盘读取大量数据，可能耗时相当长，因此绝不能每个请求都加载一次；
3. 也不能放在模块顶层——如果把加载代码写在模块/文件的顶层，那么哪怕只是运行一个与模型无关的简单自动化测试，导入模块时也会被迫加载模型、等待其就绪，导致所有测试都变慢。

正确的解法是：在代码被导入时不加载，而是在应用即将开始接收请求之前才加载模型；并在应用关闭时释放资源（如内存、GPU）。这正是 Lifespan 事件要解决的问题。

三、推荐方案：lifespan 参数 + 异步上下文管理器

这种“启动时”和“关闭时”的逻辑，通过 FastAPI 应用的 lifespan 参数配合一个**上下文管理器（context manager）**来定义。lifespan 参数的类型定义见 fastapi/applications.py：

lifespan: Annotated[
    Lifespan[AppType] | None,
    Doc(
        """
        A `Lifespan` context manager handler. This replaces `startup` and
        `shutdown` functions with a single context manager.
        ...
        """
    ),
] = None

3.1 完整示例

创建一个使用 yield 的异步函数 lifespan()，配合 @asynccontextmanager 装饰器，示例源码见 docs_src/events/tutorial003_py310.py：

from contextlib import asynccontextmanager

from fastapi import FastAPI


def fake_answer_to_everything_ml_model(x: float):
    return x * 42


ml_models = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
    yield
    # Clean up the ML models and release the resources
    ml_models.clear()


app = FastAPI(lifespan=lifespan)


@app.get("/predict")
async def predict(x: float):
    result = ml_models"answer_to_everything"
    return {"result": result}

示例代码逐段说明：

• yield 之前：往机器学习模型字典 ml_models 中放入一个（假装的）模型函数，模拟高成本的“启动时模型加载”。这段代码在应用开始接收请求之前、即启动时执行；
• yield 之后：调用 ml_models.clear() 卸载模型。这段代码在应用结束处理请求之后、关闭前执行——你可以借此释放内存、GPU 等资源；
• app = FastAPI(lifespan=lifespan)：把新建的 lifespan 异步上下文管理器传给 FastAPI，由框架在内部使用。

提示：shutdown 在应用被“停止”时触发——可能是为了启动新版本，也可能只是你决定停止运行了（🤷）。

对应的行为验证测试位于 tests/test_tutorial/test_events/，例如 test_tutorial003.py 会实际启动应用并请求 /predict，确认模型在启动时已加载。

3.2 lifespan 函数的结构

首先要注意到的是：这里定义的是一个使用 yield 的异步函数，这与“带 yield 的依赖项（yield dependencies）”的写法非常相似：

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
    yield
    # Clean up the ML models and release the resources
    ml_models.clear()

• yield 之前的前半部分：在应用启动之前执行；
• yield 之后的后半部分：在应用处理结束之后执行。

3.3 @asynccontextmanager 与非同步上下文管理器

lifespan 函数被 @asynccontextmanager 装饰，这使它成为一个非同步上下文管理器（async context manager）：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    ...

Python 的“上下文管理器”就是能在 with 语句中使用的那些对象，例如 open()：

with open("file.txt") as file:
    file.read()

现代 Python 还有“非同步上下文管理器”，使用 async with：

async with lifespan(app):
    await do_stuff()

这样制作上下文管理器（或非同步上下文管理器）后，效果是：进入 with/async with 块之前，执行 yield 之前的代码；退出该块之后，执行 yield 之后的代码。上面的示例并没有直接使用 async with，而是把它交给 FastAPI 由框架内部代为使用。

四、源码解读：FastAPI 如何解析 lifespan

从 fastapi/routing.py 的源码结构可以看到，lifespan 参数实际上接受三种形态，框架会做归一化处理：

# Determine the lifespan context to use
if lifespan is None:
    # Use the default lifespan that runs on_startup/on_shutdown handlers
    lifespan_context: Lifespan[Any] = _DefaultLifespan(self)
elif inspect.isasyncgenfunction(lifespan):
    lifespan_context = asynccontextmanager(lifespan)
elif inspect.isgeneratorfunction(lifespan):
    lifespan_context = _wrap_gen_lifespan_context(lifespan)
else:
    lifespan_context = lifespan
self.lifespan_context = lifespan_context

这带来两个实用结论：

1. 不写 @asynccontextmanager 也可以：如果你直接传一个使用 yield 的异步生成器函数（async def + yield），FastAPI 会自动帮你包一层 asynccontextmanager（isasyncgenfunction 分支）；普通同步生成器函数也会经 _wrap_gen_lifespan_context 包装为异步上下文管理器。文档示例中显式使用 @asynccontextmanager 仍是清晰、稳妥的写法；
2. 指定 lifespan 后，startup/shutdown 事件处理器将不被调用：当 lifespan is None 时才回退到 _DefaultLifespan（见 fastapi/routing.py）——它是一个兼容层，专门用于运行旧的 on_startup/on_shutdown 处理器。这正是文档警告“lifespan 和事件二选一，不能同时用”的源码依据。

此外，_merge_lifespan_context 展示了 include_router() 时的行为：子路由器的 lifespan 上下文会被嵌套合并进主应用的 lifespan（先进主、再进子，退出顺序相反），并在 yield 时合并两者的 state 字典。

事件处理器（on_event）的弃用实现

已弃用的 @app.on_event("startup") 装饰器定义在 fastapi/routing.py，其装饰器逻辑只是把函数追加到 on_startup 或 on_shutdown 列表（fastapi/routing.py）：

def decorator(func: DecoratedCallable) -> DecoratedCallable:
    self.add_event_handler(event_type, func)
    return func

该方法被明确标记为 deprecated，弃用信息直接指向本文档所在的官方页面。

五、替代方案：startup / shutdown 事件处理器（已弃用）

警告：推荐的方式是使用上文介绍的 lifespan 参数处理“启动”和“关闭”。一旦指定了 lifespan 参数，startup 和 shutdown 事件处理器就不会被调用——lifespan 或事件，只能二选一。本节主要供理解旧代码时参考。

另一种定义启动/关闭逻辑的方式是声明事件处理器函数。这些函数既可以用 async def，也可以用普通 def。

5.1 startup 事件

要在应用启动前执行函数，用事件 "startup" 声明，示例源码见 docs_src/events/tutorial001_py310.py：

from fastapi import FastAPI

app = FastAPI()

items = {}


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

这里 startup 事件处理器把 items 这个“数据库”（一个普通的 dict）初始化为若干值。你可以添加多个事件处理器函数；在所有的 startup 处理器完成之前，应用不会开始接收请求。

5.2 shutdown 事件

要在应用关闭时执行函数，用事件 "shutdown" 声明，示例源码见 docs_src/events/tutorial002_py310.py：

from fastapi import FastAPI

app = FastAPI()


@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.get("/items/")
async def read_items():
    return [{"name": "Foo"}]

这里 shutdown 处理器把文本 "Application shutdown" 写入文件 log.txt。注意 open() 的 mode="a" 表示追加（append）：不会覆盖文件中已有内容，而是把新行追加到末尾。

提示：该例中使用的是标准 Python 的 open() 函数，涉及需要“等待”磁盘写完成的 I/O（输入/输出）。但 open() 本身并不使用 async/await，因此事件处理器函数声明为普通的 def 而非 async def 是合适的——同步 I/O 函数会被放到线程池中执行，不会阻塞事件循环。

5.3 为什么 startup 和 shutdown 分开不够用

启动和关闭的逻辑往往是相关的：你想“开始”某样东西之后再“结束”它，“获取”一个资源之后再“释放”它。如果没有共享逻辑或变量，把它们拆成两个互不通信的独立函数会很困难——你不得不借助全局变量等手段在两者之间传递状态。

因此，官方现在推荐使用第三节的 lifespan 方案：yield 前后的代码天然共享同一个作用域，状态可以就地存放，无需全局变量。

六、技术细节：ASGI Lifespan 协议

对技术细节感兴趣的话：在 ASGI 技术规范内部，上述机制属于 Lifespan 协议的一部分，其中定义了 startup 与 shutdown 两类事件。Starlette 的 lifespan 处理器文档中还有更完整的说明，包括如何在代码的其他位置使用 lifespan 产生的状态（state）——当你的 lifespan 在 yield 时产出一个 state 字典，它会被 Starlette 挂载到请求的 scope["state"] 中，供依赖项和端点访问。

七、注意事项：子应用不会触发 lifespan 事件

⚠️ 请务必注意：这些 lifespan 事件（startup 与 shutdown）只对主应用（main application）执行，对通过 mount() 挂载的子应用不会执行。参见 子应用文档。

从第四节提到的 _merge_lifespan_context 也可以印证这一边界：include_router() 挂载的路由器（router），其 lifespan 会被合并进主应用统一执行；而 mount() 挂载的独立子应用（sub-application） 是独立的 ASGI 应用，拥有自己的生命周期，主应用的 lifespan 事件并不覆盖它。

八、小结

方案	写法	状态	适用场景
lifespan 参数	@asynccontextmanager + async def + yield，FastAPI(lifespan=...)	推荐	启动/关闭逻辑需要共享状态（如加载并清理模型、连接池）
@app.on_event("startup") / ("shutdown")	独立的事件处理器函数，async def 或 def 均可	已弃用	理解旧代码；新代码请迁移到 lifespan

核心要点回顾：

• 启动逻辑写在 yield 之前，只执行一次、先于任何请求；关闭逻辑写在 yield 之后，只执行一次、后于所有请求；
• 指定 lifespan 后，旧的 startup/shutdown 事件处理器不会被调用（源码依据见 fastapi/routing.py 中 lifespan is None 才回退 _DefaultLifespan 的分支）；
• 适合在此类钩子中处理的，是跨请求共享且需要清理的昂贵资源：数据库连接池、机器学习模型、缓存后端等；
• mount() 的子应用不受主应用 lifespan 事件影响。