本文解析 sqlite3 在多请求并发场景下因缺乏行级锁和事务隔离导致的“读取过期值”问题，并提供基于原子操作、合理表结构设计的可靠解决方案。
本文解析 sqlite3 在多请求并发场景下因缺乏行级锁和事务隔离导致的“读取过期值”问题，并提供基于原子操作、合理表结构设计的可靠解决方案。
在使用 SQLite3 构建轻量级 Flask 服务时，一个常见但极易被忽视的陷阱是：在高并发请求下（如压测中 20 用户并行调用 addTask/deleteTask），看似顺序执行的“查–改–写”逻辑会因竞态条件（race condition）导致数据丢失或读取陈旧值。您观察到的现象——addTask 成功插入 237 后，紧随其后的 deleteTask 却无法在 tasks_order 数组中找到该 ID——并非数据库缓存延迟，而是典型的 Read-Modify-Write（RMW）非原子性问题。
? 问题根源：非原子的“查–改–写”流程
您的当前逻辑：
tasks_order = dbm.getTasksOrderList() # Step 1: SELECT → JSON 解析 tasks_order.append(new_id) # Step 2: Python 内存修改 dbm.updateTasksOrderList(tasks_order) # Step 3: UPDATE → JSON 序列化写入
这三步跨越了多个数据库交互，中间无锁保护。当两个请求并发执行时，极易发生如下时序（时间向下流动）：
| Request 1 | Request 2 |
|---|---|
| SELECT → [] | |
| append(237) → [237] | SELECT → [] |
| UPDATE → [237] | append(456) → [456] |
| UPDATE → [456] ← 覆盖了 237！ |
结果：237 被静默丢弃，后续 DELETE 自然失败。SQLite 默认的 DEFERRED 事务隔离级别不阻止并发读取同一行，SELECT 不加锁，UPDATE 仅在执行瞬间锁定，无法保证“读取后立即更新”的语义。
✅ 正确解法一：强制行级写锁（适用于必须保留单行 JSON 结构的场景）
若因历史原因需维持 tasks_order 单行表结构，请将整个 RMW 操作置于同一事务，并显式加锁：
def updateTasksOrderListSafely(task_id, operation="add"):
conn = db.getConn()
cur = conn.cursor()
try:
# 关键：SELECT ... FOR UPDATE 确保本事务独占该行
cur.execute("SELECT order_list FROM tasks_order WHERE rowid = 1 FOR UPDATE")
row = cur.fetchone()
if not row:
raise ValueError("tasks_order table is empty")
order_list = json.loads(row["order_list"])
if operation == "add":
order_list.append(task_id)
elif operation == "delete" and task_id in order_list:
order_list.remove(task_id)
else:
raise ValueError(f"Task {task_id} not found for deletion")
cur.execute("UPDATE tasks_order SET order_list = ? WHERE rowid = 1",
(json.dumps(order_list),))
conn.commit()
return order_list
except Exception as e:
conn.rollback()
raise e
⚠️ 注意：FOR UPDATE 在 SQLite 中仅在 WAL 模式下有效（推荐启用），且需确保所有访问均走同一连接对象（避免 getConn() 多次调用创建新连接）。Flask 的 g 全局变量已满足此要求。
✅ 正确解法二（强烈推荐）：重构为符合关系模型的规范设计
将 tasks_order TEXT 单行 JSON 字段彻底废弃，改用标准关系表：
CREATE TABLE tasks (
id INTEGER PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 可选：添加唯一约束防止重复插入
CREATE UNIQUE INDEX idx_task_id ON tasks(id);
此时，增删操作天然具备原子性与并发安全性：
# addTask → 原子插入
def addTask():
new_id = random.randint(1, 999)
conn = db.getConn()
cur = conn.cursor()
try:
cur.execute("INSERT INTO tasks (id) VALUES (?)", (new_id,))
conn.commit()
return {"id": new_id}
except sqlite3.IntegrityError:
# 处理重复 ID（可重试或生成新 ID）
raise
# deleteTask → 原子删除
def deleteTask(task_id):
conn = db.getConn()
cur = conn.cursor()
try:
# 返回影响行数，确认是否删除成功
cur.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
if cur.rowcount == 0:
raise ValueError(f"Task {task_id} not found")
conn.commit()
return {}, 204
except Exception as e:
conn.rollback()
raise e
获取当前顺序（按插入时间）：
SELECT id FROM tasks ORDER BY created_at;
✅ 优势：
- 零竞态：INSERT/DELETE 是 SQLite 原子 DML 操作，自动加锁；
- 可扩展：支持索引优化、分页、条件查询；
- 可维护：无需手动序列化/反序列化 JSON，规避格式错误风险；
- 符合范式：每个任务独立成行，语义清晰。
? 总结与最佳实践
- 永远避免“查–改–写”跨事务操作：若必须，务必使用 SELECT ... FOR UPDATE + 显式事务包裹；
- 优先采用规范化表结构：单行 JSON 存储顺序是反模式，应让数据库管理关系而非应用层拼接数组；
-
启用 WAL 模式提升并发（在连接初始化时）：
conn.execute("PRAGMA journal_mode = WAL") - 日志调试技巧：在关键路径打印 threading.get_ident() 或 request.id，可清晰追踪并发线程/请求交织；
- 生产环境务必设置超时与重试：对 sqlite3.OperationalError（如数据库忙）做指数退避重试。
通过以上重构，您的 addTask/deleteTask 将在任意并发压力下稳定运行，彻底告别“刚插入就查不到”的诡异现象。
