文章目录

FastAPI 系统性学习记录

学习起点

从一个基础的 test.py 开始，问题是：“代码里没有 HTML，为什么访问 /docs 能看到页面？”

这个疑问引出了一整条学习链路：FastAPI 做了什么 → Uvicorn 是什么 → ASGI 协议怎么连接它们 → 异步编程怎么工作 → 怎么返回真正的 HTML 页面。

一、整体架构总览

你的 Python 代码 → FastAPI（框架） → ASGI（协议） → Uvicorn（服务器） → 浏览器

┌───────────────────────────────┐
│       应用层（框架）           │
│  FastAPI                      │
│  职责：定义路由、处理业务逻辑   │
│  等价于：餐厅的菜谱和厨师       │
├───────────────────────────────┤
│       协议层（接口规范）        │
│  ASGI / WSGI                  │
│  职责：框架和服务器通信标准     │
│  等价于：传菜单的标准格式       │
├───────────────────────────────┤
│       服务器层（运行实体）      │
│  Uvicorn                      │
│  职责：监听端口、接收HTTP请求   │
│  等价于：餐厅服务员和收银台     │
└───────────────────────────────┘

核心认知：这三层是解耦的。FastAPI 不需要知道 Uvicorn 的存在，Uvicorn 也不需要知道 FastAPI 的存在，它们通过 ASGI 协议通信。

二、核心组件详解

2.1 FastAPI

FastAPI 是一个基于 Starlette 和 Pydantic 的现代 Python Web 框架。

Starlette 是什么？

Starlette 是一个轻量级的 ASGI 框架/工具包，提供了 Web 框架最底层的核心能力。

FastAPI 直接复用了 Starlette 的以下能力：

能力 Starlette 提供 FastAPI 的使用
路由匹配 @route() @app.get() @app.post()
请求/响应 Request Response 对象 直接继承使用
中间件 @app.middleware() 直接继承使用
WebSocket WebSocket 类 直接继承使用
后台任务 BackgroundTasks 直接继承使用
静态文件 StaticFiles 挂载 app.mount("/static", ...)

关键认知：FastAPI 不是从零写的。它在 Starlette 的基础上增加了类型校验（Pydantic）和自动文档（OpenAPI）。

Starlette 提供：路由、请求/响应、中间件、WebSocket
     ↓
Pydantic 提供：数据校验、序列化
     ↓
FastAPI 整合两者 + 自动文档
     ↓
成品

运行以下代码理解 Starlette 和 FastAPI 的关系：

# starlette_demo.py — 直接用 Starlette 写一个 Web 服务
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
import uvicorn


async def list_items(request):
    """路由处理函数：接收 Request 对象，返回 Response"""
    items = [
        {"id": 1, "name": "笔记本电脑", "price": 5999.0},
        {"id": 2, "name": "机械键盘", "price": 299.0},
    ]
    return JSONResponse(items)


async def read_item(request):
    """从 URL 路径中提取参数"""
    item_id = request.path_params["item_id"]
    items = {1: "笔记本电脑", 2: "机械键盘"}
    return JSONResponse({"id": int(item_id), "name": items.get(int(item_id))})


# 手动注册路由（FastAPI 的 @app.get() 只是简化了这一步）
routes = [
    Route("/items", endpoint=list_items),
    Route("/items/{item_id:int}", endpoint=read_item),
]

app = Starlette(routes=routes)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

对比 FastAPI 版本：

# fastapi_demo.py — 同一功能的 FastAPI 版本
from fastapi import FastAPI
import uvicorn

app = FastAPI()

items_db = {
    1: {"name": "笔记本电脑", "price": 5999.0},
    2: {"name": "机械键盘", "price": 299.0},
}


@app.get("/items")
async def list_items():
    return list(items_db.values())


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return items_db[item_id]


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

区别一目了然：

  • Starlette：手动解析路径参数、手动构造 Response、手动注册路由
  • FastAPI：参数类型提示自动解析、自动序列化返回值、装饰器自动注册
Pydantic 是什么？

Pydantic 是一个数据校验库，利用 Python 类型注解进行运行时数据验证。

# pydantic_demo.py — Pydantic 的核心能力
from pydantic import BaseModel, ValidationError
from typing import Optional


# 定义一个数据模型
class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None
    tax: Optional[float] = None


# 用法 1：用字典创建对象（自动校验类型）
data = {"name": "笔记本电脑", "price": 5999.0, "tax": 0.13}
item = Item(**data)
print(f"名称: {item.name}, 价格: {item.price}")
print(f"含税总价: {item.price * (1 + item.tax)}")

