一、SQLite锁竞争问题的由来
SQLite作为一个轻量级的嵌入式数据库，在很多场景下都非常好用。但是当多个连接同时操作同一个数据库时，就可能会出现锁竞争的问题。这就像是在超市结账时，大家都想用同一个收银台，结果就堵在那里了。
SQLite默认使用的是WAL(Write-Ahead Logging)模式，这种模式下读写是可以并发的，但是写写之间还是需要排队。当多个线程或进程同时尝试修改数据库时，后面的操作就必须等待前面的操作完成。
举个例子，我们用Python操作SQLite时：
import sqlite3
from threading import Thread
def update_data(thread_id):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 这里模拟一个耗时操作
cursor.execute("UPDATE users SET balance = balance + 1 WHERE id = 1")
print(f"线程{thread_id}更新完成")
conn.commit()
conn.close()
# 创建并启动5个线程
threads = []
for i in range(5):
t = Thread(target=update_data, args=(i,))
threads.append(t)
t.start()
# 等待所有线程完成
for t in threads:
t.join()
运行这段代码时，你会发现线程是一个接一个完成的，而不是同时完成，这就是锁竞争的表现。
二、SQLite的锁机制详解
SQLite的锁机制其实挺有意思的，它采用了渐进式的锁策略，分为以下几种状态：
- UNLOCKED：无锁状态
- SHARED：共享读锁
- RESERVED：预留写锁
- PENDING：等待独占锁
- EXCLUSIVE：独占锁
当有多个写操作时，SQLite会使用PENDING锁来确保只有一个连接能进入EXCLUSIVE状态。这就像是在餐厅订位，你可以先预定(RESERVED)，等确定没人反对后(PENDING)，最后才能独占使用(EXCLUSIVE)。
让我们看一个更复杂的例子：
import sqlite3
import time
from threading import Thread
def long_running_transaction(thread_id):
conn = sqlite3.connect('test.db', isolation_level=None)
conn.execute("PRAGMA journal_mode=WAL") # 使用WAL模式
cursor = conn.cursor()
print(f"线程{thread_id}开始事务")
cursor.execute("BEGIN IMMEDIATE") # 立即获取RESERVED锁
# 模拟长时间操作
cursor.execute("SELECT * FROM users WHERE id = 1")
print(f"线程{thread_id}获取数据")
time.sleep(2) # 故意等待，制造锁竞争
cursor.execute("UPDATE users SET balance = balance + 1 WHERE id = 1")
conn.commit()
print(f"线程{thread_id}提交事务")
conn.close()
# 创建并启动3个线程
for i in range(3):
Thread(target=long_running_transaction, args=(i,)).start()
在这个例子中，我们使用了BEGIN IMMEDIATE来立即获取RESERVED锁，这样其他连接就只能读不能写了。你会看到线程是一个接一个执行的，因为每个线程都要等待前一个线程释放锁。
三、常见的锁竞争场景及解决方案
3.1 长时间运行的事务
这是最常见的锁竞争原因。当一个事务执行时间过长时，其他写操作就会被阻塞。
解决方案是：
- 尽量缩短事务时间
- 将大事务拆分为小事务
- 使用定时提交策略
def batch_update(thread_id, chunk_size=100):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 获取总记录数
cursor.execute("SELECT COUNT(*) FROM large_table")
total = cursor.fetchone()[0]
# 分批处理
for offset in range(0, total, chunk_size):
try:
conn.execute("BEGIN")
cursor.execute("UPDATE large_table SET status = ? WHERE id IN "
"(SELECT id FROM large_table LIMIT ? OFFSET ?)",
(thread_id, chunk_size, offset))
conn.commit()
print(f"线程{thread_id}处理了{chunk_size}条记录")
except:
conn.rollback()
raise
conn.close()
3.2 高并发写入场景
当有大量并发写入时，即使每个事务都很短，也可能出现竞争。
解决方案：
- 使用写队列，将写入操作序列化
- 采用批量插入代替单条插入
- 考虑使用内存数据库作为缓冲
from queue import Queue
import threading
write_queue = Queue()
stop_event = threading.Event()
def writer_thread():
conn = sqlite3.connect('test.db')
while not stop_event.is_set() or not write_queue.empty():
try:
sql, params = write_queue.get(timeout=1)
conn.execute(sql, params)
conn.commit()
except queue.Empty:
continue
conn.close()
# 启动写线程
threading.Thread(target=writer_thread, daemon=True).start()
# 其他线程通过队列提交写操作
def async_update(user_id, amount):
write_queue.put((
"UPDATE users SET balance = balance + ? WHERE id = ?",
(amount, user_id)
))
3.3 读多写少场景
在读多写少的应用中，读操作也可能被写操作阻塞。
解决方案：
- 使用WAL模式
- 设置合适的busy_timeout
- 考虑读写分离
# 配置WAL模式和busy_timeout
def setup_database():
conn = sqlite3.connect('test.db')
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA busy_timeout=3000") # 3秒超时
conn.close()
# 读操作使用单独的连接
def read_operation():
conn = sqlite3.connect('test.db', isolation_level=None)
conn.execute("PRAGMA read_uncommitted=1") # 脏读，避免锁等待
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
conn.close()
return results
四、高级优化技巧与实践建议
4.1 合理设置PRAGMA参数
SQLite提供了很多PRAGMA指令来调整行为：
def optimize_database():
conn = sqlite3.connect('test.db')
# WAL模式支持读写并发
conn.execute("PRAGMA journal_mode=WAL")
# 正常同步级别，在性能和数据安全间取得平衡
conn.execute("PRAGMA synchronous=NORMAL")
# 设置适当的缓存大小
conn.execute("PRAGMA cache_size=-4000") # 4000页缓存
# 设置锁等待超时为2秒
conn.execute("PRAGMA busy_timeout=2000")
conn.close()
4.2 连接池管理
频繁创建和关闭连接会增加锁竞争的概率，使用连接池可以缓解这个问题：
from queue import Queue
import sqlite3
class SQLiteConnectionPool:
def __init__(self, db_path, pool_size=5):
self._queue = Queue(maxsize=pool_size)
for _ in range(pool_size):
conn = sqlite3.connect(db_path)
# 统一配置所有连接
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=2000")
self._queue.put(conn)
def get_conn(self):
return self._queue.get()
def return_conn(self, conn):
self._queue.put(conn)
def close_all(self):
while not self._queue.empty():
conn = self._queue.get()
conn.close()
# 使用示例
pool = SQLiteConnectionPool('test.db')
def using_pool_example():
conn = pool.get_conn()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
return cursor.fetchall()
finally:
pool.return_conn(conn)
4.3 监控和诊断锁竞争
当出现锁竞争问题时，我们需要诊断工具：
def diagnose_locks():
conn = sqlite3.connect('test.db')
# 查看当前锁状态
cursor = conn.cursor()
cursor.execute("PRAGMA database_list")
databases = cursor.fetchall()
for db in databases:
print(f"\n数据库: {db[1]}")
cursor.execute(f"PRAGMA {db[1]}.locking_mode")
print(f"锁模式: {cursor.fetchone()[0]}")
cursor.execute(f"PRAGMA {db[1]}.journal_mode")
print(f"日志模式: {cursor.fetchone()[0]}")
# 查看当前活动事务
cursor.execute("SELECT * FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
print("\n活动事务:")
for table in tables:
try:
cursor.execute(f"SELECT * FROM {table[1]} LIMIT 0")
except sqlite3.OperationalError as e:
if "database is locked" in str(e):
print(f"表{table[1]}被锁定")
conn.close()
五、总结与最佳实践
经过上面的分析，我们可以总结出以下最佳实践：
- 对于写密集应用，尽量使用批量操作减少事务次数
- 合理设置WAL模式和busy_timeout参数
- 长时间运行的事务要拆分为小事务
- 使用连接池管理数据库连接
- 考虑使用队列序列化写操作
- 监控数据库锁状态，及时发现潜在问题
- 根据应用特点选择合适的同步级别
SQLite虽然轻量，但在并发处理上确实有一些限制。理解它的锁机制后，我们可以通过合理的架构设计和参数调优，让它更好地服务于我们的应用。记住，没有银弹，关键是要根据你的具体场景选择最合适的解决方案。
评论