Python sqlite3：用 BEGIN IMMEDIATE 提前拿写锁

50 次浏览2 条回复

有些小工具会用 SQLite 当本地状态库。默认事务经常是写到一半才发现锁冲突，错误点有点晚；如果这段逻辑一定要独占写入，可以一开始就执行 BEGIN IMMEDIATE。

环境：Python 3.10+，Linux/macOS/Windows，只用标准库。保存为 sqlite_lock_demo.py：

import sqlite3
import tempfile
from pathlib import Path

path = Path(tempfile.gettempdir()) / "sqlite-lock-demo.db"
try:
    path.unlink()
except FileNotFoundError:
    pass

init = sqlite3.connect(path)
init.execute("create table kv (key text primary key, value text)")
init.execute("insert into kv values ('mode', 'old')")
init.commit()
init.close()

# timeout=0 方便演示：拿不到锁就立刻报错
a = sqlite3.connect(path, timeout=0, isolation_level=None)
b = sqlite3.connect(path, timeout=0, isolation_level=None)

a.execute("BEGIN IMMEDIATE")
print("a got write lock")

try:
    b.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as exc:
    print(f"b failed early: {exc}")

a.execute("update kv set value = 'new' where key = 'mode'")
a.commit()
print(dict(a.execute("select key, value from kv")))

a.close()
b.close()

运行：

python3 sqlite_lock_demo.py

输出里第二个连接会在 BEGIN IMMEDIATE 这一行就失败。这样适合“抢不到写锁就本轮跳过”的脚本，比业务逻辑跑到一半再炸要好处理一点。

RanoLv1

32 天前

这个点挺实用。再补一个小坑：如果不是想立刻失败，可以把 timeout 或 PRAGMA busy_timeout 设成一个很短的值，比如几百毫秒，让短暂写入有机会错开。

conn = sqlite3.connect(path, timeout=0.5, isolation_level=None)
# 或者：conn.execute("PRAGMA busy_timeout = 500")
conn.execute("BEGIN IMMEDIATE")

我一般会把它配合重试次数一起用：拿不到锁就跳过本轮，别在锁上无限等。

半

半夜七号Lv1

32 天前

还可以顺手确认一下 journal_mode。默认 rollback journal 下，写事务和读事务更容易互相卡住；如果是本地小服务/脚本长期读多写少，开启 WAL 往往舒服一点：

conn = sqlite3.connect(path, isolation_level=None)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout = 500")
conn.execute("BEGIN IMMEDIATE")

不过 WAL 会多出 -wal/-shm 文件，放在网络盘或很奇怪的同步目录里我会谨慎些。普通本地磁盘上配 BEGIN IMMEDIATE 用，读写冲突会少很多。

返回主题列表