线程 IO 阻塞模拟

1import time
2
3@router.get("/blocking")
4async def blocking():
5    """IO 阻塞模拟"""
6    time.sleep(10)
7    return {"message": "This is a simulated blocking IO"}

FastAPI 并发处理

FastAPI 是基于异步编程的，能够有效处理并发请求，但当涉及到 I/O 操作时，要特别注意以下几个方面：

注意事项：

异步操作的优势：确保所有的 I/O 操作（如数据库查询、网络请求、文件操作）都使用异步版本的库，这样可以避免阻塞事件循环，提高并发处理能力。
高并发处理：对于大并发的应用，可以配置 uvicorn 服务器的 workers 参数，增加处理请求的进程数。也可以使用负载均衡器将流量分发到多个应用实例。

启动 FastAPI 应用时，可以使用多个进程：

1uvicorn app:app --workers 4

任务队列：对于一些长时间运行的任务，考虑将其放入任务队列中，例如使用 Celery，避免阻塞主应用的响应。
热重载（–reload）和多 worker 模式互斥。如果使用了 –reload，worker 将被禁用。

在使用 FastAPI 进行数据库操作、网络请求、文件上传和下载时，以下几点至关重要：

使用异步库：确保数据库、网络请求和文件操作使用异步版本的库（如 asyncpg、httpx、aiofiles）。
避免阻塞事件循环：通过异步操作避免阻塞主线程，确保 FastAPI 能够高效处理并发请求。
流式操作：对于大文件的上传和下载，使用流式读取和写入，避免占用过多内存。
后台任务：使用 BackgroundTasks 或外部任务队列处理耗时操作，不阻塞请求响应。
性能优化：考虑并发、负载均衡和异步队列等措施，提升系统的吞吐量和稳定性。

为什么 I/O 操作可能导致阻塞？

I/O 操作通常涉及等待外部资源（如文件、网络、数据库等）提供数据或响应。在这段等待时间里，程序无法继续执行其他任务，直到 I/O 操作完成。这种“等待”就会导致阻塞。

例如：

文件操作：读取或写入文件时，操作系统会与磁盘进行交互，这可能会消耗一定时间，尤其是在处理大文件或在慢速存储设备上执行时。
网络请求：向远程服务器发送请求时，可能需要等待服务器响应，尤其是当网络较慢或服务器处理请求时间较长时。
数据库查询：查询数据库时，数据库服务器需要处理请求并返回数据，这也需要一些时间，尤其是在查询复杂或数据量大的情况下。

如何避免 I/O 操作阻塞？

使用异步 I/O：通过 asyncio 等异步框架，可以在等待 I/O 操作时不阻塞主线程，从而提高程序的并发性和响应速度。例如，FastAPI 本身就是异步框架，能够有效处理网络请求和文件操作。

1import aiofiles
2
3async with aiofiles.open('file.txt', 'r') as f:
4    content = await f.read()

后台任务：对于耗时的任务，可以将其放到后台处理，而不阻塞当前请求。FastAPI 提供了 BackgroundTasks 来实现后台任务。

1from fastapi import BackgroundTasks
2
3def process_file(file_path: str):
4    # 文件处理的耗时任务
5    pass
6
7@app.post("/upload")
8async def upload_file(background_tasks: BackgroundTasks, file: UploadFile):
9    background_tasks.add_task(process_file, file.filename)
10    return {"message": "File uploaded and being processed in background"}

使用多进程或多线程：对于一些 CPU 密集型任务或者无法改成异步的 I/O 操作，可以通过多进程（multiprocessing）或者多线程（threading）的方式将耗时操作分离出去。

FastAPI StreamingResponse 控制 nginx 流式不缓存

1headers = {
2    "Cache-Control": "no-cache",
3    "Connection": "keep-alive",
4    "X-Accel-Buffering": "no",
5}
6return StreamingResponse(event_stream(), media_type="text/event-stream", headers=headers)

Nginx 会对该接口禁用缓冲，结合 proxy_http_version 1.1 和 proxy_set_header Connection “”，“proxy_buffering off”，即可实现真正的实时流式推送。

FastAPI Schema 访问属性

1class SQLSchema(BaseModel):
2    sql: str = "SELECT 1"
3
4async def test(request: SQLSchema):
5    print(request.sql)

需要用到 Pydantic 来构建请求体和请求头（请求头字段名是不区分大小写的），否则 docs 文档中没有相关的信息请求

FastAPI form-data 请求参数

