Python多进程场景下共享静态类SQLite连接的技术问询

首先得给你划个重点：SQLite的单个数据库连接绝对不能跨进程共享！因为每个进程都有自己独立的内存空间，你在父进程初始化的Utils.dbClient，子进程启动时只会复制这个对象的副本，根本不是同一个连接，强行这么搞只会导致数据库锁冲突、数据写丢或者各种奇怪的异常。

下面给你说正确的实现思路和代码调整方案：

1. 让每个子进程独立初始化连接 #

别在父进程提前初始化Utils.dbClient了，改成让每个子进程启动后自己初始化专属的连接。调整Utils类如下：

class Utils:
    dbClient = None

    @staticmethod
    def initialize(settings):
        from storage.sqlite.SQLiteClient import SqliteClient
        # 只有当前进程还没初始化连接时才创建
        if Utils.dbClient is None:
            Utils.dbClient = SqliteClient(settings)
    
    @staticmethod
    def get_db_client():
        # 确保调用前已经完成初始化
        if Utils.dbClient is None:
            raise RuntimeError("先调用Utils.initialize(settings)初始化数据库连接啊！")
        return Utils.dbClient

然后在app.py里：

• 父进程不用提前调用Utils.initialize(settings)
• 每个子进程启动后的第一件事，就是调用Utils.initialize(settings)来创建自己的数据库连接

2. 配置SQLite的多进程兼容参数 #

为了减少多进程操作时的锁冲突，必须给SQLite加上这两个关键配置：

• 开启WAL模式（Write-Ahead Logging）：这是SQLite官方推荐的多进程读写兼容模式，能大幅降低锁的概率
• 设置合理的超时时间：让进程遇到锁时自动等待一段时间再重试，而不是直接报错

修改你的SqliteClient初始化代码：

class SqliteClient:
    def __init__(self, settings):
        import sqlite3
        self.conn = sqlite3.connect(
            settings.db_path,
            timeout=30,  # 遇到锁时等待30秒再重试，可根据业务调整
            check_same_thread=False  # 只有当前进程内有多线程操作时需要，多进程下不影响
        )
        # 启用WAL模式，必须执行并提交
        self.conn.execute("PRAGMA journal_mode=WAL;")
        self.conn.commit()

3. 可选优化：用完连接及时关闭 #

如果你的子进程是处理完任务就退出的短生命周期进程，可以加个关闭方法来释放资源：

@staticmethod
def close_db_client():
    if Utils.dbClient is not None:
        Utils.dbClient.conn.close()
        Utils.dbClient = None

再啰嗦一句为啥不能共享连接 #

本质原因是SQLite的连接对象是绑定到当前进程内存的，子进程没法访问父进程的内存空间。父进程初始化的连接到了子进程里就是个“假对象”，底层的数据库连接早就失效了，操作起来要么抛异常，要么搞坏数据库文件。

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