Name: 数据库管理Skill
Rating: 5 (1 reviews)

name: 数据库 description: 数据库优化、查询调优、迁移和管理。用于数据库性能问题、模式设计或操作任务。

数据库管理

优化查询、管理模式并确保可靠性。

何时使用

慢查询优化
模式设计和迁移
索引策略
数据库操作
性能调优

查询优化

分析查询

-- PostgreSQL
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM orders WHERE user_id = 123;

-- MySQL
EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 123;

常见优化

-- 添加缺失索引
CREATE INDEX CONCURRENTLY idx_orders_user_id
ON orders(user_id);

-- 复合索引用于常见查询
CREATE INDEX idx_orders_user_date
ON orders(user_id, created_at DESC);

-- 部分索引用于过滤查询
CREATE INDEX idx_active_orders
ON orders(user_id) WHERE status = 'active';

-- 覆盖索引以避免表查找
CREATE INDEX idx_orders_covering
ON orders(user_id) INCLUDE (total, status);

迁移最佳实践

-- 安全列添加（无锁）
ALTER TABLE users ADD COLUMN preferences JSONB;

-- 安全列重命名（使用视图兼容性）
ALTER TABLE users RENAME COLUMN name TO full_name;
CREATE VIEW users_compat AS
  SELECT *, full_name as name FROM users;

-- 安全索引创建
CREATE INDEX CONCURRENTLY idx_new ON table(column);

-- 批量回填
UPDATE users SET new_col = compute(old_col)
WHERE id BETWEEN 1 AND 10000;
-- 重复下一批

操作查询

健康检查

-- 活动连接（PostgreSQL）
SELECT state, count(*)
FROM pg_stat_activity
GROUP BY state;

-- 长时间运行查询
SELECT pid, now() - query_start as duration, query
FROM pg_stat_activity
WHERE state = 'active' AND now() - query_start > '5 minutes'::interval;

-- 表大小
SELECT relname, pg_size_pretty(pg_total_relation_size(relid))
FROM pg_catalog.pg_statio_user_tables
ORDER BY pg_total_relation_size(relid) DESC
LIMIT 10;

-- 索引使用
SELECT indexrelname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
ORDER BY idx_scan DESC;

维护

-- PostgreSQL 清理和分析
VACUUM ANALYZE table_name;

-- 重新索引
REINDEX INDEX CONCURRENTLY idx_name;

-- 终止长时间查询
SELECT pg_terminate_backend(pid);

缓存策略

import redis

cache = redis.Redis()

def get_user(user_id: int) -> dict:
    # 首先尝试缓存
    cached = cache.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    # 查询数据库
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)

    # 使用 TTL 缓存
    cache.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

def invalidate_user(user_id: int):
    cache.delete(f"user:{user_id}")

示例

输入: “这个查询很慢” 操作: 运行 EXPLAIN，识别缺失索引或不良计划，优化