SQLite 完全指南 / 16 - 性能调优
16 - 性能调优：WAL 优化、批量操作与内存映射
16.1 性能优化总览
SQLite 性能优化的四个层面：
┌────────────────────────────────┐
│ 1. PRAGMA 配置优化 │ ← 最简单、见效最快
├────────────────────────────────┤
│ 2. Schema 与索引设计 │ ← 结构性优化
├────────────────────────────────┤
│ 3. SQL 查询优化 │ ← 逐条优化
├────────────────────────────────┤
│ 4. 应用层优化 │ ← 批量、缓存、并发
└────────────────────────────────┘
16.2 PRAGMA 性能配置
16.2.1 推荐配置
-- ⚡ 读密集应用（Web、API）
PRAGMA journal_mode = WAL; -- WAL 模式
PRAGMA synchronous = NORMAL; -- WAL 下足够安全
PRAGMA cache_size = -20000; -- 20MB 缓存
PRAGMA mmap_size = 268435456; -- 256MB 内存映射
PRAGMA temp_store = MEMORY; -- 临时表存内存
PRAGMA busy_timeout = 5000; -- 5 秒忙等待
PRAGMA foreign_keys = ON;
PRAGMA wal_autocheckpoint = 1000;
-- ⚡ 写密集应用（日志采集）
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = 10000;
PRAGMA temp_store = MEMORY;
PRAGMA busy_timeout = 10000;
-- ⚡ 批量数据导入
PRAGMA journal_mode = WAL;
PRAGMA synchronous = OFF; -- 临时降低安全级别
PRAGMA cache_size = 100000; -- 大缓存
PRAGMA locking_mode = EXCLUSIVE; -- 独占锁
PRAGMA temp_store = MEMORY;
PRAGMA mmap_size = 536870912; -- 512MB
-- 导入完成后恢复
PRAGMA synchronous = NORMAL;
PRAGMA locking_mode = NORMAL;
16.2.2 配置性能对比
| 配置 | INSERT 1000 行 | SELECT 10000 行 |
|---|---|---|
| 默认配置 | ~500ms | ~15ms |
| WAL + NORMAL | ~200ms | ~10ms |
| WAL + NORMAL + 大缓存 | ~150ms | ~8ms |
| WAL + OFF + 独占锁 | ~50ms | ~8ms |
| 事务批量插入 | ~5ms | ~8ms |
16.3 批量操作优化
16.3.1 批量插入
import sqlite3
import time
conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE test (id INTEGER PRIMARY KEY, val TEXT)')
# ❌ 慢：逐条插入（每条一个事务）
start = time.time()
for i in range(10000):
conn.execute('INSERT INTO test VALUES (?, ?)', (i, f'value_{i}'))
conn.commit()
print(f"逐条插入: {time.time() - start:.3f}s")
# ✅ 快：批量插入（一个事务）
conn.execute('DELETE FROM test')
start = time.time()
conn.execute('BEGIN')
for i in range(10000):
conn.execute('INSERT INTO test VALUES (?, ?)', (i, f'value_{i}'))
conn.execute('COMMIT')
print(f"事务批量: {time.time() - start:.3f}s")
# ✅✅ 更快：executemany
conn.execute('DELETE FROM test')
start = time.time()
conn.executemany('INSERT INTO test VALUES (?, ?)',
[(i, f'value_{i}') for i in range(10000)])
conn.commit()
print(f"executemany: {time.time() - start:.3f}s")
# ✅✅✅ 最快：VALUES 批量插入
conn.execute('DELETE FROM test')
start = time.time()
data = [(i, f'value_{i}') for i in range(10000)]
for chunk_start in range(0, len(data), 500):
chunk = data[chunk_start:chunk_start + 500]
placeholders = ','.join(['(?,?)'] * len(chunk))
flat = [v for row in chunk for v in row]
conn.execute(f'INSERT INTO test VALUES {placeholders}', flat)
conn.commit()
print(f"批量 VALUES: {time.time() - start:.3f}s")
16.3.2 批量更新与删除
# ✅ 使用 CASE WHEN 批量更新
conn.execute("""
UPDATE products SET price = CASE id
WHEN 1 THEN 100
WHEN 2 THEN 200
WHEN 3 THEN 300
END
WHERE id IN (1, 2, 3)
""")
# ✅ 使用临时表批量更新
conn.execute("""
CREATE TEMP TABLE updates (id INTEGER, new_price REAL);
INSERT INTO updates VALUES (1, 100), (2, 200), (3, 300);
UPDATE products SET price = (SELECT new_price FROM updates WHERE updates.id = products.id)
WHERE id IN (SELECT id FROM updates);
DROP TABLE updates;
""")
16.3.3 INSERT OR REPLACE 批量
-- 使用 INSERT ... ON CONFLICT 批量 upsert
INSERT INTO users (id, name, email) VALUES
(1, '张三', 'zs@example.com'),
(2, '李四', 'ls@example.com'),
(3, '王五', 'ww@example.com')
ON CONFLICT(id) DO UPDATE SET
name = excluded.name,
email = excluded.email;
16.4 内存映射（mmap）
16.4.1 配置 mmap
-- 设置 mmap 大小（字节）
PRAGMA mmap_size = 268435456; -- 256MB
-- 关闭 mmap
PRAGMA mmap_size = 0;
-- 查看当前设置
PRAGMA mmap_size;
16.4.2 mmap 适用场景
| 场景 | 是否推荐 | 说明 |
|---|---|---|
| 只读数据库 | ✅ 推荐 | mmap 可以显著加速 |
| 读多写少 | ✅ 推荐 | 大部分读取命中缓存 |
| 写入密集 | ⚠️ 谨慎 | 可能影响写入性能 |
| 32 位系统 | ❌ 不推荐 | 虚拟地址空间有限 |
| 数据库 > 可用内存 | ⚠️ 谨慎 | 可能导致频繁换页 |
16.5 查询优化技巧
16.5.1 避免 SELECT *
-- ❌ 返回所有列
SELECT * FROM orders WHERE user_id = 42;
-- ✅ 只返回需要的列（可能使用覆盖索引）
SELECT id, status, amount FROM orders WHERE user_id = 42;
16.5.2 分页优化
-- ❌ 大 OFFSET（慢）
SELECT * FROM products ORDER BY id LIMIT 20 OFFSET 100000;
-- ✅ 游标分页（快）
SELECT * FROM products WHERE id > 100000 ORDER BY id LIMIT 20;
-- ✅ 延迟关联
SELECT p.* FROM products p
INNER JOIN (SELECT id FROM products ORDER BY name LIMIT 20 OFFSET 100000) t
ON p.id = t.id;
16.5.3 COUNT 优化
-- ❌ 精确计数（全表扫描）
SELECT COUNT(*) FROM big_table WHERE status = 'active';
-- ✅ 使用索引
CREATE INDEX idx_status ON big_table(status);
-- ✅ 近似计数（统计信息）
SELECT seq FROM sqlite_sequence WHERE name = 'big_table';
-- ✅ 维护计数表
CREATE TABLE counters (name TEXT PRIMARY KEY, cnt INTEGER);
16.5.4 存在性检查
-- ❌ COUNT 判断
SELECT COUNT(*) FROM orders WHERE user_id = 42;
-- 然后在应用层判断 count > 0
-- ✅ EXISTS（找到一个就停止）
SELECT EXISTS(SELECT 1 FROM orders WHERE user_id = 42);
-- ✅ LIMIT 1
SELECT 1 FROM orders WHERE user_id = 42 LIMIT 1;
16.6 数据库维护
16.6.1 VACUUM 与碎片整理
-- 查看碎片程度
PRAGMA freelist_count; -- 空闲页面
PRAGMA page_count; -- 总页面
-- 碎片率 = freelist_count / page_count
-- 如果 > 20%，建议 VACUUM
-- 执行 VACUUM
VACUUM;
-- VACUUM 到新文件（不影响当前连接）
VACUUM INTO 'compact.db';
16.6.2 ANALYZE
-- 收集统计信息（优化器依赖这些信息选择索引）
ANALYZE;
-- 只分析特定表
ANALYZE orders;
-- 限制采样量（加速 ANALYZE）
PRAGMA analysis_limit = 1000;
-- 查看统计信息
SELECT * FROM sqlite_stat1;
16.6.3 定期维护脚本
#!/bin/bash
# sqlite_maintenance.sh
DB="/var/data/mydb.db"
# Checkpoint + Analyze + Vacuum
sqlite3 "$DB" <<EOF
PRAGMA wal_checkpoint(TRUNCATE);
ANALYZE;
VACUUM;
PRAGMA optimize;
EOF
echo "维护完成: $(date)"
16.7 应用层优化
16.7.1 连接池
# ✅ 使用连接池
from contextlib import contextmanager
import sqlite3
class SQLitePool:
def __init__(self, db_path, max_connections=5):
self.db_path = db_path
self.pool = []
self.max_connections = max_connections
@contextmanager
def get_connection(self):
if self.pool:
conn = self.pool.pop()
else:
conn = sqlite3.connect(self.db_path)
conn.execute('PRAGMA journal_mode = WAL')
conn.execute('PRAGMA foreign_keys = ON')
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
if len(self.pool) < self.max_connections:
self.pool.append(conn)
else:
conn.close()
# 使用
pool = SQLitePool('mydb.db')
with pool.get_connection() as conn:
users = conn.execute('SELECT * FROM users').fetchall()
16.7.2 预编译语句缓存
# ✅ 重用 prepared statements
conn = sqlite3.connect('mydb.db')
stmt = conn.prepare('SELECT * FROM users WHERE id = ?')
for user_id in user_ids:
row = stmt.execute((user_id,)).fetchone()
16.7.3 读写分离
# 多个读连接 + 一个写连接
import threading
read_conn = sqlite3.connect('file:mydb.db?mode=ro', uri=True)
write_conn = sqlite3.connect('mydb.db')
# 读操作可以在多线程中并发（WAL 模式下）
# 写操作需要序列化
16.8 内存数据库
-- 内存数据库（最快，但数据不持久）
CREATE TABLE cache (key TEXT PRIMARY KEY, value TEXT, expires_at INTEGER);
-- 使用 TEMP 表（也存在内存中，取决于 temp_store 设置）
CREATE TEMP TABLE tmp_cache AS SELECT * FROM remote_data;
-- 附加内存数据库作为缓存
ATTACH DATABASE ':memory:' AS cache_db;
CREATE TABLE cache_db.search_cache (query TEXT PRIMARY KEY, results TEXT);
16.9 性能测量
16.9.1 计时
-- CLI 计时
.timer on
SELECT COUNT(*) FROM big_table;
-- Run Time: real 0.015 user 0.014 sys 0.001
# Python 计时
import time
start = time.perf_counter()
conn.execute('SELECT COUNT(*) FROM big_table').fetchone()
elapsed = time.perf_counter() - start
print(f"查询耗时: {elapsed*1000:.2f}ms")
16.9.2 分析工具
# sqlite3_analyzer 分析数据库结构
sqlite3_analyzer mydb.db
# 输出包含每张表、索引的页面数和空间使用
⚠️ 注意事项
synchronous = OFF有数据丢失风险——仅在可接受数据丢失时使用- mmap 在某些文件系统上可能有问题——NFS、FUSE 等
- VACUUM 需要两倍磁盘空间——大数据库需要预留空间
- ANALYZE 在大表上可能较慢——使用
analysis_limit限制采样 - 连接池大小不要超过 5-10——SQLite 的写入瓶颈不在连接数
- 不要频繁打开关闭数据库连接——连接建立有开销
💡 技巧
- 先测量再优化——用
.timer on定位瓶颈 - WAL + NORMAL 是黄金组合——适合大多数场景
- 批量操作用事务——性能提升 100 倍
- 覆盖索引是终极优化——避免回表查询
- 游标分页替代 OFFSET——大数据集分页性能差距巨大
📌 业务场景
场景一：日志采集系统
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
BEGIN;
INSERT INTO logs (level, message, source, created_at)
VALUES ('INFO', 'User login', 'auth', datetime('now'));
-- ... 批量插入
COMMIT;
场景二：API 读取优化
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = -20000;
PRAGMA mmap_size = 268435456;
-- 覆盖索引
CREATE INDEX idx_orders_user_status ON orders(user_id, status, id, total_amount);
-- 游标分页
SELECT id, status, total_amount FROM orders
WHERE user_id = ? AND id > ?
ORDER BY id LIMIT 20;
🔗 扩展阅读
📖 下一章：17 - 容器化 —— Docker 中使用 SQLite、数据持久化