# 用法 2：传错类型会报错
try:
    Item(name="鼠标", price="不是数字")  # 应该传 float
except ValidationError as e:
    print(f"校验失败: {e}")

# 用法 3：转回字典/JSON
print(f"字典: {item.model_dump()}")
print(f"JSON: {item.model_dump_json()}")

Pydantic 在 FastAPI 中的作用：

  • 请求到达时：自动校验请求体是否符合模型定义
  • 响应返回时：自动将返回值序列化为 JSON
  • 文档生成时：自动生成 OpenAPI 的 schema 定义

2.2 Uvicorn

Uvicorn 是一个基于 uvloop 和 httptools 的高性能 ASGI 服务器。

uvloop 是什么？

uvloop 是 asyncio 事件循环的替代品，用 Cython 编写，底层使用 libuv（与 Node.js 相同的事件循环库）。

标准 asyncio 事件循环：纯 Python 实现
     ↓ 替换
uvloop：C 语言实现（libuv）
     ↓ 效果
处理速度提升 2-4 倍

# 理解 uvloop 的作用（不需要手动调用，Uvicorn 自动使用）
import asyncio
import uvloop


# 标准事件循环
loop = asyncio.new_event_loop()

# uvloop 事件循环（更快）
loop = uvloop.new_event_loop()

# Uvicorn 启动时自动使用 uvloop（如果安装了 uvloop 包）
httptools 是什么？

httptools 是一个高性能 HTTP 解析器，用 C 语言编写，是对 Node.js 的 http-parser 的 Python 绑定。

原始 TCP 字节流：
    b"GET /items/123?q=hello HTTP/1.1\r\nHost: 127.0.0.1:8000\r\n..."

httptools 解析（C 语言，极快）：
    method = "GET"
    path = "/items/123"
    query = "q=hello"
    headers = {"Host": "127.0.0.1:8000", ...}

uvloop + httptools 组合的效果：Uvicorn 是目前最快的 Python Web 服务器之一。

Uvicorn 到底做了什么？

最重要的三件事：

1. 监听端口（默认为 127.0.0.1:8000）
2. 用 httptools 解析 HTTP 请求 → 构造成 ASGI scope 字典
3. 用 uvloop 事件循环调用 FastAPI：await app(scope, receive, send)

完整演示：手动模拟 Uvicorn 做的事（理解底层原理）：

# manual_uvicorn.py — 模拟 Uvicorn 的内部工作
import asyncio
import json


# 这是 FastAPI 应用
async def my_fastapi_app(scope, receive, send):
    """符合 ASGI 规范的 FastAPI 应用"""
    method = scope["method"]
    path = scope["path"]

    # 路由匹配
    if method == "GET" and path == "/":
        body = json.dumps({"message": "Hello World"}).encode()
        await send({
            "type": "http.response.start",
            "status": 200,
            "headers": [(b"content-type", b"application/json")],
        })
        await send({"type": "http.response.body", "body": body})
    else:
        await send({
            "type": "http.response.start",
            "status": 404,
            "headers": [(b"content-type", b"text/plain")],
        })
        await send({"type": "http.response.body", "body": b"Not Found"})


# 模拟 Uvicorn 的内部工作
async def mock_uvicorn():
    """手动模拟 Uvicorn 做的事情"""
    print("Uvicorn 启动，监听 127.0.0.1:8000...")

    # 模拟接收了一个 HTTP 请求
    raw_request = b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"

    # 用 httptools 解析（这里手动构造 scope）
    scope = {
        "type": "http",
        "method": "GET",
        "path": "/",
        "headers": [(b"host", b"localhost")],
        "query_string": b"",
    }

    # 定义 receive 和 send
    async def receive():
        return {"type": "http.request", "body": b""}

    async def send(message):
        if message["type"] == "http.response.start":
            print(f"发送状态: {message['status']}")
        elif message["type"] == "http.response.body":
            print(f"发送响应体: {message['body'].decode()}")

    # 调用 FastAPI 应用
    print("调用 app(scope, receive, send)")
    await my_fastapi_app(scope, receive, send)
    print("响应发送完成")


asyncio.run(mock_uvicorn())

2.3 ASGI

ASGI 是一个接口规范，定义了框架和服务器之间的通信标准。

核心定义

一个 ASGI 应用就是一个可调用对象，必须符合以下签名：