1@router.post("/webFileUploadToCloud")
2async def web_file_upload_to_cloud(
3    table_name: Literal['milvus_case', 'milvus_course', 'milvus_terms'] = Form(...),
4    kind: Literal['doc', 'ppt', 'video', 'pdf'] = Form(...),
5    db: Session = Depends(get_db), file: UploadFile = File(...)):

这样写会在 docs 中可以手动选择这些值

FastAPI 离线使用 docs Swagger 文档

1app.mount("/static", StaticFiles(directory="static"), name="static")
2
3@app.get("/hr/model/api/docs", include_in_schema=False)
4async def custom_swagger_ui_html():
5    return get_swagger_ui_html(
6        openapi_url=app.openapi_url,
7        title="Custom Swagger UI",
8        swagger_js_url="/static/swagger/swagger-ui-bundle.js",
9        swagger_css_url="/static/swagger/swagger-ui.css",
10        swagger_favicon_url="/static/img/favicon.png"
11    )

FastAPI 未定义在函数里的代码执行 2 次

ASGI Server（如 Uvicorn）的热加载机制

当开发环境中启用了 --reload 参数时，ASGI 服务器会开启自动热重载。

热重载的实现方式是：

1- 启动一个主进程来监视文件变动。
2- 启动一个子进程运行应用程序代码。
3- 如果文件发生变动，重新加载子进程。

因此，文件会被加载两次：

1- 一次在主进程中。
2- 一次在子进程中。

2. ASGI Worker 多实例运行

如果你在启动服务器时指定了多个 workers，每个 worker 都会加载一次应用文件，从而导致全局代码被执行多次。

模块加载特性

Python 脚本中的顶层代码（未封装在函数或类中）会在模块被导入时直接执行。

FastAPI 应用实例通常是顶层代码的一部分，因此在不同进程加载应用时会触发两次执行。

FastAPI GET 类型数组字段传参

1from fastapi import FastAPI, Query
2from typing import List
3
4app = FastAPI()
5
6@app.get("/intelligence/ai/api/generate-article")
7async def generate_article(
8    session_id: str,
9    article_ids: List[int] = Query(...)
10):
11    return {"session_id": session_id, "article_ids": article_ids}

1GET /intelligence/ai/api/generate-article?session_id=123123123&article_ids=49838&article_ids=49839

传统多值参数

.env 文件重载策略

在 FastAPI 项目中，.env 文件的内容通常会在应用启动时通过库（如 python-dotenv 或 pydantic 的BaseSettings）加载到环境变量中。如果 .env 文件发生了更改，通常需要重启应用以使更改生效。

.env 文件的加载方式：

.env 文件通常在应用启动时加载，例如在 FastAPI 项目的 settings 模块中使用 pydantic.BaseSettings 或手动通过 dotenv 加载。
加载后的变量会存储在环境变量中（如 os.environ），不会自动监控文件的变化。

重载机制不会重新加载环境变量：

FastAPI 的 –reload 选项使用 Uvicorn 提供的自动重载功能，只会监控文件的修改并重启应用，但它并不会重新加载环境变量。

如果想在不重启应用的情况下应用新的 .env 文件内容，可以手动重新加载环境变量。

1from dotenv import load_dotenv
2import os
3
4def reload_env():
5    """
6    手动重新加载 .env 文件
7    """
8    load_dotenv()  # 重新加载 .env 文件
9    print("已重新加载 .env 文件")
10    print(f"当前环境变量: {os.environ.get('SOME_KEY')}")  # 测试加载结果

uvicorn 关闭服务

如果希望在某个请求时优雅地关闭 FastAPI 应用，可以通过 uvicorn 提供的信号机制来实现。uvicorn 支持通过发送 SIGTERM 信号来优雅关闭服务。

1from fastapi import FastAPI
2import uvicorn
3import os
4import signal
5
6app = FastAPI()
7
8@app.get("/shutdown")
9async def shutdown():
10    os.kill(os.getpid(), signal.SIGTERM)  # 发送终止信号，优雅关闭服务
11    return {"message": "Server shutting down..."}
12
13if __name__ == "__main__":
14    uvicorn.run(app, host="0.0.0.0", port=8000)

FastAPI 启动多 worker 模式

1uvicorn.run("main:app", host="0.0.0.0", port=9101, reload=False, log_config=None, workers=4)

如果设置了 –workers，日志中会显示类似以下内容

1INFO: Started server process [PID1]
2INFO: Started server process [PID2]

热重载默认只支持单个 worker，因为多进程不支持动态重启。

热重载（–reload）和多 worker 模式互斥。如果使用了 –reload，worker 将被禁用。