07-数据库优化与调优¶

# 数据库优化标准流程
class DatabaseOptimization:
    """
    数据库优化流程
    """
    def optimization_workflow(self):
        steps = [
            "1. 监控与诊断：收集性能指标，定位瓶颈",
            "2. 分析原因：使用EXPLAIN、慢查询日志等工具",
            "3. 制定方案：索引优化、SQL改写、配置调整",
            "4. 实施优化：逐步应用优化措施",
            "5. 验证效果：对比优化前后的性能指标",
            "6. 持续监控：建立长期监控机制"
        ]
        return steps

# SQLAlchemy连接池配置
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

# 生产环境推荐配置
engine = create_engine(
    'mysql+pymysql://user:pass@localhost/dbname',
    poolclass=QueuePool,
    pool_size=10,           # 常驻连接数
    max_overflow=20,        # 最大额外连接
    pool_timeout=30,        # 获取连接超时时间
    pool_recycle=3600,      # 连接回收时间（防止超时）
    pool_pre_ping=True,     # 连接前ping检测
    echo=False              # 关闭SQL日志（生产环境）
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

# PostgreSQL连接池（使用psycopg2）
engine_pg = create_engine(
    'postgresql+psycopg2://user:pass@localhost/dbname',
    pool_size=20,
    max_overflow=30,
    pool_timeout=30,
    pool_recycle=3600,
    pool_pre_ping=True,
    connect_args={
        'connect_timeout': 10,
        'options': '-c statement_timeout=30000'  # 30秒查询超时
    }
)

# 异步连接池（asyncpg）
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/dbname',
    pool_size=20,
    max_overflow=30,
    pool_timeout=30,
    pool_recycle=3600,
    echo=False
)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 连接池监控工具
class ConnectionPoolMonitor:
    """连接池监控器"""

    def __init__(self, engine):
        self.engine = engine
        self.pool = engine.pool

    def get_pool_stats(self):
        """获取连接池统计信息"""
        return {
            'size': self.pool.size(),           # 当前连接数
            'checked_in': self.pool.checkedin(), # 空闲连接
            'checked_out': self.pool.checkedout(), # 使用中连接
            'overflow': self.pool.overflow(),    # 超出pool_size的连接
        }

    def log_pool_status(self):
        """记录连接池状态"""
        stats = self.get_pool_stats()
        print(f"连接池状态: {stats}")

        # 告警：使用率过高
        usage_rate = stats['checked_out'] / (stats['size'] or 1)
        if usage_rate > 0.8:
            print(f"⚠️ 警告：连接池使用率 {usage_rate:.1%}，考虑增加pool_size")

        return stats

# 使用示例
monitor = ConnectionPoolMonitor(engine)
monitor.log_pool_status()

import functools
import hashlib
import json
from typing import Optional, Any
import redis