async def app(scope: dict, receive: callable, send: callable):
    pass

三个参数的含义：

  • scope：Uvicorn 把 HTTP 请求翻译成的 Python 字典（包含 method、path、headers 等）
  • receive：用来接收请求体数据的异步函数
  • send：用来发送 HTTP 响应的异步函数
Scope 长什么样

运行以下代码查看 scope 的完整结构：

# scope_structure.py — 查看 ASGI scope 完整结构
scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "GET",
    "path": "/items/123",
    "raw_path": b"/items/123",
    "root_path": "",
    "scheme": "http",
    "query_string": b"q=hello",
    "headers": [
        (b"host", b"127.0.0.1:8000"),
        (b"user-agent", b"Mozilla/5.0"),
        (b"accept", b"*/*"),
    ],
    "client": ("127.0.0.1", 54321),
    "server": ("127.0.0.1", 8000),
}

print("scope 是 Python 字典，包含以下字段：")
for key in scope:
    print(f"  {key}")
print(f"\nFastAPI 从 scope 提取信息：")
print(f"  path → /items/123  → 路由匹配 → 提取路径参数 id=123")
print(f"  method → GET       → 匹配 @app.get()")
print(f"  query_string → q=hello → 自动解析查询参数 q='hello'")

2.4 Swagger UI：自动文档

文档从哪里来？
你的代码 → @app.get() / Pydantic 模型 / 类型提示 / 文档字符串
    ↓  FastAPI 自动解析（无需手动编写）
/openapi.json（符合 OpenAPI 规范的 JSON）
    ↓  Swagger UI（JavaScript 前端）读取并渲染
交互式 API 文档页面

关键认知：Swagger UI 不是 FastAPI 特有的。它是一个独立的前端项目，只要提供符合 OpenAPI 规范的 JSON，任何后端框架都可以使用它。FastAPI 的优势在于自动生成这个 JSON，不需要手动编写。

三、怎么返回真正的 HTML 页面

正式生产环境中的页面不能只有 /docs，怎么写其他页面？

FastAPI 不仅可以返回 JSON，也可以返回 HTML。有三种方式：

3.1 直接返回 HTML 字符串

# html_demo.py — 直接返回 HTML
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()


@app.get("/", response_class=HTMLResponse)
async def home():
    """返回完整 HTML 页面"""
    html_content = """
    <!DOCTYPE html>
    <html>
    <head>
        <title>FastAPI 示例站点</title>
        <style>
            body { font-family: Arial; max-width: 800px; margin: 0 auto; padding: 20px; }
            h1 { color: #007bff; }
            .card { border: 1px solid #ddd; padding: 15px; margin: 10px 0; border-radius: 5px; }
        </style>
    </head>
    <body>
        <h1>欢迎来到 FastAPI 站点</h1>
        <div class="card">
            <h2>这是一个动态页面</h2>
            <p>这个 HTML 是由 FastAPI 路由直接返回的。</p>
            <p><a href="/items">查看商品列表</a></p>
            <p><a href="/docs">查看 API 文档</a></p>
        </div>
    </body>
    </html>
    """
    return HTMLResponse(content=html_content)

3.2 使用 Jinja2 模板引擎（生产推荐）

# template_demo.py — 使用 Jinja2 模板渲染 HTML
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
import uvicorn
import os

app = FastAPI()

# 挂载静态文件目录（CSS、JS、图片）
# 需要项目目录下有 static 文件夹
# os.makedirs("static", exist_ok=True)
# app.mount("/static", StaticFiles(directory="static"), name="static")

# 配置模板目录
# 需要项目目录下有 templates 文件夹
# os.makedirs("templates", exist_ok=True)
# templates = Jinja2Templates(directory="templates")

# 如果不想创建目录，可以用这个简化版本：
from starlette.templating import _TemplateResponse
templates = Jinja2Templates(directory=".")

items_db = {
    1: {"name": "笔记本电脑", "price": 5999.0, "stock": 10},
    2: {"name": "机械键盘", "price": 299.0, "stock": 50},
    3: {"name": "鼠标", "price": 99.0, "stock": 100},
}


@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
    """渲染模板页面"""
    return templates.TemplateResponse(
        # 如果使用独立的 templates 目录，改为 "home.html"
        name="inline_template.html",
        context={
            "request": request,
            "title": "FastAPI 商品展示",
            "item_count": len(items_db),
            "items": items_db,
        },
    )

