数据库实战指南：从索引优化到分布式架构

-- 普通索引
CREATE INDEX idx_users_email ON users(email);

-- 唯一索引
CREATE UNIQUE INDEX idx_users_username ON users(username);

-- 复合索引
CREATE INDEX idx_users_name_age ON users(name, age);

-- 全文索引
CREATE FULLTEXT INDEX idx_posts_content ON posts(content);

CREATE INDEX idx_users_name_age_gender ON users(name, age, gender);

-- 能完整使用索引
WHERE name = 'Alice' AND age = 30 AND gender = 'F'

-- 能使用部分索引
WHERE name = 'Alice' AND age = 30
WHERE name = 'Alice'

-- 不能使用索引（不满足最左前缀）
WHERE age = 30
WHERE gender = 'F'

-- 1. LIKE 前导通配符
WHERE name LIKE '%Alice%'   -- 失效
WHERE name LIKE 'Alice%'    -- 有效

-- 2. 索引列上计算
WHERE YEAR(created_at) = 2023           -- 失效
WHERE created_at BETWEEN '2023-01-01' AND '2023-12-31'  -- 有效

-- 3. 隐式类型转换
WHERE phone_number = 1234567890      -- 失效（字段是字符串）
WHERE phone_number = '1234567890'    -- 有效

-- 4. OR 条件（部分字段无索引）
WHERE name = 'Alice' OR age = 30     -- 全表扫描

-- 选择性 = COUNT(DISTINCT col) / COUNT(*)
-- 接近 1 适合建索引，接近 0 不适合
SELECT COUNT(DISTINCT email) / COUNT(*) FROM users;     -- ~1.0 适合
SELECT COUNT(DISTINCT gender) / COUNT(*) FROM users;    -- ~0.0001 不适合

-- 查询 1：WHERE name = ? AND age = ?
-- 查询 2：WHERE name = ?

-- 正确：name 在前
CREATE INDEX idx_users_name_age ON users(name, age);

-- 错误：age 在前，查询 2 用不上
CREATE INDEX idx_users_age_name ON users(age, name);

# 定期重建索引
0 0 1 * * mysql -u root -p -e "OPTIMIZE TABLE users;"

spring:
  datasource:
    hikari:
      maximum-pool-size: 32          # 8核机器推荐
      minimum-idle: 32               # 保持所有连接常驻
      connection-timeout: 10000      # 10秒获取超时
      idle-timeout: 300000           # 5分钟空闲超时
      max-lifetime: 1500000          # 25分钟（< MySQL wait_timeout）
      leak-detection-threshold: 60000  # 60秒泄漏检测
      validation-timeout: 3000
      connection-test-query: SELECT 1
      pool-name: BusinessHikariCP

// 错误：return 时连接没关闭
public void processData() {
    Connection conn = dataSource.getConnection();
    try {
        if (someCondition) return;  // 泄漏！
    } finally {
        // 忘了 conn.close()
    }
}

// 正确：try-with-resources
try (Connection conn = dataSource.getConnection()) {
    // ...
}

// 错误：事务里有 HTTP 调用
@Transactional
public void longTask() {
    List<Data> data = repo.findAll();
    for (Data item : data) {
        externalApiCall(item);  // 几秒一个，连接一直占着
    }
}

// 正确：拆分事务
public void longTask() {
    List<Data> data = repo.findAll();
    for (Data item : data) {
        externalApiCall(item);
        repo.updateStatus(item.getId());  // 每次单独事务
    }
}

-- MySQL 默认 wait_timeout = 28800 (8小时)
SHOW VARIABLES LIKE 'wait_timeout';

@Configuration
public class MetricsConfig {
    @Bean
    public MeterRegistryCustomizer<HikariDataSource> metrics() {
        return (dataSource, registry) -> {
            HikariPoolMXBean pool = dataSource.getHikariPoolMXBean();
            registry.gauge("hikari.active", pool, HikariPoolMXBean::getActiveConnections);
            registry.gauge("hikari.idle", pool, HikariPoolMXBean::getIdleConnections);
            registry.gauge("hikari.threads.awaiting", pool, HikariPoolMXBean::getThreadsAwaitingConnection);
        };
    }
}

- alert: HikariPoolNearlyFull
  expr: hikari_active / hikari_total > 0.8
  for: 5m

- alert: HikariConnectionWait
  expr: hikari_threads_awaiting > 5
  for: 2m

-- MySQL InnoDB 默认 RR
SELECT @@transaction_isolation;  -- REPEATABLE-READ

-- PostgreSQL 默认 RC
SHOW transaction_isolation;      -- read committed

-- 会话 1
SET SESSION transaction_isolation = 'READ-COMMITTED';
START TRANSACTION;
SELECT qty FROM stock WHERE id = 1001;  -- 100

-- 会话 2
UPDATE stock SET qty = 97 WHERE id = 1001;
COMMIT;

-- 回到会话 1
SELECT qty FROM stock WHERE id = 1001;  -- 97，不一致
COMMIT;

