Flask-SQLAlchemy搭配多gunicorn worker读不一致引发唯一约束冲突如何解决

你遇到的是典型的先查后写（Check-Then-Act）竞态条件问题，不管是初始化root用户还是普通用户同名提交场景，本质都是读操作和写操作之间有时间差，并发请求同时拿到了相同的读结果，进而触发重复写入冲突，需要从多个层面配合解决。

优先按优先级选择以下方案：

• 方案1：将初始化逻辑从worker进程中剥离。多worker部署时不要让每个worker都自行执行初始化逻辑，改成部署启动流程里先单独执行一次初始化脚本，确认初始化完成后再启动gunicorn worker集群，从根源上避免多个worker同时初始化的问题，这是该场景最推荐的解法。
• 方案2：加分布式锁控制初始化流程。如果无法剥离初始化逻辑，可以引入基于数据库、Redis的分布式锁，只有拿到锁的worker能执行初始化操作，其他worker等待锁释放后再查询一次数据状态，就不会触发重复写入。
• 方案3：异常捕获兜底兼容。如果不想改动现有初始化流程，可以对写入逻辑加唯一约束异常捕获，代码改造成本最低，示例如下：

from sqlalchemy.exc import IntegrityError

root_user = User.query.filter_by(username='root').one_or_none()
if not root_user:
    try:
        new_user = User(username="root", password_hash="SomeLongAndSecurePasswordHash")
        new_user.roles = [serveradmin_role]
        db.session.add(new_user)
        db.session.commit()
    except IntegrityError:
        # 唯一约束冲突说明其他进程已完成写入，直接回滚当前会话即可
        db.session.rollback()

注意：捕获异常后必须调用db.session.rollback()，否则当前会话会处于失效状态，后续所有数据库操作都会报错。

三层防护配合处理：

• 第一层：数据库层面必须加唯一约束。username字段一定要设置UNIQUE约束，这是最后一道数据一致性防线，就算上层逻辑漏判，数据库也会拦住重复写入。
• 第二层：业务代码层做异常捕获兼容。和初始化逻辑类似，用户注册的写入逻辑要捕获IntegrityError，捕获后直接给前端返回「用户名已存在」的提示即可，不要直接抛出异常导致接口500报错。
• 第三层：高并发场景可选加行级锁。如果业务并发量很高，可以用数据库的FOR UPDATE行锁在查询阶段就锁住对应范围，避免后续并发写入冲突，示例如下：

# 开启事务后加排他锁查询，只有拿到锁的请求能继续执行
exist_user = User.query.filter_by(username=request_username).with_for_update().one_or_none()
if not exist_user:
    new_user = User(username=request_username, password_hash=hash_val)
    db.session.add(new_user)
    db.session.commit()

如果继续使用SQLite做多worker部署，建议开启WAL模式（执行PRAGMA journal_mode=WAL;），能大幅提升多进程并发读写的性能，减少锁冲突概率。如果业务并发量持续走高，更建议替换为MySQL、PostgreSQL这类生产级关系型数据库，SQLite本身设计定位就是单进程低并发场景，多worker高并发下性能和锁支持能力都比较有限。

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