配套模板文件 inline_template.html（跟 .py 文件放一起）：

<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: 'Microsoft YaHei', Arial; padding: 30px;
               background: #f5f5f5; }
        h1 { color: #333; margin-bottom: 10px; }
        .subtitle { color: #666; margin-bottom: 30px; }
        .card { background: white; border-radius: 8px; padding: 20px;
                margin-bottom: 15px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
        .card h3 { color: #007bff; margin-bottom: 8px; }
        .card .price { font-size: 24px; color: #e74c3c; font-weight: bold; }
        .card .stock { font-size: 14px; color: #27ae60; }
        .count { background: #007bff; color: white; padding: 5px 15px;
                 border-radius: 20px; display: inline-block; }
        nav { margin: 20px 0; }
        nav a { color: #007bff; text-decoration: none; margin-right: 15px; }
        nav a:hover { text-decoration: underline; }
    </style>
</head>
<body>
    <h1>{{ title }}</h1>
    <p class="subtitle">
        共 <span class="count">{{ item_count }}</span> 件商品
    </p>
    <nav>
        <a href="/">首页</a>
        <a href="/api/items">API (JSON)</a>
        <a href="/docs">API 文档</a>
    </nav>
    <hr style="margin-bottom: 20px;">

    {% for id, item in items.items() %}
    <div class="card">
        <h3>
            <a href="/items/{{ id }}">{{ item.name }}</a>
        </h3>
        <div class="price">¥{{ "%.2f"|format(item.price) }}</div>
        <div class="stock">库存: {{ item.stock }} 件</div>
    </div>
    {% endfor %}
</body>
</html>

3.3 完整可运行示例

以下是一个同时提供 HTML 页面和 JSON API 的完整项目：

# complete_app.py — 完整可运行示例
"""一个同时提供 HTML 页面和 JSON API 的 FastAPI 应用

安装依赖：
    pip install fastapi uvicorn jinja2

运行：
    python complete_app.py
"""
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from typing import Optional
import asyncio
import uvicorn

# ==================== 应用初始化 ====================
app = FastAPI(title="FastAPI 完整示例")
templates = Jinja2Templates(directory=".")

# ==================== 数据模型 ====================
class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None
    stock: int = 0

# ==================== 模拟数据库 ====================
items_db = {
    1: {"name": "笔记本电脑", "price": 5999.0, "description": "高性能", "stock": 10},
    2: {"name": "机械键盘", "price": 299.0, "description": "RGB背光", "stock": 50},
    3: {"name": "鼠标", "price": 99.0, "description": "无线", "stock": 100},
}
next_id = 4

# ==================== HTML 页面路由 ====================
@app.get("/", response_class=HTMLResponse)
async def home_page(request: Request):
    """首页：商品列表"""
    return templates.TemplateResponse("inline_template.html", {
        "request": request,
        "title": "FastAPI 商品展示",
        "item_count": len(items_db),
        "items": items_db,
    })

@app.get("/items/{item_id}", response_class=HTMLResponse)
async def item_page(request: Request, item_id: int):
    """商品详情页"""
    item = items_db.get(item_id)
    if not item:
        return HTMLResponse("商品不存在", status_code=404)
    return templates.TemplateResponse("inline_detail.html", {
        "request": request,
        "item_name": item["name"],
        "item": item,
        "item_id": item_id,
    })

# ==================== JSON API 路由 ====================
@app.get("/api/items")
async def list_items():
    return list(items_db.values())

@app.get("/api/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    return items_db[item_id]

@app.post("/api/items")
async def create_item(item: Item):
    global next_id
    new_id = next_id
    next_id += 1
    items_db[new_id] = item.model_dump()
    return {"id": new_id, **item.model_dump()}

# ==================== 后台任务示例 ====================
@app.post("/api/orders")
async def create_order(item_id: int, quantity: int,
                       background_tasks: BackgroundTasks):
    """下单：立即返回，后台处理库存"""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")

    item = items_db[item_id]
    if item["stock"] < quantity:
        raise HTTPException(status_code=400, detail="库存不足")

    # 后台处理库存扣减
    background_tasks.add_task(update_stock, item_id, quantity)

    return {
        "status": "accepted",
        "message": f"已下单 {quantity} 件 {item['name']}，正在处理",
        "item_id": item_id,
        "quantity": quantity,
    }

async def update_stock(item_id: int, quantity: int):
    """后台扣减库存"""
    await asyncio.sleep(3)  # 模拟耗时处理
    items_db[item_id]["stock"] -= quantity
    print(f"库存已更新：{items_db[item_id]['name']} "
          f"剩余 {items_db[item_id]['stock']} 件")

# ==================== 启动入口 ====================
if __name__ == "__main__":
    print("访问以下地址：")
    print("  - 首页:      http://127.0.0.1:8000")
    print("  - API文档:   http://127.0.0.1:8000/docs")
    print("  - JSON API:  http://127.0.0.1:8000/api/items")
    uvicorn.run(app, host="127.0.0.1", port=8000)

配套的商品详情页模板 inline_detail.html：

<!DOCTYPE html>
<html>
<head>
    <title>{{ item_name }} - FastAPI</title>
    <style>
        body { font-family: 'Microsoft YaHei', Arial; max-width: 800px;
               margin: 0 auto; padding: 30px; background: #f5f5f5; }
        .detail { background: white; border-radius: 8px; padding: 30px;
                  box-shadow: 0 2px 8px rgba(0,0,0,0.1); }
        .name { font-size: 28px; color: #333; margin-bottom: 20px; }
        .price { font-size: 36px; color: #e74c3c; font-weight: bold; }
        .desc { color: #666; margin: 20px 0; line-height: 1.6; }
        .stock { color: #27ae60; font-size: 16px; }
        .back { display: inline-block; margin-top: 30px;
                color: #007bff; text-decoration: none; }
        .stock-low { color: #e67e22; }
        .stock-out { color: #e74c3c; }
        label, input { display: block; margin: 10px 0; }
        input { padding: 8px; width: 100px; }
        button { background: #007bff; color: white; border: none;
                 padding: 10px 20px; border-radius: 5px; cursor: pointer; }
        button:hover { background: #0056b3; }
    </style>
</head>
<body>
    <div class="detail">
        <h1 class="name">{{ item.name }}</h1>
        <div class="price">¥{{ "%.2f"|format(item.price) }}</div>
        <div class="desc">{{ item.description or '暂无描述' }}</div>
        {% set s = item.stock %}
        <div class="stock {% if s <= 0 %}stock-out{% elif s <= 20 %}stock-low{% endif %}">
            库存: {{ s }} 件
        </div>
        <a class="back" href="/">← 返回商品列表</a>
    </div>
</body>
</html>

四、异步编程深入

4.1 同步 vs 异步

# 同步：阻塞等待
time.sleep(2)      # 整个线程停住，什么都做不了

# 异步：非阻塞等待
await asyncio.sleep(2)  # 挂起当前任务，事件循环切换到其他任务

本质区别：

  • 同步：我等着，谁也别想动
  • 异步：我先挂着，你去忙别的，好了叫我

4.2 async 和 await 的关系

async def task():
    """async 声明这是一个协程函数，返回协程对象"""
    await asyncio.sleep(1)  # await 触发事件循环调度
  • async：声明这是协程函数（标识符）
  • await：真正让异步生效的关键（告诉事件循环"可以切换了"）

4.3 事件循环的工作机制

# ❌ 串行：逐个 await
result1 = await task(1)  # 任务1完成后再创建任务2
result2 = await task(2)
result3 = await task(3)

# ✅ 并发：用 gather
results = await asyncio.gather(
    task(1),  # gather 内部调用 create_task 注册所有任务
    task(2),
    task(3)
)

关键认知：await 确实说"可以去干别的"，但前提是得有别的可干。

串行（逐个 await）：
  创建任务1 → await → 检查：有其他任务吗？没有 → 等待
  任务1完成 → 创建任务2 → 检查：没其他任务 → 等待
  ...

并发（gather / create_task）：
  创建任务1 → 注册到事件循环
  创建任务2 → 注册到事件循环
  创建任务3 → 注册到事件循环
  任务1 await → 挂起 → 检查：有！任务2任务3已注册 → 切换
  ...

并发的关键是 create_task() 提前注册任务，而不是 await 本身。

4.4 协程 vs 线程 vs 进程

维度 线程 协程 差异倍数
内存占用 1-8 MB/个 1-4 KB/个 1000-2000x
创建时间 50-100 μs 1-5 μs 20-100x
切换时间 5-20 μs 0.5-2 μs 10x
调度者 操作系统内核 用户态事件循环 -
上限 几百~几千 几万~几十万 -

为什么协程更省资源？

线程（抢占式）：
  操作系统随时可能中断线程 → 需要完整保存寄存器、栈、页表
  → 每个线程必须分配独立栈空间（1-8 MB）
  → 切换开销大

协程（协作式）：
  只在 await 处切换 → 只需保存少量变量和执行位置
  → 共享线程栈空间，每个协程只需 ~4 KB
  → 切换开销极小

4.5 协程安全性：为什么不会数据覆盖

三个机制保证：

  1. 单线程交替执行：同一时刻只有一个协程在运行
  2. 独立栈帧：每个协程有自己的栈帧存储局部变量
  3. 状态保存：await 时保存完整执行状态，恢复时继续
线程栈空间：
┌──────────────┐
│ task1 栈帧   │ ← 当前激活，x=1
├──────────────┤
│ task2 栈帧   │ ← 已挂起，x=10（独立存储，不影响 task1）
└──────────────┘

五、WSGI vs ASGI

特性 WSGI ASGI
模式 同步 异步
协议支持 仅 HTTP HTTP + WebSocket + HTTP/2
应用签名 app(environ, start_response) app(scope, receive, send)
返回值 返回列表/迭代器 通过 send() 异步发送
代表框架 Flask、Django（旧） FastAPI、Starlette
代表服务器 Gunicorn、uWSGI Uvicorn、Daphne

关于"WSGI 是否只能处理一个请求"的澄清：WSGI 本身是同步的（一个 worker 一次处理一个请求），但可以通过多 worker 实现并发（如 gunicorn -w 4）。真正的差别不在于"能不能并发"，而在于如何实现并发：

WSGI 多进程：
  请求1 → worker 1 进程（阻塞等待）
  请求2 → worker 2 进程（阻塞等待）
  ...worker 数满后排队
  限制：1000 并发 ≈ 1000 进程 → 内存爆炸

ASGI 单进程：
  请求1 → 协程1 → await → 挂起 → 切请求2
  请求2 → 协程2 → await → 挂起 → 切请求3
  ...
  限制：10000 并发 ≈ 10000 协程 → 约 40 MB 内存

六、后台任务模式

# BackgroundTasks（FastAPI 内置，响应发送后执行）
background_tasks.add_task(long_running_task, task_id)
return {"status": "accepted"}  # 立即返回，任务在响应发送后执行

# create_task（手动控制，立即执行）
asyncio.create_task(long_running_task(task_id))
return {"status": "accepted"}  # 立即返回，任务在后台运行

BackgroundTasks 的工作原理：add_task 只是把任务添加到队列（耗时 0ms），FastAPI 在响应发送后才从队列取出任务执行。

七、系统性思考总结

7.1 完整请求链路

用户访问 http://127.0.0.1:8000/items/1
    ↓
[Uvicorn] 监听端口，收到 TCP 数据包
    ↓
[Uvicorn] httptools 解析 HTTP → 构建 ASGI scope 字典
    ↓
[Uvicorn] await app(scope, receive, send)
    ↓
[FastAPI] 匹配路由 /items/{item_id} → read_item
    ↓
[FastAPI] 提取路径参数 item_id=1 → 执行业务逻辑
    ↓
[FastAPI] 通过 send() 发送响应
    ↓
[Uvicorn] 将响应写回 TCP 连接
    ↓
浏览器收到数据

7.2 三个核心关系

关系 本质 类比
FastAPI ↔ ASGI 实现关系：实现了 ASGI 接口 厨师遵循菜单格式
Uvicorn ↔ ASGI 调用关系：按 ASGI 规范调用 服务员按菜单格式下单
FastAPI ↔ Uvicorn 间接关系：通过 ASGI 协议通信 厨师和服务员不直接耦合

7.3 学习要点

写 FastAPI 时：

  • 路由用 async def：异步执行，不阻塞事件循环
  • 路由用 def：FastAPI 自动用线程池执行
  • I/O 操作（数据库、API 调用）用 await
  • CPU 密集型操作用多进程

任务调度：

  • create_task 负责注册任务到事件循环
  • gather 是批量注册 + 批量等待
  • await 只是等待，不负责注册
  • BackgroundTasks 在响应发送后执行
# 学习 # fastapi
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

cover

LOS：面向大语言模型的可控、可验证、可执行操作系统

avatar openEuler 社区
cover

DLOS：基于双环验证内核的可控人工智能操作系统

avatar openEuler 社区
cover

DLOS AI 操作系统 v1.0：面向大语言模型的可验证、可控、可执行操作系统内核

avatar openEuler 社区