class DatabaseCache:
    """数据库查询缓存装饰器"""

    def __init__(self, redis_client: redis.Redis,
                 default_ttl: int = 300,
                 key_prefix: str = "db_cache"):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.key_prefix = key_prefix

    def _generate_key(self, func_name: str, args: tuple, kwargs: dict) -> str:
        """生成缓存key"""
        cache_key = f"{func_name}:{str(args)}:{str(kwargs)}"
        hash_key = hashlib.md5(cache_key.encode()).hexdigest()
        return f"{self.key_prefix}:{hash_key}"

    def cache_query(self, ttl: Optional[int] = None):
        """查询缓存装饰器"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
                cache_key = self._generate_key(func.__name__, args, kwargs)

                # 尝试从缓存获取
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)

                # 执行查询
                result = func(*args, **kwargs)

                # 写入缓存
                self.redis.setex(
                    cache_key,
                    ttl or self.default_ttl,
                    json.dumps(result, default=str)
                )

                return result
            return wrapper
        return decorator

    def invalidate_cache(self, pattern: str = "*"):
        """清除缓存"""
        keys = self.redis.keys(f"{self.key_prefix}:{pattern}")
        if keys:
            self.redis.delete(*keys)

# 使用示例
redis_client = redis.Redis(host='localhost', port=6379, db=0)
cache = DatabaseCache(redis_client, default_ttl=600)

class UserService:
    @cache.cache_query(ttl=300)
    def get_user_stats(self, user_id: int):
        """获取用户统计（缓存5分钟）"""
        # 复杂的数据库查询
        result = db.execute("""
            SELECT u.*, COUNT(o.order_id) as order_count
            FROM users u
            LEFT JOIN orders o ON u.user_id = o.user_id
            WHERE u.user_id = %s
            GROUP BY u.user_id
        """, (user_id,))
        return result.fetchone()

    @cache.cache_query(ttl=60)
    def get_model_predictions(self, model_id: str, limit: int = 100):
        """获取模型预测结果（缓存1分钟）"""
        result = db.execute("""
            SELECT * FROM predictions
            WHERE model_id = %s
            ORDER BY created_at DESC
            LIMIT %s
        """, (model_id, limit))
        return result.fetchall()

class CacheStrategy:
    """缓存策略管理"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def cache_aside(self, key: str, db_query_func, ttl: int = 300):
        """旁路缓存（Cache-Aside）策略"""
        # 1. 先查缓存
        value = self.redis.get(key)
        if value:
            return json.loads(value)  # json.loads将JSON字符串转为Python对象

        # 2. 缓存未命中，查数据库
        value = db_query_func()

        # 3. 写入缓存
        if value:
            self.redis.setex(key, ttl, json.dumps(value, default=str))  # json.dumps将Python对象转为JSON字符串

        return value

    def write_through(self, key: str, value: Any,
                      db_write_func, ttl: int = 300):
        """直写（Write-Through）策略"""
        # 1. 先写数据库
        db_write_func(value)

        # 2. 同步写缓存
        self.redis.setex(key, ttl, json.dumps(value, default=str))

    def write_behind(self, key: str, value: Any, ttl: int = 300):
        """回写（Write-Behind）策略"""
        # 1. 先写缓存
        self.redis.setex(key, ttl, json.dumps(value, default=str))

        # 2. 异步写数据库（使用消息队列）
        # message_queue.publish('db_write_queue', {'key': key, 'value': value})
        pass

    def invalidate_on_write(self, key_pattern: str):
        """写入时失效策略"""
        # 数据更新时，删除相关缓存
        keys = self.redis.keys(key_pattern)
        if keys:
            self.redis.delete(*keys)

# 批量插入优化（模型训练数据）
from sqlalchemy import insert
import time

