Python SQLite WAL 并发模式与线程安全连接管理实践

一、问题背景：默认回滚日志下的并发写锁瓶颈

在云端轻量级微服务、边缘计算节点或本地任务调度器中，SQLite 常被用作嵌入式状态存储。但在默认的回滚日志模式（journal_mode=DELETE）下，写入事务会获取排他锁（Exclusive Lock），导致读写操作互相阻塞；当多个工作线程同时写入或边写边读时，极易抛出 sqlite3.OperationalError: database is locked 异常。

要解决多线程读写冲突并保证连接生命周期安全，核心需要两项改造：

1. 开启 WAL（Write-Ahead Logging）预写日志模式，将追加写与读取分离，实现读不阻塞写、写不阻塞读。
2. 使用 threading.local() 线程本地存储隔离每个工作线程的数据库连接，配合合理的 busy_timeout 锁等待超时。

二、核心配置参数对比

• PRAGMA journal_mode=WAL;：启用预写日志，写操作追加到 -wal 文件，读操作直接读取一致快照。
• PRAGMA synchronous=NORMAL;：在 WAL 模式下兼顾落盘安全性与吞吐性能，避免每次事务都触发完整的 fsync 阻塞。
• PRAGMA busy_timeout=5000;：当多个写线程争用写锁时，底层引擎自动重试等待最多 5000 毫秒，而非立即报错返回。

三、完整可运行代码实现

以下脚本在 Python 3.11 标准库环境下直接运行，无需任何第三方依赖，包含完整的 ThreadSafeSQLitePool 封装与 4 写线程 + 4 读线程并发压测验证：

import os
import sqlite3
import tempfile
import threading
import time


class ThreadSafeSQLitePool:
    """基于 threading.local 与 WAL 模式的轻量级 SQLite 线程安全连接管理器"""

    def __init__(self, db_path: str, busy_timeout_ms: int = 5000):
        self.db_path = db_path
        self.busy_timeout_ms = busy_timeout_ms
        self._local = threading.local()
        self._init_wal_schema()

    def _create_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path, timeout=self.busy_timeout_ms / 1000.0)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL;")
        conn.execute("PRAGMA synchronous=NORMAL;")
        conn.execute(f"PRAGMA busy_timeout={self.busy_timeout_ms};")
        conn.execute("PRAGMA foreign_keys=ON;")
        return conn

    def get_conn(self) -> sqlite3.Connection:
        if getattr(self._local, "conn", None) is None:
            self._local.conn = self._create_connection()
        return self._local.conn

    def close_thread_conn(self):
        conn = getattr(self._local, "conn", None)
        if conn is not None:
            conn.close()
            self._local.conn = None

    def _init_wal_schema(self):
        conn = self._create_connection()
        try:
            mode = conn.execute("PRAGMA journal_mode;").fetchone()[0]
            print(f"[Init] SQLite journal_mode = {mode}", flush=True)
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS task_metrics (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    worker_id TEXT NOT NULL,
                    latency_ms INTEGER NOT NULL,
                    created_at REAL NOT NULL
                );
                """
            )
            conn.commit()
        finally:
            conn.close()


def verify_concurrent_wal():
    with tempfile.TemporaryDirectory() as tmpdir:
        db_file = os.path.join(tmpdir, "metrics.db")
        pool = ThreadSafeSQLitePool(db_file, busy_timeout_ms=5000)
        errors = []
        read_counts = []

        def writer_worker(wid: int):
            try:
                conn = pool.get_conn()
                for i in range(25):
                    with conn:
                        conn.execute(
                            "INSERT INTO task_metrics (worker_id, latency_ms, created_at) VALUES (?, ?, ?)",
                            (f"writer-{wid}", 10 + i, time.time()),
                        )
                    time.sleep(0.005)
            except Exception as e:
                errors.append(f"writer-{wid}: {e}")
            finally:
                pool.close_thread_conn()

        def reader_worker(rid: int):
            try:
                conn = pool.get_conn()
                local_reads = 0
                for _ in range(30):
                    row = conn.execute("SELECT COUNT(*) AS cnt FROM task_metrics").fetchone()
                    if row["cnt"] >= 0:
                        local_reads += 1
                    time.sleep(0.004)
                read_counts.append(local_reads)
            except Exception as e:
                errors.append(f"reader-{rid}: {e}")
            finally:
                pool.close_thread_conn()

        threads = (
            [threading.Thread(target=writer_worker, args=(i,)) for i in range(4)]
            + [threading.Thread(target=reader_worker, args=(j,)) for j in range(4)]
        )
        t0 = time.monotonic()
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        elapsed_ms = round((time.monotonic() - t0) * 1000)

        conn = pool.get_conn()
        total_rows = conn.execute("SELECT COUNT(*) FROM task_metrics").fetchone()[0]
        pool.close_thread_conn()
        print(
            f"[Result] 4写线程+4读线程并发完成: 总写入行数={total_rows}, "
            f"各读线程成功读取次数={read_counts}, 错误数={len(errors)}, 总耗时={elapsed_ms}ms",
            flush=True,
        )


if __name__ == "__main__":
    verify_concurrent_wal()

四、实际运行验证与生产注意事项

在 Python 3.11 环境下执行上述脚本，实测输出如下：

• [Init] SQLite journal_mode = wal
• [Result] 4写线程+4读线程并发完成: 总写入行数=100, 各读线程成功读取次数=[30, 30, 30, 30], 错误数=0, 总耗时=179ms

生产环境落地要点：

1. 避免跨网络文件系统挂载 WAL：WAL 依赖操作系统底层的共享内存映射（-shm 文件），数据库文件必须位于容器本地卷或宿主机本地磁盘，切勿放在 NFS 等无锁一致性保障的网络挂载盘上。
2. 控制长事务与定期 Checkpoint：使用 with conn: 上下文管理器让写事务尽快提交，避免未提交事务阻止 WAL 文件回收。