FastAPI+SQLAlchemy连接池遇16分钟延迟问题排查求助

问题背景 #

FastAPI搭建的后端采用SQLAlchemy管理数据库连接池，近期频繁出现API请求16分钟延迟，并发场景下尤为明显，请求会挂起约16分钟后才响应。无论使用create_engine还是create_async_engine配置连接池，问题均会出现。日志显示某请求从2024-09-12 10:10:24开始到10:26:23才响应（时长959秒），同时存在SQLAlchemy连接池检测到断开并失效连接的记录。

连接池及Session配置代码 #

engine = create_async_engine(
    URL,
    echo_pool=True,
    echo=False,
    poolclass=AsyncAdaptedQueuePool,
    pool_pre_ping=True,
    pool_size=64,
    max_overflow=64,
    pool_timeout=3,
    pool_recycle=3600
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    class_=AsyncSession,
    bind=engine,
    expire_on_commit=False,
)

async def get_db_session() -> AsyncSession:
    async with SessionLocal() as session:
        logger.debug(
            f"pid:{os.getpid()} | {session.get_bind().pool.status()} | {session.is_active} {session._is_asyncio}")
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_db_session)]

API接口代码 #

@router.get("", response_model=ResponseData)
@exception_handler
async def read_chat(session: SessionDep, current_user: CurrentUser, customer_id: int = None, page: int = 0, page_size: int = 10) -> Any:
    """
    Retrieve chats
    """
    chats = await crud_chat.get_user_chats(user_id=current_user.id, db_session=session, page=page-1, page_size=page_size)
    router_ids = list(set([chat.router_id for chat in chats]))
    customers = await crud_customer.get_customer_by_router_id(router_ids=router_ids, db_session=session)
    router_id_to_name = {customer.router_id: customer.name for customer in customers}
    router_id_to_avatar = {customer.router_id: customer.avatar for customer in customers}
    total_count = await crud_chat.get_user_chats_count(user_id=current_user.id, db_session=session)
    
    res = []
    for chat in chats:
        chat = chat.dict()
        chat["customer_name"] = router_id_to_name.get(chat["router_id"])
        chat["avatar"] = router_id_to_avatar.get(chat["router_id"])
        res.append(chat)

    return ResponseData(
        data={"chats": res,
              "count": len(chats),
              "total_count": total_count}
    )

CRUD操作代码 #

async def get_user_chats(self, user_id: int, db_session: AsyncSession = None, page: int = 0,
                         page_size: int = 10):
    async with get_db_session() if db_session is None else db_session as session:
        chats = await session.execute(
            select(Chat).where(Chat.uid == user_id).order_by(desc(Chat.ctime)).offset(page * page_size).limit(
                page_size))
        return chats.scalars().all()


async def get_customer_by_router_id(self, router_ids: list[int], db_session: AsyncSession = None):
    async with get_db_session() if db_session is None else db_session as session:
        customers = await session.execute(select(Customer).where(Customer.router_id.in_(router_ids)))
        return customers.scalars().all()


async def get_user_chats_count(self, user_id: int, db_session: AsyncSession = None):
    async with get_db_session() if db_session is None else db_session as session:
        response = await session.execute(
            select(func.count()).select_from(Chat).where(Chat.uid == user_id))
        return response.scalar_one_or_none()

排查方向及建议 #

• 修正CRUD中的Session复用逻辑：CRUD方法中对传入的db_session再次使用async with包裹，会导致AsyncSession上下文管理器重复进入，引发资源泄漏或连接池死锁。直接使用传入的session即可，无需二次上下文包裹：

  async def get_user_chats(self, user_id: int, db_session: AsyncSession = None, page: int = 0,
                           page_size: int = 10):
      # 复用传入的session，否则新建
      session = db_session
      if not session:
          session_gen = get_db_session()
          session = await anext(session_gen)
      try:
          chats = await session.execute(
              select(Chat).where(Chat.uid == user_id).order_by(desc(Chat.ctime)).offset(page * page_size).limit(
                  page_size))
          return chats.scalars().all()
      finally:
          # 仅关闭自己新建的session
          if not db_session:
              await session.close()
  

• 核对连接池与数据库超时配置：当前pool_recycle=3600（1小时），需确保小于数据库端的连接超时时间（如MySQL的wait_timeout），避免数据库主动回收连接后连接池仍认为连接有效。结合echo_pool=True的日志，对比连接失效回收时间与请求延迟时间点的关联。
• 排查连接池死锁场景：总连接数（pool_size+max_overflow）为128，若并发请求数超过该值，正常会在pool_timeout=3秒后超时，但16分钟延迟说明连接被占用后未释放。检查异常场景下Session是否未正确回收，结合get_db_session的日志确认每个请求的Session都正常关闭。
• 检查数据库端慢查询与锁等待：查看数据库慢查询日志，确认是否存在耗时极长的查询对应请求延迟的时间点；同时排查数据库是否存在表锁、行锁等待，导致连接被长时间占用。

内容的提问来源于stack exchange，提问作者BarryAllen