练习题

推荐方案：混合数据库方案

┌─────────────────────────────────────┐
│  应用层                              │
└──────┬────────────────────────┬─────┘
       │                        │
       ▼                        ▼
┌─────────────┐         ┌──────────────┐
│  MySQL      │         │  MongoDB     │
│  (主数据库) │         │  (日志存储)  │
│             │         │              │
│  - 用户表   │         │  - 浏览日志  │
│  - 商品表   │         │  (30 天自动过期)│
│  - 订单表   │         │              │
│  - 库存表   │         │              │
└─────────────┘         └──────────────┘
       │
       ▼
┌─────────────┐
│  Redis      │
│  (缓存层)   │
│             │
│  - 热点商品 │
│  - 会话缓存 │
│  - 库存计数 │
└─────────────┘

理由分析：

1. MySQL：

   • 订单需要事务保证（ACID）
   • 复杂的 JOIN 查询需求
   • 用户、商品数据结构化程度高

2. MongoDB：

   • 浏览日志写多读少
   • 数据结构灵活
   • 可设置 TTL 自动过期

3. Redis：

   • 缓存热点商品数据
   • 库存计数器（高并发）
   • 用户会话存储

1. 问题分析：

缺少的关键代码：

# 开启事务
db.begin()

# 异常处理
try:
    # ... 转账操作
    db.commit()
except Exception as e:
    db.rollback()  # 回滚
    raise

后果：如果步骤 2 执行后系统崩溃，A 的钱扣了但 B 没收到。数据不一致！

2. ACID 详解：

3. 符合 ACID 的代码：

def transfer(user_a, user_b, amount):
    cursor = db.cursor()
    try:
        db.begin()  # 开启事务

        # 检查余额并加锁
        cursor.execute(
            'SELECT balance FROM users WHERE id = %s FOR UPDATE',
            (user_a,)
        )
        balance_a = cursor.fetchone()['balance']

        if balance_a < amount:
            db.rollback()
            return {'success': False, 'message': '余额不足'}

        # A 扣钱
        cursor.execute(
            'UPDATE users SET balance = balance - %s WHERE id = %s',
            (amount, user_a)
        )

        # B 加钱
        cursor.execute(
            'UPDATE users SET balance = balance + %s WHERE id = %s',
            (amount, user_b)
        )

        db.commit()
        return {'success': True}

    except Exception as e:
        db.rollback()
        return {'success': False, 'message': str(e)}

关键点：

推荐索引方案：

-- users 表
CREATE INDEX idx_users_email ON users(email);
-- 理由：根据邮箱查找是常见操作

CREATE INDEX idx_users_created ON users(created_at);
-- 理由：可能需要按注册时间筛选用户

-- orders 表
CREATE INDEX idx_orders_user_id ON orders(user_id);
-- 理由：根据用户 ID 查订单（外键查询）

CREATE INDEX idx_orders_created ON orders(created_at);
-- 理由：时间范围查询频繁

CREATE INDEX idx_orders_user_created ON orders(user_id, created_at);
-- 理由：复合索引，优化"查询某用户某时间段的订单"

CREATE INDEX idx_orders_status ON orders(status);
-- 理由：可能需要按订单状态筛选

索引选择原则：

• ✅ 高频查询的 WHERE 条件字段
• ✅ JOIN 操作的外键字段
• ✅ 范围查询的字段（时间）
• ⚠️ 避免过度索引（影响写入性能）
• ⚠️ 区分度低的字段不适合索引（如性别）

1. 缓存方案：

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_user_profile(user_id):
    cache_key = f'user:profile:{user_id}'

    # 先查缓存
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # 缓存未命中，查数据库
    cursor = db.cursor()
    cursor.execute(
        'SELECT * FROM users WHERE id = %s',
        (user_id,)
    )
    user = cursor.fetchone()

    if user:
        # 存入缓存，过期时间 30 分钟
        redis_client.setex(cache_key, 1800, json.dumps(user))

    return user

2. 缓存更新策略：

def update_user_profile(user_id, data):
    # 1. 更新数据库
    cursor = db.cursor()
    cursor.execute(
        'UPDATE users SET email=%s, name=%s WHERE id=%s',
        (data['email'], data['name'], user_id)
    )
    db.commit()

    # 2. 删除缓存（下次读取时重新加载）
    cache_key = f'user:profile:{user_id}'
    redis_client.delete(cache_key)

3. 过期时间策略：

1. 归档方案：

采用分层归档策略：

┌─────────────────────────────────────┐
│  热数据层（最近 30 天）              │
│  ├─ MySQL 主表                      │
│  └─ 保持索引，支持快速查询          │
├─────────────────────────────────────┤
│  温数据层（30-90 天）               │
│  ├─ MySQL 归档表（按月分区）         │
│  └─ 从库查询，减轻主库压力          │
├─────────────────────────────────────┤
│  冷数据层（90 天以上）              │
│  ├─ CSV 文件 + 对象存储              │
│  └─ 按需加载查询                    │
└─────────────────────────────────────┘

归档脚本：

from datetime import datetime, timedelta

def archive_old_logs(days: int = 30):
    cutoff_date = datetime.now() - timedelta(days=days)
    month_str = cutoff_date.strftime('%Y%m')

    with get_db_cursor() as cursor:
        # 创建归档表
        cursor.execute(f'''
            CREATE TABLE IF NOT EXISTS api_logs_{month_str}
            LIKE api_logs
        ''')

        # 移动数据
        cursor.execute(f'''
            INSERT INTO api_logs_{month_str}
            SELECT * FROM api_logs
            WHERE created_at < %s
        ''', (cutoff_date,))

        # 删除已归档的数据
        cursor.execute(
            'DELETE FROM api_logs WHERE created_at < %s',
            (cutoff_date,)
        )

2. 归档周期选择：

3. 历史数据查询：

def get_historical_logs(user_id: int, start_date: date, end_date: date):
    """查询历史日志（可能跨越多个归档表）"""
    results = []

    # 查询主表
    with get_db_cursor() as cursor:
        cursor.execute('''
            SELECT * FROM api_logs
            WHERE user_id = %s AND created_at BETWEEN %s AND %s
        ''', (user_id, start_date, end_date))
        results.extend(cursor.fetchall())

    # 查询归档表
    current = start_date
    while current <= end_date:
        month = current.strftime('%Y%m')
        try:
            with get_db_cursor() as cursor:
                cursor.execute(f'''
                    SELECT * FROM api_logs_{month}
                    WHERE user_id = %s AND created_at BETWEEN %s AND %s
                ''', (user_id, start_date, end_date))
                results.extend(cursor.fetchall())
        except:
            pass  # 归档表可能不存在
        # 移动到下一月
        current = (current.replace(day=1) + timedelta(days=32)).replace(day=1)

    return sorted(results, key=lambda x: x['created_at'])