PyQt5+SQLite多用户并发写入网络数据库问题求助

核心问题分析 #

SQLite的WAL模式依赖本地文件系统的高效锁机制，而网络共享存储（如SMB/NFS）的锁实现存在延迟或兼容性缺陷，导致多连接写操作时，只有最后一个成功获取锁的连接能完成WAL数据合并，其他连接的修改仅停留在本地连接缓存，无法同步到主数据库。

具体修复措施 #

1. 写操作后强制触发WAL检查点 #

WAL模式下数据默认异步合并到主库，网络环境下异步合并极易因锁冲突失败。每次写操作提交事务后，手动执行全量检查点，确保WAL数据立即写入主库：

def write_data(db_path, data):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        # 开启排他事务，避免写冲突
        conn.execute("BEGIN EXCLUSIVE;")
        # 执行写入操作（示例：插入数据）
        cursor.execute("INSERT INTO your_table (col1, col2) VALUES (?, ?)", data)
        conn.commit()
        # 强制全量检查点，将WAL数据合并到主库
        conn.execute("PRAGMA wal_checkpoint(FULL);")
    except sqlite3.Error as e:
        conn.rollback()
        raise e
    finally:
        conn.close()

2. 使用短连接替代长连接 #

不要在UI中保持长期数据库连接，每次写操作时临时打开连接，完成后立即关闭，减少锁持有时间：

• 移除全局/类级别的数据库连接对象
• 每次执行写入、查询操作时，单独创建连接，操作结束后关闭

3. 调整SQLite同步参数 #

将synchronous设置为FULL，强制WAL文件写入后同步到磁盘，避免网络缓存导致的数据不同步：

# 连接数据库时配置参数
conn = sqlite3.connect(db_path, timeout=10)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA synchronous=FULL;")

4. 增加锁冲突重试机制 #

捕获SQLITE_BUSY和SQLITE_LOCKED异常，自动重试操作，避免单次锁失败导致数据丢失：

import time

def write_with_retry(db_path, data, max_retries=3, delay=1):
    for attempt in range(max_retries):
        try:
            return write_data(db_path, data)
        except sqlite3.OperationalError as e:
            if "database is locked" in str(e) or "busy" in str(e):
                time.sleep(delay)
                continue
            raise
    raise Exception("Max retries reached, failed to write data")

5. 数据库操作移至子线程 #

PyQt中UI线程的阻塞会延长锁持有时间，将所有数据库操作放到QThread子线程中执行，确保锁能快速释放：

from PyQt5.QtCore import QThread, pyqtSignal

class DBWorker(QThread):
    finished = pyqtSignal(bool)
    error = pyqtSignal(str)

    def __init__(self, db_path, data):
        super().__init__()
        self.db_path = db_path
        self.data = data

    def run(self):
        try:
            write_with_retry(self.db_path, self.data)
            self.finished.emit(True)
        except Exception as e:
            self.error.emit(str(e))

# 在UI代码中调用
worker = DBWorker(db_path, your_data)
worker.finished.connect(self.on_write_success)
worker.error.connect(self.on_write_error)
worker.start()

额外注意事项 #

• 确保所有客户端使用相同版本的SQLite，避免因WAL格式差异导致的兼容性问题
• 定期清理网络存储上的孤立-wal和-shm文件（程序异常退出时可能残留）
• 拆分大规模批量写入为小批量操作，减少锁冲突概率

内容的提问来源于stack exchange，提问作者R41nMak3R