Python FastAPI 性能优化：异步协程与数据库连接池最佳实践

1. 异步协程的核心优化

• 避免阻塞操作：所有I/O密集型操作（网络请求/文件读写）必须使用异步库

  # 错误示例：同步阻塞
  def read_file():
      with open("data.txt") as f:  # 阻塞事件循环
          return f.read()
  
  # 正确示例：异步非阻塞
  async def async_read_file():
      async with aiofiles.open("data.txt") as f:
          return await f.read()
  

• 协程任务调度优化：

  • 使用asyncio.gather()并行执行独立任务

    async def fetch_data():
        task1 = asyncio.create_task(get_user())
        task2 = asyncio.create_task(get_products())
        user, products = await asyncio.gather(task1, task2)
    

  • 避免嵌套事件循环（禁止在协程内调用asyncio.run()）

2. 数据库连接池最佳实践

• 连接池配置公式：
  理想连接数 = (核心数 × 2) + 磁盘数
  例如4核SSD服务器：$$pool_size = (4 \times 2) + 1 = 9$$

• 异步驱动选择：

  数据库	推荐驱动	连接池实现
  PostgreSQL	asyncpg	asyncpg.pool.Pool
  MySQL	aiomysql	aiomysql.Pool
  SQLite	aiosqlite	内置连接池

• 连接池生命周期管理：

  from asyncpg import create_pool
  
  async def init_pool():
      return await create_pool(
          dsn="postgresql://user:pass@localhost/db",
          min_size=5,    # 最小空闲连接
          max_size=20,   # 最大连接数
          max_queries=500,  # 单个连接最大查询次数
          timeout=30     # 获取连接超时(秒)
      )
  

3. 集成实践示例

from fastapi import FastAPI, Depends
from asyncpg import Pool

app = FastAPI()
db_pool: Pool = None

@app.on_event("startup")
async def startup():
    global db_pool
    db_pool = await create_pool(...)  # 初始化连接池

@app.get("/users/{id}")
async def get_user(id: int):
    async with db_pool.acquire() as conn:  # 从池获取连接
        return await conn.fetchrow("SELECT * FROM users WHERE id=$1", id)

4. 性能监控指标

• 关键度量值：
  $$ \text{连接利用率} = \frac{\text{活跃连接数}}{\text{最大连接数}} \times 100% $$
  建议保持60%-80%区间
• 警报阈值：
  • 连接等待时间 > 100ms
  • 查询错误率 > 0.5%

5. 高级优化技巧

• 语句缓存：复用预编译SQL

  async with conn.transaction():
      stmt = await conn.prepare("UPDATE users SET status=$1 WHERE id=$2")
      await stmt.fetch("active", 123)  # 复用预编译语句
  

• 批量写入优化：

  records = [(1, "data1"), (2, "data2")]
  await conn.copy_records_to_table("logs", records=records)
  

压测建议

使用locust模拟高并发场景：

locust -f stress_test.py --users 1000 --spawn-rate 50

重点关注连接池的wait_queue指标，当队列持续增长时需要扩容连接池。

通过结合异步协程与智能连接池管理，FastAPI可支撑10K+ QPS的数据库密集型应用。实际部署时需根据$$ \lambda = \frac{\text{平均请求率}}{\text{平均处理时间}} $$动态调整连接池参数。