SQLite 在独立开发中的优化实战：用轻量架构应对高并发

graph TD
    A[应用并发请求] --> B[应用进程内 SQLite 引擎]
    B -->|开启 WAL 模式| C{读写分离控制}
    C -->|读操作| D[直接读取原始数据 db 文件]
    C -->|读操作| E[读取未同步的 wal 文件]
    D --> F[快速响应结果，读操作不阻塞]
    E --> F
    C -->|写操作| G[将增量写入 wal 缓冲区文件]
    G --> H[写操作不阻塞后续读操作]
    H -->|定期触发 Checkpoint| I[自动将 wal 中的增量同步回主 db 文件]

import sqlite3
import threading
import time
import os
from typing import List
class SQLiteProductionConfig:
    def __init__(self, db_path: str):
        self.db_path = db_path
    def get_connection(self) -> sqlite3.Connection:
        """获取连接，配置高并发参数"""
        # timeout 设为 10 秒，遇到锁自动等待，而不是直接抛异常
        conn = sqlite3.connect(self.db_path, timeout=10.0)
        # 开启 WAL 模式，实现读写并发
        conn.execute("PRAGMA journal_mode=WAL;")
        # 优化内存页缓存（4000 页，约 16MB）
        conn.execute("PRAGMA cache_size=-4000;")
        # 同步模式设为 NORMAL，在安全前提下提升写入速度
        conn.execute("PRAGMA synchronous=NORMAL;")
        return conn
    def initialize_db(self):
        """初始化数据表"""
        conn = self.get_connection()
        conn.execute(
            "CREATE TABLE IF NOT EXISTS sys_logs ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT,"
            "thread_name TEXT,"
            "timestamp REAL"
            ");"
        )
        conn.commit()
        conn.close()
def concurrent_writer_task(config: SQLiteProductionConfig, thread_idx: int):
    """模拟多线程高频并发写入"""
    conn = config.get_connection()
    thread_name = f"Thread-{thread_idx}"
    try:
        for _ in range(50): # 每个线程写入 50 条
            conn.execute(
                "INSERT INTO sys_logs (thread_name, timestamp) VALUES (?, ?);",
                (thread_name, time.time())
            )
            conn.commit()
            time.sleep(0.01) # 模拟微小的 IO 间歇
    except sqlite3.OperationalError as e:
        print(f"[{thread_name}] 写入异常: {e}")
    finally:
        conn.close()
if __name__ == "__main__":
    db_file = "production_wal.db"
    # 清理旧文件
    if os.path.exists(db_file):
        os.remove(db_file)
    config = SQLiteProductionConfig(db_file)
    print("初始化 SQLite 并开启 WAL 优化...")
    config.initialize_db()
    # 启动 10 个线程并发写入
    threads: List[threading.Thread] = []
    print("启动 10 个线程模拟高并发写入测试...")
    start_time = time.time()
    for i in range(10):
        t = threading.Thread(target=concurrent_writer_task, args=(config, i))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    duration = time.time() - start_time
    # 统计结果
    conn = config.get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM sys_logs;")
    total_count = cursor.fetchone()[0]
    conn.close()
    print(f"\n【并发写入测试报告】")
    print(f"  - 总写入行数: {total_count} 行 (预期 500 行)")
    print(f"  - 总耗时: {duration:.4f} 秒")
    print("测试成功！WAL 模式下，多线程并发写入未触发 'database is locked' 错误。")
    # 清理测试文件
    for suffix in ["", "-shm", "-wal"]:
        f_path = db_file + suffix
        if os.path.exists(f_path):
            os.remove(f_path)