@Transactional(isolation = Isolation.READ_COMMITTED)
@Retryable(value = DeadlockLoserDataAccessException.class, maxAttempts = 3)
public void deductStock(Long productId, int qty) {
    Stock row = stockRepo.findByIdForUpdate(productId);  // SELECT ... FOR UPDATE
    if (row.getAvailable() < qty) {
        throw new InsufficientStockException();
    }
    row.setAvailable(row.getAvailable() - qty);
}

-- 错误：事务里做 Excel 生成
START TRANSACTION;
SELECT ...;
-- 生成 Excel（几十秒）
COMMIT;

-- 正确：事务里只做数据库操作
START TRANSACTION;
SELECT ...;
COMMIT;
-- 事务外做 Excel

# my.cnf 主库
[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
gtid_mode = ON
enforce_gtid_consistency = ON
sync_binlog = 1
innodb_flush_log_at_trx_commit = 1

-- 创建复制用户
CREATE USER 'repl'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';
FLUSH PRIVILEGES;

-- 查看主库状态
SHOW MASTER STATUS;

# my.cnf 从库
[mysqld]
server-id = 2
relay_log = relay-bin
read_only = ON
super_read_only = ON
relay_log_purge = OFF

-- GTID 方式连接主库（推荐）
CHANGE MASTER TO
  MASTER_HOST='master-ip',
  MASTER_USER='repl',
  MASTER_PASSWORD='password',
  MASTER_AUTO_POSITION=1;

START SLAVE;

-- 查看复制状态
SHOW SLAVE STATUS\G

-- 从库可能落后主库几秒到几分钟
-- 对实时性要求高的读，强制走主库

-- NOW()、UUID()、RAND() 等在主从上结果不同
-- 使用 binlog_format = ROW 避免

def get_db_index(user_id):
    return user_id % 4

def get_table_index(user_id):
    return (user_id // 4) % 8

def migrate_batch(start_id, batch_size, source, targets):
    """批量迁移，按 ID 范围分批"""
    cursor = source.cursor()
    cursor.execute("SELECT * FROM orders WHERE user_id >= %s AND user_id < %s",
                   (start_id, start_id + batch_size))

    for row in cursor.fetchall():
        db_idx = get_db_index(row['user_id'])
        table_idx = get_table_index(row['user_id'])

        target_cursor = targets[db_idx].cursor()
        target_cursor.execute(
            f"INSERT INTO orders_{table_idx} (...) VALUES (...) "
            f"ON DUPLICATE KEY UPDATE ...",
            (...)
        )
        targets[db_idx].commit()

def create_order_with_compensation(user_id, items):
    # 1. 创建订单（PROCESSING 状态）
    order_id = create_order(user_id, items, status='PROCESSING')

    try:
        # 2. 异步扣库存
        deduct_inventory(items)
        # 3. 成功，更新状态
        update_order_status(order_id, 'COMPLETED')
    except Exception:
        # 异常时订单保持 PROCESSING，等补偿任务处理
        pass

def compensation_task():
    """定时查找处理中超时的订单"""
    stale_orders = find_stale_orders(minutes=5)
    for order in stale_orders:
        compensate_order(order)

class MigrationRouter:
    def __init__(self):
        self.gray_ratio = 0.0
        self.gray_user_ids = set()

    def should_use_new_system(self, user_id):
        if user_id in self.gray_user_ids:
            return True
        user_hash = int(hashlib.md5(str(user_id).encode()).hexdigest()[:8], 16)
        return (user_hash % 100) < (self.gray_ratio * 100)

# 灰度推进
router.gray_ratio = 0.01  # 1%
router.gray_ratio = 0.1   # 10%
router.gray_ratio = 0.5   # 50%
router.gray_ratio = 1.0   # 100%

# 出问题立即回滚
router.gray_ratio = 0.0

# 雪花算法生成全局唯一 ID
def snowflake_id(worker_id, sequence):
    timestamp = int(time.time() * 1000) - EPOCH
    return (timestamp << 22) | (worker_id << 12) | sequence

-- 开启慢查询日志
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL long_query_time = 1;

-- 分析慢查询
EXPLAIN SELECT * FROM orders WHERE user_id = 123;

-- 关键字段
-- type: const > eq_ref > ref > range > index > ALL
-- key: 实际使用的索引
-- rows: 扫描行数
-- Extra: Using index（覆盖索引）、Using temporary（临时表）、Using filesort（文件排序）

-- 1. 只查需要的列（避免 SELECT *）
SELECT name, email FROM users WHERE id = 1;

-- 2. 用 LIMIT 1 优化已知唯一的查询
SELECT 1 FROM users WHERE email = '[email protected]' LIMIT 1;

-- 3. 批量插入
INSERT INTO users (name, email) VALUES
('a', '[email protected]'),
('b', '[email protected]'),
('c', '[email protected]');

-- 4. JOIN 用小表驱动大表
SELECT * FROM small_table s
INNER JOIN big_table b ON s.id = b.sid;

-- 5. 避免子查询，改用 JOIN
-- 慢
SELECT * FROM orders WHERE user_id IN (SELECT id FROM users WHERE vip = 1);
-- 快
SELECT o.* FROM orders o
INNER JOIN users u ON o.user_id = u.id
WHERE u.vip = 1;