def batch_insert_optimized(session, table, data_list, batch_size=10000):
    """
    优化的批量插入

    Args:
        session: 数据库会话
        table: 表对象
        data_list: 数据列表
        batch_size: 每批大小
    """
    total = len(data_list)
    inserted = 0
    start_time = time.time()

    try:
        for i in range(0, total, batch_size):
            batch = data_list[i:i + batch_size]

            # 使用批量插入
            session.execute(
                insert(table),
                batch
            )

            # 每10批提交一次
            if (i // batch_size + 1) % 10 == 0:
                session.commit()

            inserted += len(batch)

            if i % (batch_size * 10) == 0:
                elapsed = time.time() - start_time
                rate = inserted / elapsed if elapsed > 0 else 0
                print(f"已插入: {inserted}/{total}, 速率: {rate:.0f} 条/秒")

        # 最后提交
        session.commit()

    except Exception as e:
        session.rollback()
        raise e

    elapsed = time.time() - start_time
    print(f"完成！共插入 {inserted} 条，耗时 {elapsed:.2f} 秒，"
          f"平均 {inserted/elapsed:.0f} 条/秒")

# MySQL专用：LOAD DATA优化
def bulk_load_mysql(connection, table_name, csv_file):
    """使用LOAD DATA进行超大批量导入

    安全提示：table_name 和 csv_file 应来自受信内部配置，
    不可直接接受用户输入，以防止SQL注入。
    DDL语句（ALTER TABLE等）无法使用参数化查询，
    生产环境应通过白名单验证表名。
    """
    cursor = connection.cursor()

    # 临时禁用索引和检查
    cursor.execute(f"ALTER TABLE {table_name} DISABLE KEYS")
    cursor.execute("SET FOREIGN_KEY_CHECKS = 0")
    cursor.execute("SET UNIQUE_CHECKS = 0")
    cursor.execute("SET AUTOCOMMIT = 0")

    # 执行LOAD DATA
    cursor.execute(f"""
        LOAD DATA LOCAL INFILE '{csv_file}'
        INTO TABLE {table_name}
        FIELDS TERMINATED BY ','
        ENCLOSED BY '"'
        LINES TERMINATED BY '\n'
        IGNORE 1 ROWS
    """)

    # 恢复设置
    cursor.execute("COMMIT")
    cursor.execute("SET FOREIGN_KEY_CHECKS = 1")
    cursor.execute("SET UNIQUE_CHECKS = 1")
    cursor.execute(f"ALTER TABLE {table_name} ENABLE KEYS")

    cursor.close()

# 特征存储优化方案
class FeatureStore:
    """特征存储优化"""

    def __init__(self, db_engine, redis_client):
        self.db = db_engine
        self.cache = redis_client

    def get_features_online(self, entity_id: str, feature_names: list):
        """
        在线特征获取（低延迟）

        策略：
        1. 优先从Redis获取
        2. 缓存未命中时从数据库获取并缓存
        """
        cache_key = f"features:{entity_id}"

        # 尝试从缓存获取
        cached = self.cache.hmget(cache_key, feature_names)
        if all(cached):
            return dict(zip(feature_names, cached))  # zip并行遍历多个可迭代对象

        # 缓存未命中，查询数据库
        missing_features = [
            name for name, val in zip(feature_names, cached) if val is None
        ]

        query = f"""
            SELECT {', '.join(missing_features)}
            FROM feature_store
            WHERE entity_id = %s
        """
        result = self.db.execute(query, (entity_id,)).fetchone()

        if result:
            # 更新缓存
            feature_dict = dict(result)
            self.cache.hset(cache_key, mapping=feature_dict)
            self.cache.expire(cache_key, 3600)  # 1小时过期

            # 合并结果
            return {**dict(zip(feature_names, cached)), **feature_dict}

        return None

    def get_features_offline(self, entity_ids: list, feature_names: list):
        """
        离线特征获取（批量）

        策略：
        1. 使用批量查询
        2. 分区并行读取
        """
        # 分批查询避免超长IN列表
        batch_size = 1000
        all_results = []

        for i in range(0, len(entity_ids), batch_size):
            batch_ids = entity_ids[i:i + batch_size]

            query = f"""
                SELECT entity_id, {', '.join(feature_names)}
                FROM feature_store
                WHERE entity_id = ANY(%s)
            """

            result = self.db.execute(query, (batch_ids,))
            all_results.extend(result.fetchall())

        return all_results

# 模型推理服务的数据库优化
import asyncio
from concurrent.futures import ThreadPoolExecutor
from sqlalchemy import text

class ModelServingDB:
    """模型服务数据库优化"""

    def __init__(self, engine):
        self.engine = engine
        self.executor = ThreadPoolExecutor(max_workers=10)

    async def get_prediction_async(self, model_id: str, input_hash: str):  # async def定义异步函数；用await调用
        """异步获取预测结果"""
        loop = asyncio.get_event_loop()

        def _query():
            with self.engine.connect() as conn:
                result = conn.execute(text("""
                    SELECT prediction, confidence, latency_ms
                    FROM model_predictions
                    WHERE model_id = :model_id AND input_hash = :input_hash
                    AND created_at > NOW() - INTERVAL '1 hour'
                """), {"model_id": model_id, "input_hash": input_hash})
                return result.fetchone()

        # 在线程池中执行同步查询
        return await loop.run_in_executor(self.executor, _query)  # await等待异步操作完成

    def batch_log_predictions(self, predictions: list):
        """批量记录预测日志（异步写入）"""
        # 使用COPY或批量INSERT
        # 这里使用后台线程异步写入
        def _insert():
            with self.engine.connect() as conn:
                for pred in predictions:
                    conn.execute(text("""
                        INSERT INTO prediction_logs
                        (model_id, input_data, output_data, latency_ms, created_at)
                        VALUES (:model_id, :input_data, :output_data, :latency_ms, NOW())
                    """), pred)
                conn.commit()

        # 提交到线程池，不阻塞主线程
        self.executor.submit(_insert)

# 慢查询日志分析工具
import re
from collections import defaultdict  # defaultdict带默认值的字典，避免KeyError
from datetime import datetime

class SlowQueryAnalyzer:
    """慢查询日志分析器"""

    def __init__(self):
        self.queries = []

    def parse_mysql_slow_log(self, log_file):
        """解析MySQL慢查询日志"""
        pattern = re.compile(
            r'# Time: (?P<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z?)\n'
            r'# User@Host: .*\n'
            r'# Query_time: (?P<query_time>[\d.]+) .*\n'
            r'(?P<sql>SELECT|INSERT|UPDATE|DELETE.*?)\n(?=# Time:|$)',
            re.DOTALL | re.IGNORECASE
        )

        with open(log_file, 'r') as f:  # with自动管理资源，确保文件正确关闭
            content = f.read()

        for match in pattern.finditer(content):
            self.queries.append({
                'time': match.group('time'),
                'query_time': float(match.group('query_time')),
                'sql': match.group('sql').strip()
            })

        return self.analyze()

    def analyze(self):
        """分析慢查询"""
        if not self.queries:
            return "没有找到慢查询"

        # 按SQL指纹分组
        fingerprint_stats = defaultdict(lambda: {  # lambda匿名函数：简洁的单行函数
            'count': 0,
            'total_time': 0,
            'max_time': 0,
            'samples': []
        })

        for q in self.queries:
            # 生成SQL指纹（去除具体值）
            fingerprint = self._fingerprint(q['sql'])

            stats = fingerprint_stats[fingerprint]
            stats['count'] += 1
            stats['total_time'] += q['query_time']
            stats['max_time'] = max(stats['max_time'], q['query_time'])
            if len(stats['samples']) < 3:
                stats['samples'].append(q['sql'])

        # 按总耗时排序
        sorted_queries = sorted(
            fingerprint_stats.items(),
            key=lambda x: x[1]['total_time'],
            reverse=True
        )

        # 生成报告
        report = ["=== 慢查询分析报告 ===\n"]
        for i, (fingerprint, stats) in enumerate(sorted_queries[:10], 1):  # enumerate同时获取索引和值
            avg_time = stats['total_time'] / stats['count']
            report.append(
                f"\nTop {i}:"
                f"\n  SQL指纹: {fingerprint[:100]}..."
                f"\n  执行次数: {stats['count']}"
                f"\n  总耗时: {stats['total_time']:.2f}s"
                f"\n  平均耗时: {avg_time:.2f}s"
                f"\n  最大耗时: {stats['max_time']:.2f}s"
                f"\n  示例SQL: {stats['samples'][0][:100]}..."
            )

        return '\n'.join(report)

    def _fingerprint(self, sql: str) -> str:
        """生成SQL指纹"""
        # 移除注释
        sql = re.sub(r'/\*.*?\*/', '', sql)
        sql = re.sub(r'--.*?\n', '', sql)

        # 替换字符串和数字
        sql = re.sub(r"'[^']*'", '?', sql)
        sql = re.sub(r'\b\d+\b', '?', sql)

        # 规范化空白
        sql = ' '.join(sql.split())

        return sql.lower()

# 使用示例
analyzer = SlowQueryAnalyzer()
report = analyzer.parse_mysql_slow_log('/var/log/mysql/slow.log')
print(report)

# 实时性能监控
# 需要安装: pip install psutil
import time
import psutil
from dataclasses import dataclass
from typing import List
import threading  # 线程池/多线程：并发执行任务
from sqlalchemy import text

@dataclass  # 自动生成__init__等方法
class DBMetrics:
    """数据库指标"""
    timestamp: float
    connections_active: int
    connections_total: int
    queries_per_second: float
    slow_queries: int
    cache_hit_ratio: float
    disk_io_read: float
    disk_io_write: float

class DBPerformanceMonitor:
    """数据库性能监控器"""

    def __init__(self, engine, interval: int = 5):
        self.engine = engine
        self.interval = interval
        self.metrics_history: List[DBMetrics] = []
        self._running = False
        self._thread = None

    def start(self):
        """启动监控"""
        self._running = True
        self._thread = threading.Thread(target=self._monitor_loop)
        self._thread.daemon = True
        self._thread.start()
        print("性能监控已启动")

    def stop(self):
        """停止监控"""
        self._running = False
        if self._thread:
            self._thread.join()
        print("性能监控已停止")

    def _monitor_loop(self):
        """监控循环"""
        last_queries = 0

        while self._running:
            try:  # try/except捕获异常
                with self.engine.connect() as conn:
                    # 获取MySQL状态
                    result = conn.execute(text("SHOW GLOBAL STATUS"))
                    status = {row[0]: row[1] for row in result}

                    # 计算QPS
                    current_queries = int(status.get('Queries', 0))
                    qps = (current_queries - last_queries) / self.interval
                    last_queries = current_queries

                    # 获取连接数
                    conn_result = conn.execute(text("SHOW PROCESSLIST"))
                    active_connections = conn_result.rowcount

                    # 计算缓冲池命中率（MySQL 8.0+ 使用InnoDB缓冲池指标）
                    buffer_reads = int(status.get('Innodb_buffer_pool_reads', 0))
                    buffer_requests = int(status.get('Innodb_buffer_pool_read_requests', 1))
                    cache_ratio = 1 - (buffer_reads / buffer_requests) if buffer_requests > 0 else 0

                    metrics = DBMetrics(
                        timestamp=time.time(),
                        connections_active=active_connections,
                        connections_total=int(status.get('Threads_connected', 0)),
                        queries_per_second=qps,
                        slow_queries=int(status.get('Slow_queries', 0)),
                        cache_hit_ratio=cache_ratio,
                        disk_io_read=psutil.disk_io_counters().read_bytes,
                        disk_io_write=psutil.disk_io_counters().write_bytes
                    )

                    self.metrics_history.append(metrics)

                    # 保持最近100个数据点
                    if len(self.metrics_history) > 100:
                        self.metrics_history.pop(0)

                    # 告警检查
                    self._check_alerts(metrics)

            except Exception as e:
                print(f"监控异常: {e}")

            time.sleep(self.interval)

    def _check_alerts(self, metrics: DBMetrics):
        """检查告警条件"""
        alerts = []

        if metrics.connections_active > 400:
            alerts.append(f"⚠️ 连接数过高: {metrics.connections_active}")

        if metrics.queries_per_second > 10000:
            alerts.append(f"⚠️ QPS过高: {metrics.queries_per_second:.0f}")

        if metrics.cache_hit_ratio < 0.8:
            alerts.append(f"⚠️ 缓存命中率低: {metrics.cache_hit_ratio:.1%}")

        for alert in alerts:
            print(alert)

    def get_summary(self) -> str:
        """获取监控摘要"""
        if not self.metrics_history:
            return "暂无监控数据"

        recent = self.metrics_history[-10:]
        avg_qps = sum(m.queries_per_second for m in recent) / len(recent)
        max_connections = max(m.connections_active for m in recent)

        return f"""
=== 性能监控摘要 ===
最近 {len(recent)} 次采样:
- 平均QPS: {avg_qps:.1f}
- 最大连接数: {max_connections}
- 当前缓存命中率: {recent[-1].cache_hit_ratio:.1%}
- 慢查询总数: {recent[-1].slow_queries}
        """