数据库性能优化
要点
- 索引是数据库性能优化的第一道手段，缺索引和错索引的代价差一个数量级
- EXPLAIN 是分析查询的起点，看 actual rows、loops、filter 比看 cost 更可靠
- N+1 查询是 ORM 最常见的性能问题，关系预加载是标准解法
- 慢查询日志和
pg_stat_statements是定位瓶颈的入口 - 反范式化（冗余字段、聚合缓存）可以解决特定读热点，但写入成本和维护复杂度会上升
- 性能优化应该有测量依据，而不是凭直觉加索引或改结构
1. 索引
1.1 B-tree 索引
PostgreSQL 默认使用 B-tree 索引，适用于等值查询和范围查询：
// Drizzle
export const users = pgTable('users', {
id: uuid('id').primaryKey().defaultRandom(),
email: text('email').notNull().unique(),
name: text('name').notNull(),
createdAt: timestamp('created_at').defaultNow().notNull(),
}, (table) => ({
emailIdx: uniqueIndex('users_email_idx').on(table.email),
createdAtIdx: index('users_created_at_idx').on(table.createdAt),
}))// Prisma
model User {
id String @id @default(uuid())
email String @unique
name String
createdAt DateTime @default(now())
@@index([createdAt])
}1.2 复合索引
多列查询使用复合索引，遵循最左前缀原则：
// 查询: WHERE tenant_id = ? AND status = ? ORDER BY created_at DESC
export const conversations = pgTable('conversations', {
id: uuid('id').primaryKey().defaultRandom(),
tenantId: uuid('tenant_id').notNull(),
status: text('status').notNull(),
createdAt: timestamp('created_at').defaultNow().notNull(),
}, (table) => ({
// 索引列顺序：等值条件在前，排序条件在后
tenantStatusCreatedIdx: index('conv_tenant_status_created_idx')
.on(table.tenantId, table.status, table.createdAt),
}))最左前缀原则意味着索引 (a, b, c) 可以支持：
WHERE a = ?WHERE a = ? AND b = ?WHERE a = ? AND b = ? AND c = ?WHERE a = ? ORDER BY b
但不支持 WHERE b = ? 或 WHERE c = ?。
1.3 部分索引
只索引满足条件的行，减少索引大小：
-- 只索引未删除的记录
CREATE INDEX conversations_active_idx
ON conversations (tenant_id, created_at)
WHERE deleted_at IS NULL;// Drizzle
tenantActiveIdx: index('conv_tenant_active_idx')
.on(table.tenantId, table.createdAt)
.where(isNull(table.deletedAt)),1.4 表达式索引
对函数结果建索引：
-- 对 email 小写建索引（支持不区分大小写查询）
CREATE INDEX users_email_lower_idx ON users (LOWER(email));-- 对 JSONB 字段建索引
CREATE INDEX documents_metadata_idx ON documents USING GIN (metadata);1.5 索引的代价
索引不是免费的：
- 写入成本：每次 INSERT/UPDATE 都要维护索引
- 存储空间：索引占用磁盘空间
- 锁竞争：高并发写入时索引维护会产生锁
原则：
- 查询频率高、选择性好的列建索引
- 低选择性列（如 status 只有几个值）单独建索引效果差
- 小表（几百行）不需要索引，全表扫描更快
- 定期清理未使用的索引
2. 查询分析
2.1 EXPLAIN
使用 EXPLAIN ANALYZE 查看查询执行计划：
EXPLAIN ANALYZE
SELECT * FROM conversations
WHERE tenant_id = 'tenant-1'
AND status = 'active'
ORDER BY created_at DESC
LIMIT 10;输出示例：
Limit (cost=12.45..12.48 rows=10 width=256) (actual time=0.045..0.048 rows=10 loops=1)
-> Sort (cost=12.45..12.52 rows=28 width=256) (actual time=0.044..0.045 rows=10 loops=1)
Sort Key: created_at DESC
Sort Method: top-N heapsort Memory: 27kB
-> Index Scan using conv_tenant_status_created_idx on conversations
(cost=0.28..11.85 rows=28 width=256) (actual time=0.015..0.032 rows=28 loops=1)
Index Cond: ((tenant_id = 'tenant-1'::uuid) AND (status = 'active'::text))
Planning Time: 0.234 ms
Execution Time: 0.078 ms
关键指标：
- actual rows vs rows：估算行数与实际行数差距大说明统计信息不准
- loops：嵌套循环次数，loops 高说明有 N+1 问题
- filter：过滤掉的行数，filter 高说明索引选择性差
- Sort Method：
external merge表示排序溢出到磁盘，需要优化
2.2 常见查询模式
覆盖索引：查询的所有列都在索引中，不需要回表：
-- 索引
CREATE INDEX users_email_name_idx ON users (email, name);
-- 查询（只需要 email 和 name，覆盖索引）
SELECT email, name FROM users WHERE email = '[email protected]';避免函数包裹索引列：
-- ❌ 不走索引
SELECT * FROM users WHERE UPPER(email) = '[email protected]';
-- ✅ 使用表达式索引
CREATE INDEX users_email_lower_idx ON users (LOWER(email));
SELECT * FROM users WHERE LOWER(email) = '[email protected]';避免隐式类型转换：
-- ❌ id 是 UUID，传入字符串不走索引
SELECT * FROM users WHERE id = 123;
-- ✅ 类型匹配
SELECT * FROM users WHERE id = '123'::uuid;3. N+1 查询
3.1 问题
N+1 查询是指先查询主表（1 次），再循环查询关联表（N 次）：
// ❌ N+1 问题
const conversations = await db.select().from(conversations).limit(10)
for (const conv of conversations) {
// 每次循环都查询一次 messages
const messages = await db
.select()
.from(messages)
.where(eq(messages.conversationId, conv.id))
}10 个对话，每个查询一次 messages，总共 11 次查询。
3.2 关系预加载
Drizzle 使用 relation API 预加载：
// ✅ 使用 relation 预加载
const conversations = await db.query.conversations.findMany({
limit: 10,
with: {
messages: true,
},
})Drizzle 会自动生成 2 次查询（1 次 conversations + 1 次 messages），而不是 11 次。
Prisma 同样支持：
// ✅ Prisma include
const conversations = await prisma.conversation.findMany({
take: 10,
include: {
messages: true,
},
})3.3 手动批量查询
如果不用 relation API，可以手动批量查询：
// ✅ 手动批量查询
const conversations = await db.select().from(conversations).limit(10)
const conversationIds = conversations.map(c => c.id)
const messages = await db
.select()
.from(messages)
.where(inArray(messages.conversationId, conversationIds))
// 按 conversationId 分组
const messagesByConv = messages.reduce((acc, msg) => {
if (!acc[msg.conversationId]) acc[msg.conversationId] = []
acc[msg.conversationId].push(msg)
return acc
}, {})
// 合并
const result = conversations.map(c => ({
...c,
messages: messagesByConv[c.id] ?? [],
}))3.4 检测 N+1
开发环境可以拦截查询，检测 N+1：
// 简单检测：短时间内同一个表的查询超过阈值
const queryLog = new Map<string, number>()
db.on('query', (query) => {
const table = extractTable(query)
const key = `${table}:${Date.now() - (Date.now() % 1000)}` // 按秒分组
const count = (queryLog.get(key) ?? 0) + 1
queryLog.set(key, count)
if (count > 10) {
console.warn(`Potential N+1 query detected: ${table} queried ${count} times`)
}
})4. 慢查询分析
4.1 pg_stat_statements
PostgreSQL 的 pg_stat_statements 扩展统计所有查询的执行情况：
-- 启用扩展
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
-- 查看最慢的查询
SELECT
query,
calls,
total_exec_time / 1000 AS total_seconds,
mean_exec_time / 1000 AS avg_seconds,
rows
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 10;输出：
query | calls | total_seconds | avg_seconds | rows
-----------------------------------------------|-------|---------------|-------------|------
SELECT * FROM conversations WHERE tenant_id... | 1000 | 45.23 | 0.045 | 10
UPDATE users SET last_active_at = NOW() WHE... | 5000 | 12.56 | 0.003 | 1
4.2 慢查询日志
配置 PostgreSQL 记录慢查询：
# postgresql.conf
log_min_duration_statement = 1000 # 记录超过 1 秒的查询
log_checkpoints = on
log_connections = on
log_disconnections = on
log_lock_waits = on
log_temp_files = 0 # 记录所有临时文件（排序溢出）4.3 监控指标
生产环境应该监控：
- QPS：每秒查询数
- 平均响应时间：查询平均耗时
- 慢查询比例：超过阈值的查询占比
- 连接数：活跃连接、等待连接
- 缓存命中率：
shared_blks_hit / (shared_blks_hit + shared_blks_read)
// 从 pg_stat_database 获取缓存命中率
const stats = await db.execute(sql`
SELECT
sum(blks_hit) AS hits,
sum(blks_read) AS reads
FROM pg_stat_database
`)
const hitRate = stats[0].hits / (stats[0].hits + stats[0].reads)
console.log(`Cache hit rate: ${(hitRate * 100).toFixed(2)}%`)缓存命中率低于 99% 时，考虑增加 shared_buffers。
5. 反范式化
5.1 冗余字段
把关联查询的结果冗余存储，避免 JOIN：
// 对话表冗余消息数
export const conversations = pgTable('conversations', {
id: uuid('id').primaryKey().defaultRandom(),
title: text('title').notNull(),
messageCount: integer('message_count').default(0).notNull(), // 冗余
})
// 插入消息时更新计数
await db.insert(messages).values({ conversationId: 'conv-1', content: 'Hello' })
await db
.update(conversations)
.set({ messageCount: sql`${conversations.messageCount} + 1` })
.where(eq(conversations.id, 'conv-1'))代价：写入时需要维护冗余字段，可能不一致。
5.2 聚合缓存
把聚合查询的结果缓存起来：
// 用户统计
export const userStats = pgTable('user_stats', {
userId: uuid('user_id').primaryKey(),
totalConversations: integer('total_conversations').default(0).notNull(),
totalMessages: integer('total_messages').default(0).notNull(),
lastActiveAt: timestamp('last_active_at'),
})
// 定时更新
await db.execute(sql`
INSERT INTO user_stats (user_id, total_conversations, total_messages, last_active_at)
SELECT
u.id,
COUNT(DISTINCT c.id),
COUNT(m.id),
MAX(m.created_at)
FROM users u
LEFT JOIN conversations c ON c.user_id = u.id
LEFT JOIN messages m ON m.conversation_id = c.id
GROUP BY u.id
ON CONFLICT (user_id) DO UPDATE SET
total_conversations = EXCLUDED.total_conversations,
total_messages = EXCLUDED.total_messages,
last_active_at = EXCLUDED.last_active_at
`)5.3 适用场景
反范式化适合：
- 读多写少：聚合结果变化不频繁
- 查询热点：某个查询特别慢，优化索引无效
- 展示数据：统计面板、用户主页
不适合：
- 强一致性要求：冗余数据可能不一致
- 写入频繁：维护成本高
6. 其他优化手段
6.1 分区表
大表按时间或租户分区：
-- 按月分区
CREATE TABLE messages (
id UUID PRIMARY KEY,
conversation_id UUID NOT NULL,
content TEXT NOT NULL,
created_at TIMESTAMP NOT NULL
) PARTITION BY RANGE (created_at);
CREATE TABLE messages_2026_01 PARTITION OF messages
FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
CREATE TABLE messages_2026_02 PARTITION OF messages
FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');6.2 读写分离
主库写入，从库查询：
const writeDb = drizzle(pool) // 主库
const readDb = drizzle(readPool) // 从库
// 写入
await writeDb.insert(conversations).values({ title: 'New' })
// 查询（从从库读）
const convs = await readDb.select().from(conversations)6.3 查询缓存
应用层缓存查询结果：
import { Redis } from 'ioredis'
const redis = new Redis(process.env.REDIS_URL!)
async function getUserWithCache(userId: string) {
const cached = await redis.get(`user:${userId}`)
if (cached) return JSON.parse(cached)
const user = await db.query.users.findFirst({
where: eq(users.id, userId),
})
await redis.setex(`user:${userId}`, 3600, JSON.stringify(user))
return user
}总结
数据库性能优化是一个系统性的工作，需要有测量依据，而不是凭直觉。
这一节涉及到的几个实践：
- 索引：B-tree、复合索引、部分索引、表达式索引
- 查询分析：EXPLAIN ANALYZE、常见查询模式
- N+1 查询：关系预加载、批量查询
- 慢查询分析：pg_stat_statements、慢查询日志
- 反范式化：冗余字段、聚合缓存
- 分区表：大表按时间分区
- 读写分离：主库写入，从库查询
- 查询缓存：应用层缓存
索引是第一道手段，EXPLAIN 是分析的起点，N+1 是 ORM 的常见问题，慢查询日志是定位瓶颈的入口。
下一篇看 AI 项目中的数据库实践——对话存储、向量检索、多租户隔离。