SQLite 出现“Database is locked” 如何解决？

Question

项目中有两个进程操作sqlite, 会遇到同时写的情况，频繁出现database is locked的情况，怎么解决？

下面是我操作sqlite数据库的工具类

class SqliteTool():
    """
       sqlite数据库工具类
       """
    def __init__(self, dbName="data.db"):
        """
        初始化连接——使用完需关闭连接
        :param dbName: 连接库的名字，注意，以'.db'结尾
        """
        # 连接数据库
        self._conn = sqlite3.connect(dbName)
        self._conn.row_factory = dict_factory
        # 创建游标
        self._cur = self._conn.cursor()
    # 查询多条数据
    def query_many(self, sql: str, params=None):
        """
        查询多条数据
        :param sql: select语句
        :param params: 查询参数，形如()
        :return: 语句查询多条结果
        """
        try:
            if params:
                self._cur.execute(sql, params)
            else:
                self._cur.execute(sql)
            # 调用fetchall()方法
            r = self._cur.fetchall()
            print("[select many records success]")
            return r
        except Exception as e:
            print("[select many records error]", e)

    # 插入或更新表数据，一次插入或更新一条数据
    def operate_one(self, sql: str, value: tuple):
        """
        插入或更新单条表记录
        :param sql: insert语句或update语句
        :param value: 插入或更新的值，形如（）
        :return: True表示插入或更新成功
        """
        try:
            self._cur.execute(sql, value)
            self._conn.commit()
            if 'INSERT' in sql.upper():
                print("[insert one record success]")
            if 'UPDATE' in sql.upper():
                print("[update one record success]")
            return True
        except Exception as e:
            self._conn.rollback()
            return False

Answer 1

参考免费微信小程序：皆我百晓生 作答，经由专家-皆知整理回复，如问题解决，望采纳；如仍有疑问，请评论回复。

在SQLite数据库操作中，出现“Database is locked”错误通常是因为多个进程或线程同时尝试写入数据库，导致数据库被锁定。为了解决这个问题，你可以采取以下几种策略：

1. 使用写入锁：确保在写入数据库时，其他进程或线程不会同时进行写入操作。
2. 增加超时时间：调整数据库锁定超时时间，以便在等待锁释放时给予更长的时间。
3. 使用连接池：通过连接池管理数据库连接，减少因频繁开启和关闭连接导致的锁冲突。

针对你的代码，我建议进行以下修改：

1. 增加锁超时时间：在连接数据库时，增加超时时间参数。
2. 使用连接池：如果可能，使用连接池管理数据库连接。

下面是修改后的代码：

import sqlite3
from contextlib import contextmanager

class SqliteTool():
    """
    sqlite数据库工具类
    """
    def __init__(self, dbName="data.db", timeout=60):
        """
        初始化连接——使用完需关闭连接
        :param dbName: 连接库的名字，注意，以'.db'结尾
        :param timeout: 锁定超时时间（秒）
        """
        # 连接数据库
        self._conn = sqlite3.connect(dbName, timeout=timeout)
        self._conn.row_factory = dict_factory
        # 创建游标
        self._cur = self._conn.cursor()

    @contextmanager
    def _locked_cursor(self):
        try:
            yield self._cur
            self._conn.commit()
        except Exception as e:
            self._conn.rollback()
            raise e

    def query_many(self, sql: str, params=None):
        """
        查询多条数据
        :param sql: select语句
        :param params: 查询参数，形如()
        :return: 语句查询多条结果
        """
        with self._locked_cursor() as cur:
            try:
                if params:
                    cur.execute(sql, params)
                else:
                    cur.execute(sql)
                # 调用fetchall()方法
                r = cur.fetchall()
                print("[select many records success]")
                return r
            except Exception as e:
                print("[select many records error]", e)

    def operate_one(self, sql: str, value=()):
        """
        插入或更新单条表记录
        :param sql: insert语句或update语句
        :param value: 插入或更新的值，形如(())
        :return: True表示插入或更新成功
        """
        with self._locked_cursor() as cur:
            try:
                cur.execute(sql, value)
                if 'INSERT' in sql.upper():
                    print("[insert one record success]")
                if 'UPDATE' in sql.upper():
                    print("[update one record success]")
                return True
            except Exception as e:
                print("[operate one error]", e)
                return False

# Helper function to create a dictionary factory
def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d

关键点解释：

1. 增加超时时间：

   • 在sqlite3.connect函数中增加timeout参数，设置数据库锁定超时时间。

2. 使用连接池：

   • 虽然SQLite本身不支持连接池，但可以通过管理连接的生命周期来减少锁冲突。这里使用contextmanager来确保每次操作后都正确提交或回滚事务，并释放游标。

3. 异常处理：

   • 在_locked_cursor上下文中处理异常，确保事务正确提交或回滚。

通过这些修改，你可以减少数据库锁定的情况，提高数据库操作的稳定性和性能。