FastAPI WebSocket连接数据库时服务冻结问题排查与优化咨询

核心问题定位 #

服务器冻结、无法处理新WebSocket连接的根源，全部集中在request.py的数据库操作实现上：

1. 数据库连接管理逻辑错误 #

你当前的连接处理代码存在严重缺陷：

try:
    db = SessionLocal()
except:
    db.close()
else:
    # 数据库操作
    db.commit()
    db.commit()
    db.close()

• 若SessionLocal()创建失败（如连接池耗尽），except块调用db.close()时db变量未定义，会抛出NameError，导致连接资源无法回收。
• 两次db.commit()完全冗余，嵌套事务仅需一次提交，重复提交会导致事务状态异常，连接无法正常释放。
• 未处理数据库操作中的异常（如查询/插入失败），异常发生时连接不会关闭，最终耗尽整个连接池。

2. 同步数据库操作阻塞异步事件循环 #

request_handler是异步函数，但你调用的是同步版SQLAlchemy的Session操作。异步上下文里执行同步IO（数据库查询/写入）会直接阻塞FastAPI的事件循环，导致所有WebSocket连接、API请求彻底无法处理——这是服务器冻结的最核心原因，移除数据库操作后恢复正常，就是因为消除了同步IO阻塞。

3. 过度严格的表锁导致阻塞 #

你使用的LOCK TABLE friend_requests IN ACCESS EXCLUSIVE MODE;是最严格的表锁，会阻塞该表的所有读写操作。只要有一个请求持有该锁，其他所有涉及该表的请求都会被卡住；若锁因异常未释放，会导致永久阻塞。

正确实现方案 #

一、改用异步SQLAlchemy #

在异步上下文（WebSocket处理函数）中操作数据库，必须使用异步SQLAlchemy避免阻塞事件循环：

1. 修改数据库配置为异步引擎：

# db.py 示例
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

2. 用上下文管理器管理异步连接：

# request.py 修正后的代码
from sqlalchemy import select
from datetime import datetime

async def request_handler(fromID, msg, conMan):
    try:
        async with AsyncSessionLocal() as db:
            async with db.begin():
                # 异步查询替代同步查询
                request_result = await db.execute(
                    select(Friend_Request)
                    .where(Friend_Request.sender_id.in_([fromID, msg['toID']]))
                    .where(Friend_Request.recipient_id.in_([fromID, msg['toID']]))
                )
                request = request_result.scalar_one_or_none()

                friendship_result = await db.execute(
                    select(Friendship)
                    .where(Friendship.user_id.in_([fromID, msg['toID']]))
                    .where(Friendship.new_friend_id.in_([fromID, msg['toID']]))
                )
                friendship_entry = friendship_result.scalar_one_or_none()

                blocked_result = await db.execute(
                    select(Blocked_User)
                    .where(Blocked_User.sender_id.in_([fromID, msg['toID']]))
                    .where(Blocked_User.recipient_id.in_([fromID, msg['toID']]))
                )
                blocked_entry = blocked_result.scalar_one_or_none()

                if not (request or friendship_entry or blocked_entry):
                    sender = await db.get(User, fromID)
                    receiver = await db.get(User, msg['toID'])
                    
                    current_time = datetime.now()
                    request_obj = Friend_Request(
                        sender_id=sender.id, sender_firstname=sender.firstname,
                        sender_lastname=sender.lastname, sender_username=sender.username,
                        time_sent=current_time, recipient_id=receiver.id
                    )
                    db.add(request_obj)

                    message = {
                        'event': 'request',
                        'sender-id': sender.id,
                        'sender-firstname': sender.firstname.title(),
                        'sender-lastname': sender.lastname.title(),
                        'sender-username': sender.username,
                        'time': str(current_time),
                        'confirmed': False,
                        'deleted': False
                    }

                    await conMan.send_personal_message(msg['toID'], message)
    except Exception as e:
        # 记录异常日志，根据需要处理错误
        print(f"好友请求处理失败: {str(e)}")

二、移除手动表锁，用数据库约束替代 #

手动锁表风险极高，可在Friend_Request模型中添加复合唯一约束，防止重复请求：

# models/friend_request.py
from sqlalchemy import Column, Integer, String, DateTime, UniqueConstraint

class Friend_Request(Base):
    __tablename__ = "friend_requests"
    id = Column(Integer, primary_key=True, index=True)
    sender_id = Column(Integer, nullable=False)
    recipient_id = Column(Integer, nullable=False)
    # 其他字段...
    
    # 复合唯一约束：同一对用户之间只能存在一个好友请求
    __table_args__ = (
        UniqueConstraint('sender_id', 'recipient_id', name='unique_friend_request'),
    )

重复插入时数据库会抛出唯一约束异常，捕获该异常即可，无需手动锁表。

三、WebSocket依赖中使用异步数据库连接 #

修改main.py的依赖，适配异步Session：

# main.py 修正后的get_user依赖
from sqlalchemy.ext.asyncio import AsyncSession
from db import get_db

async def get_user(websocket: WebSocket, db: AsyncSession = Depends(get_db)):
    token = websocket.cookies.get("sessionid")
    if not token:
        return None
    try:
        token_data = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user = await crud.get_user_by_id(db=db, id=token_data["id"])
        if not user or not user.is_active:
            return None
        current_time = datetime.now()
        if datetime.strptime(str(token_data["validity_period"]), "%Y-%m-%d %H:%M:%S.%f") < current_time:
            return None
        return token_data["id"]
    except JWTError:
        return None

# 对应的异步get_db依赖
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

关键总结 #

1. 异步上下文（WebSocket、异步路由）中必须使用异步SQLAlchemy，禁止调用同步数据库操作，否则会阻塞事件循环导致服务器冻结。
2. 用上下文管理器（with语句）管理数据库连接，确保连接自动关闭，避免资源泄漏。
3. 用数据库约束（唯一键、外键）替代手动锁表，保证数据一致性的同时避免阻塞。
4. 捕获所有异常，防止未处理异常导致连接资源无法回收。

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