先查后写仍然报重复冲突，SQLite 并发写入的排查记录

2026 年 7 月 9 日1510 字约 6 分钟

摘要代码的逻辑看似没有问题：先用 SELECT 查一下，没有就 INSERT，有就 UPDATE。但在并发请求下，SQLite 的唯一索引仍然报错。本文介绍如何用原生 Upsert 替代先查后写。

在 Auto Email Sender 的诊断日志中，我看到了几条错误堆栈：

sqlite3.IntegrityError: UNIQUE constraint failed: llm_adapter_cache.endpoint, llm_adapter_cache.model_name

出错的是一张本地缓存表，用于保存 LLM 模型能力的自适应探测结果。

我的第一反应是：“这不可能。我在写入之前明明做了检查。”

翻开当时的代码，逻辑很简单，和教科书上的写法一样：

# 当年的写法
cached = db.query(LLMAdapterCache).filter_by(endpoint=ep, model_name=model).first()
if cached:
    cached.config = new_config
    cached.updated_at = now()
else:
    new_record = LLMAdapterCache(endpoint=ep, model_name=model, config=new_config)
    db.add(new_record)
db.commit()

“先查一下，有就更新，没有就插入。”写过业务代码的人大多写过类似的逻辑。

db.query 已经确认记录不存在，为什么几毫秒之后的 db.add 还会撞上唯一索引约束（Unique Constraint）？

检查与写入之间的时间差

原因是：这段代码在单线程下没有问题，但在并发时会出错。

通俗比喻：抢自习室座位

为什么“先查一下，没有再新建”在并发时一定会出问题？假设你去自习室占座：

你（线程 A）看了一眼 1 号桌，没有人；

几乎同一时间，另一个人（线程 B）也看了一眼 1 号桌，同样没有人；

你去洗手间，回来把书包放在桌上（写入成功）；

另一个人还记得两秒前“桌子是空的”，于是也把电脑放到了 1 号桌上。

两个人肯定会发生冲突。在数据库里，这就是抛出 UNIQUE constraint failed。 “我刚才看到没人占”只代表过去的情况，不代表下一毫秒没有人插队。 在多任务并发时，如果检查和写入不是一个原子操作，中间就存在竞争窗口。

软件在批量生成邮件草稿时，前端会同时启动多个任务，向同一个大模型网关发起调用。这就产生了一个典型的竞态条件，在安全和并发领域被称为 TOCTOU（Time-of-Check to Time-of-Use，检查时与使用时不一致）。

把时间线放大到毫秒级，大致是这样：

时间	协程 Worker A	协程 Worker B
T1	执行 SELECT：数据库为空，返回 `None`
T2		执行 SELECT：Worker A 还没有写入，同样返回 `None`
T3	执行 INSERT 并 commit：写入成功
T4		根据 T2 的结果，也执行 INSERT 并 commit
T5	正常结束	报错：UNIQUE constraint failed

检查的结果只代表过去

Worker B 在 T2 时查到的“记录不存在”是当时的真实情况。但在 T3 时，Worker A 已经写入了这条记录。 “查询”和“写入”是两次独立的数据库操作，中间存在竞争窗口。没有锁或原子事务保护的“先查后写”，在并发下并不安全。

捕获异常后再查询，会让事务出问题

遇到这种报错，很多人的第一反应是加一个 try-catch：

try:
    insert_record()
except IntegrityError:
    db.rollback()
    # 既然别人已经插入了，那我重新查出来做更新
    record = query_again()
    update_record(record)

这种写法虽然能运行，但有两个明显的问题：

把异常当成常规分支： 在并发场景下，唯一索引冲突是经常发生的情况。用异常回滚来处理常规业务，代码难读，也会拖慢性能；
SQLAlchemy 的事务状态会被污染： 一旦底层的 flush() 抛出 IntegrityError，整个 Session 就会进入 inactive 或 failed 状态。在嵌套事务或复杂的业务中，一次草率的 rollback() 可能会把外层尚未提交的其他正常数据一并回滚。

数据库已经提供了合适的写法

这是一个典型的“不存在则插入，存在则覆盖”的场景，没有必要在应用层做两次数据库往返。SQLite 从 3.24.0 版本开始支持 Upsert 语法（ON CONFLICT DO UPDATE）。

用原生 Upsert 交给数据库处理

让数据库在一条原子语句中完成判断，是更合适的做法。

在 SQLAlchemy Core 中，可以直接使用 SQLite 方言的语法：

from sqlalchemy.dialects.sqlite import insert
 
stmt = insert(LLMAdapterCache).values(
    endpoint=ep,
    model_name=model,
    config=new_config,
    updated_at=func.now()
)
 
# 一条原子语句完成冲突处理
upsert_stmt = stmt.on_conflict_do_update(
    index_elements=['endpoint', 'model_name'], # 判断冲突的唯一键
    set_={
        'config': stmt.excluded.config,
        'updated_at': stmt.excluded.updated_at
    }
)
 
db.execute(upsert_stmt)
db.commit()

这次修改带来了以下效果：

真正的原子性： 插入和更新发生在同一个 B-Tree 事务锁内部，并发请求无法插进检查和写入之间的空隙；
减少一次查询： 不再需要多余的 SELECT，一条 SQL 就完成了整个流程；
日志更干净： 不再出现 IntegrityError 的异常堆栈，真正不可预期的数据库故障也更容易被发现。

注意 ORM Identity Map 中的旧数据

使用 Core 级别的 on_conflict_do_update 绕过 ORM、直接执行底层 SQL 时，需要注意：如果当前 Session 之前已经加载过对应的 ORM 对象，SQLAlchemy 的一级缓存（Identity Map）中保存的仍然是旧数据。如果后续代码还要使用这个 ORM 对象，需要显式调用 db.refresh(obj) 或 db.expire(obj)，否则会读到过期的数据。

什么时候不适合覆盖式 Upsert

需要说明的是：原生 Upsert 并不能替代所有的“先查后写”。

这次能直接使用 ON CONFLICT DO UPDATE，是因为：

这是一张自适应缓存表，新配置覆盖旧配置（Last-Write-Wins）符合业务预期；
冲突后没有多表的级联依赖，也没有严格的状态机流转。

但如果业务是下面这样，覆盖式 Upsert 就不适用了：

银行账户的扣款和入账： 应该在数据库内做原子累加，例如 balance = balance - 100，而不是直接覆盖；
发信任务的状态机： 任务必须按 pending -> sending -> sent 的顺序单向流转，已经处于 sent 的任务，绝不能被迟到的并发请求通过 Upsert 改回 sending。

这类场景应该使用乐观锁的版本号（Version Column），或者带条件的更新，例如 WHERE status = 'pending'。

总结

编写涉及并发的存储层代码时，要避免“只在单线程中运行”的假设：

不要依赖“刚才查过”的结果： 在多线程和分布式环境中，之前查到的状态随时可能已经改变；
把原子性交给存储引擎： 能用一条 SQL（Upsert、CAS、条件更新）解决的事情，就不要在应用层分多步处理；
区分幂等覆盖和状态流转： 缓存适合直接覆盖，状态机则需要用条件阻止非法的状态变化。

检查与写入之间的时间差

捕获异常后再查询，会让事务出问题

用原生 Upsert 交给数据库处理

什么时候不适合覆盖式 Upsert

总结

评论