本文深入解析 sqlite3 在多请求并发场景下因缺乏行级锁和事务隔离导致的“读取过期值”问题，并提供从 json 数组反模式到规范化表结构的完整重构方案。
本文深入解析 sqlite3 在多请求并发场景下因缺乏行级锁和事务隔离导致的“读取过期值”问题，并提供从 json 数组反模式到规范化表结构的完整重构方案。
在 Flask 应用中使用 SQLite3 时，若多个请求（如 addTask 和 deleteTask）并发操作同一行数据（例如仅含一个 JSON 字段的 tasks_order 表），极易触发竞态条件（Race Condition），导致数据丢失或逻辑错误——这正是你日志中 237 被成功插入却在后续删除时“消失”的根本原因。
? 问题根源：无锁读-改-写（Read-Modify-Write）流程
你的当前逻辑本质上是典型的 RMW 反模式：
# 伪代码示意（非原子操作） tasks_order = getTasksOrderList() # 1. SELECT → 得到 [54, 950, ...] tasks_order.append(new_id) # 2. Python 内存修改 updateTasksOrderList(tasks_order) # 3. UPDATE → 写回整个数组
该流程不是原子的。当两个请求几乎同时执行时，会发生如下竞争：
| 时间 | Request 1 | Request 2 |
|---|---|---|
| t₁ | SELECT → [54, 950] | |
| t₂ | SELECT → [54, 950] | |
| t₃ | append(237) → [54, 950, 237] | append(151) → [54, 950, 151] |
| t₄ | UPDATE → [54, 950, 237] | UPDATE → [54, 950, 151] ✅ 覆盖了前一次写入 |
结果：237 被永久丢弃，deleteTask 自然无法找到它 —— 这并非 SQLite “看不见提交”，而是两次独立事务各自读取了旧快照，并各自覆盖写入，后写者胜出。
⚠️ 注意：SQLite 默认 isolation_level=None（即 autocommit 模式），且 SELECT 不加锁。SELECT ... FOR UPDATE 在 SQLite 中不被支持（仅适用于 WAL 模式下的某些变体，且行为受限），因此不能依赖此方案修复。
✅ 正确解法：摒弃 JSON 数组，采用关系型建模
将单行 JSON 数组重构为标准关系表，是彻底规避竞态、提升可维护性与性能的唯一推荐方案：
1. 重建数据库表结构
-- 删除旧表（谨慎操作！）
DROP TABLE IF EXISTS tasks_order;
-- 创建规范任务表（支持自然排序与高效增删）
CREATE TABLE tasks (
id INTEGER PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-- 可选：添加显式 order_index 列用于自定义拖拽排序
order_index INTEGER UNIQUE NOT NULL DEFAULT 0
);
2. 重写服务逻辑（原子、无竞态）
# addTask：单条 INSERT，天然线程安全
def addTask(hash):
try:
conn = db.getConn()
new_task_id = random.randint(1, 999)
# 原子插入，无需读取-修改-写入
conn.execute("INSERT INTO tasks (id) VALUES (?)", (new_task_id,))
conn.commit()
print(f"{hash} added {new_task_id}")
return {"id": new_task_id}
except Exception as e:
conn.rollback()
raise e
# deleteTask：单条 DELETE，同样原子
def deleteTask(hash, task_id):
try:
conn = db.getConn()
# 直接删除，不存在竞态风险
cursor = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
if cursor.rowcount == 0:
raise ValueError(f"Task {task_id} not found")
conn.commit()
print(f"{hash} deleted {task_id}")
return {}, 204
except Exception as e:
conn.rollback()
raise e
3. 获取有序任务列表（按插入时间或自定义序号）
# 按插入时间排序（默认保序）
def getTasksInOrder():
cur = db.getConn().cursor()
cur.execute("SELECT id FROM tasks ORDER BY created_at")
return [row["id"] for row in cur.fetchall()]
# 或按自定义 order_index 排序（支持前端拖拽重排）
def getTasksByCustomOrder():
cur = db.getConn().cursor()
cur.execute("SELECT id FROM tasks ORDER BY order_index")
return [row["id"] for row in cur.fetchall()]
?️ 额外加固建议（如需自定义排序）
若用户需自由拖拽调整任务顺序，引入 order_index 列并配合 原子化更新：
# 更新某任务到指定位置（示例：移到索引 2）
def moveTaskToPosition(task_id, target_index):
conn = db.getConn()
try:
# 使用 SQLite 的 UPDATE ... CASE 实现批量重排（避免多次 UPDATE）
conn.execute("""
UPDATE tasks
SET order_index = CASE
WHEN id = ? THEN ?
WHEN order_index >= ? THEN order_index + 1
ELSE order_index
END
""", (task_id, target_index, target_index))
conn.commit()
except Exception as e:
conn.rollback()
raise e
✅ 总结
- ❌ 不要在 SQLite 中用单行 JSON 存储动态列表 —— 它违背关系型设计原则，必然引发并发问题。
- ✅ 务必将集合数据建模为独立行（tasks 表），利用数据库原生的 INSERT/DELETE 原子性。
- ✅ 所有涉及“读取-修改-写入”的逻辑，均应重构为单语句、无中间状态的操作。
- ✅ 需要排序时，优先使用 created_at 或专用 order_index 列，而非依赖 JSON 数组下标。
此方案不仅彻底解决你遇到的“过期值”问题，更使代码更简洁、查询更高效、扩展更灵活——这才是 SQLite 在 Web 应用中的正确打开方式。
