数据库迁移：单机不够用了之后

数据库迁移 分库分表 MySQL ShardingSphere 数据一致性
作者: 醉月思 发布日期: 2021-06-15 分类: 技术实践

数据库迁移：单机不够用了之后一旦进项目，好看的架构图就没那么管用了。

项目一开始很简单，一个 MySQL 单实例，几百万数据，响应时间也就几十毫秒。

问题是怎么来的

项目一开始很简单，一个 MySQL 单实例，几百万数据，响应时间也就几十毫秒。业务一路跑得顺畅，谁也没往那边想。直到某天线上监控开始告警，数据库 CPU 长期在 80% 以上波动，主从延迟从几百毫秒变成了几秒，有时候甚至分钟级。

这时候团队才意识到：原来的方案已经撑不住了。单机再怎么调优，硬件再怎么升级，总有个上限。物理存储也有极限，一张订单表跑到 8000 万行，全量扫一遍要十几秒，索引文件占了 200G。

当时我接手这个任务，第一反应是先摸底：监控、慢查询、表结构、业务逻辑全部过一遍。得先搞清楚哪块先扛不住，再谈迁不迁。

先搭主从，再做打算

在真正动手分库分表之前，先把主从搭起来。读流量暂时还走主库，这一步主要是给后续迁移留操作窗口。

我们用的是 MySQL 8.0，主从配置不算复杂，但有几个细节坑不少：

# 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

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

server-id 必须全局唯一，这个是基础。gtid_mode 开启之后，故障切换会好办很多，不用再计算 binlog 位置。sync_binlog 和 innodb_flush_log_at_trx_commit 都是 1 的时候，性能会差一截，但数据丢失的风险降到最低——在迁移阶段，宁可慢一点，也不能让同步链路断掉。

主从启动后，先跑一段时间双写验证，读流量暂时不切。我们在代码里埋了个钩子，每次写主库时异步把同样请求发一份到从库，再对比结果。

这个过程持续了两周，发现了几处业务逻辑上的不一致：比如某些触发器在主从环境下的行为不一样，有个存储过程依赖了非确定性的函数。这些问题不在迁移阶段暴露，切线上的时候就要出大事故。

分库分表的那些坑

主从只是第一步，真正的硬骨头是分库分表。

当时订单表 8000 万行，每天新增 10 万左右。按照用户 ID 哈希分片，切成 4 个库、每个库 8 张表，一共 32 张表。路由规则是：db_index = user_id % 4，table_index = user_id / 4 % 8。

这个策略看着简单，但实施的时候遇到不少实际问题。

第一个问题是历史数据迁移。不能一次性把 8000 万行全搬过去，否则业务停机时间太长。我们采用的是"双写 + 数据校验 + 逐步切换"的方案：

import pymysql
import hashlib
import time

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_conn, target_conns):
    """
    批量迁移数据，按用户ID范围分批
    """
    cursor = source_conn.cursor()
    
    query = """
    SELECT * FROM orders
    WHERE user_id >= %s AND user_id < %s
    ORDER BY user_id
    """
    cursor.execute(query, (start_id, start_id + batch_size))
    
    rows = cursor.fetchall()
    
    for row in rows:
        user_id = row['user_id']
        db_idx = get_db_index(user_id)
        table_idx = get_table_index(user_id)
        
        target_cursor = target_conns[db_idx].cursor()
        
        # 插入目标表
        insert_query = """
        INSERT INTO orders_%s (user_id, order_no, amount, created_at)
        VALUES (%s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            amount = VALUES(amount),
            created_at = VALUES(created_at)
        """ % table_idx
        
        target_cursor.execute(insert_query, (
            user_id,
            row['order_no'],
            row['amount'],
            row['created_at']
        ))
        
        target_conns[db_idx].commit()
        target_cursor.close()
    
    cursor.close()
    return len(rows)

def verify_consistency(source_conn, target_conns, user_id):
    """
    校验数据一致性
    """
    source_cursor = source_conn.cursor()
    source_cursor.execute(
        "SELECT * FROM orders WHERE user_id = %s", (user_id,)
    )
    source_row = source_cursor.fetchone()
    
    db_idx = get_db_index(user_id)
    table_idx = get_table_index(user_id)
    
    target_cursor = target_conns[db_idx].cursor()
    target_cursor.execute(
        "SELECT * FROM orders_%s WHERE user_id = %s" % (table_idx, user_id),
        (user_id,)
    )
    target_row = target_cursor.fetchone()
    
    # 比较关键字段
    fields_to_check = ['order_no', 'amount', 'created_at']
    for field in fields_to_check:
        if source_row[field] != target_row[field]:
            print(f"Inconsistency detected for user_id {user_id}, field {field}")
            print(f"Source: {source_row[field]}, Target: {target_row[field]}")
            return False
    
    source_cursor.close()
    target_cursor.close()
    return True

这个脚本跑了整整 5 天，每批 5000 条，中间因为主键冲突、字符集不匹配、时间戳精度问题卡了几次。最麻烦的是有两个外键约束在目标环境里建不起来，最后只能先去掉约束，迁移完之后再补回来。

第二个问题是路由改造成本。原有代码里，大部分 SQL 都没有带上 user_id 作为条件，导致没法直接路由。我们做了一个轻量级的 SQL 解析器，在执行之前先检测语句能不能推导出分片键。如果不能，就通过一个中间表查一下用户对应的分片信息。

这个方案虽然能跑，但性能损耗不小，平均每次查询多了 5-10ms。后来我们把用户信息做了本地缓存，才勉强降下来。

数据一致性怎么保

分布式环境下，最头疼的就是跨分片事务。

一个订单可能关联多个商品，而这些商品又属于不同的分片。如果订单在分片 A，商品在分片 B，那么下单的时候要么做两阶段提交，要么改业务逻辑接受最终一致性。

我们当时没有上分布式事务框架（比如 Seata），而是选择了业务层面的补偿机制：

  1. 先在订单所在分片创建订单记录，状态设为"处理中"
  2. 异步扣减库存，如果失败就更新订单状态为"失败"
  3. 如果成功，更新订单状态为"完成"
  4. 有个定时任务定期检查"处理中"超过一定时间的订单，做补偿处理
def create_order_with_compensation(user_id, items):
    """
    创建订单，带补偿机制
    """
    db_idx = get_db_index(user_id)
    table_idx = get_table_index(user_id)
    
    order_no = generate_order_no()
    total_amount = sum(item['price'] * item['quantity'] for item in items)
    
    # 第一步：创建订单记录
    cursor = db_conns[db_idx].cursor()
    try:
        cursor.execute("""
            INSERT INTO orders_%s
            (user_id, order_no, total_amount, status, created_at)
            VALUES (%s, %s, %s, 'PROCESSING', NOW())
        """ % table_idx, (user_id, order_no, total_amount))
        
        order_id = cursor.lastrowid
        db_conns[db_idx].commit()
    except Exception as e:
        db_conns[db_idx].rollback()
        print(f"创建订单失败: {e}")
        return None
    
    # 第二步：异步扣减库存
    try:
        inventory_result = deduct_inventory(items)
        if not inventory_result['success']:
            # 扣减失败，更新订单状态
            cursor.execute("""
                UPDATE orders_%s
                SET status = 'FAILED', fail_reason = %s
                WHERE id = %s
            """ % table_idx, (inventory_result['message'], order_id))
            db_conns[db_idx].commit()
            return None
        
        # 第三步：更新订单状态为完成
        cursor.execute("""
            UPDATE orders_%s
            SET status = 'COMPLETED'
            WHERE id = %s
        """ % table_idx, (order_id,))
        db_conns[db_idx].commit()
        
        return order_id
        
    except Exception as e:
        # 异常情况，记录下来等待补偿
        print(f"扣减库存异常: {e}")
        # 订单保持 PROCESSING 状态，由补偿任务处理
        return order_id

def compensation_task():
    """
    定时补偿任务
    """
    for db_idx in range(4):
        cursor = db_conns[db_idx].cursor()
        
        # 查找处理中超过 5 分钟的订单
        cursor.execute("""
            SELECT id, order_no, user_id
            FROM orders
            WHERE status = 'PROCESSING'
            AND created_at < DATE_SUB(NOW(), INTERVAL 5 MINUTE)
        """)
        
        stale_orders = cursor.fetchall()
        
        for order in stale_orders:
            # 尝试重新处理或取消订单
            order_id = order['id']
            try:
                result = compensate_order(order_id)
                if result['success']:
                    cursor.execute("""
                        UPDATE orders
                        SET status = %s, fail_reason = %s
                        WHERE id = %s
                    """ % (result['status'], result.get('message', ''), order_id))
                else:
                    cursor.execute("""
                        UPDATE orders
                        SET status = 'FAILED', fail_reason = %s
                        WHERE id = %s
                    """ % (result['message'], order_id))
                
                db_conns[db_idx].commit()
            except Exception as e:
                print(f"补偿订单 {order_id} 失败: {e}")
                db_conns[db_idx].rollback()
        
        cursor.close()

这个方案谈不上优雅，但在有限的时间和人力约束下，是一个可接受的折中。后续订单量继续增长的时候，我们才慢慢把核心业务迁移到了基于 Saga 模式的框架上。

回滚策略怎么想

迁移最怕的，往往是迁完出问题却回不去。

我们在做迁移计划的时候，专门留了"快速回滚"的预案。核心思路是：新旧系统并行运行，通过灰度开关控制流量，任何时候都能一键切回旧系统。

class MigrationRouter:
    def __init__(self):
        self.new_system_enabled = False
        self.gray_user_ids = set()
        self.gray_ratio = 0.0
    
    def should_use_new_system(self, user_id):
        """
        判断是否应该使用新系统
        """
        if not self.new_system_enabled:
            return False
        
        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)
    
    def enable_new_system(self, gray_ratio=0.0):
        """
        启用新系统，并设置灰度比例
        """
        self.new_system_enabled = True
        self.gray_ratio = gray_ratio
    
    def disable_new_system(self):
        """
        禁用新系统（回滚）
        """
        self.new_system_enabled = False
        self.gray_ratio = 0.0
    
    def add_gray_user(self, user_id):
        """
        添加灰度用户
        """
        self.gray_user_ids.add(user_id)

# 使用示例
router = MigrationRouter()

# 先灰度 1% 的流量
router.enable_new_system(gray_ratio=0.01)

# 监控一段时间后，逐步放开
router.enable_new_system(gray_ratio=0.1)
router.enable_new_system(gray_ratio=0.5)

# 如果出现问题，立即回滚
router.disable_new_system()

灰度期间，我们主要监控这几个指标：P99 响应时间、错误率、主从延迟、数据不一致次数。任何一个指标超过阈值，就暂停扩灰，先定位问题。

有一次我们在扩到 30% 的时候，发现某个分片的 CPU 突然飙升。查了一圈，原来是一个新上线的报表功能没有适配分片逻辑，把全量查询都打到了同一个分片上。这个不应该是迁移的问题，但它确实暴露了旧系统就存在的隐患。我们连夜修了这个报表，才继续推进。

事后复盘

从单机迁到分布式，动的是整套架构，不是换个数据库版本那么简单。几条我后来才想明白的：

别指望一次迁完。 分阶段、小步快跑，每一步都留回滚预案，风险才可控。

一致性没有银弹。 要么接受最终一致，要么上分布式事务框架——后者性能和复杂度都会上去。我们初期用业务补偿，量上来后再慢慢升级。

监控得提前就位。 分片不均、热点、主从延迟这些问题，靠人眼看不出，得靠指标。没监控就是盲人摸象。

团队要时间适应。 新架构意味着新的排查方式和开发规范，一天学不会。

数据库迁移很多时候是必答题，没有完美方案，只有适应当前业务和人力约束的妥协。路走完之后回头看，那些坑和弯路也算系统演进的一部分。

当然，下次有人问你「要不要分库分表」，先反问一句：确定当前的问题真的是分库分表能解决的吗？有时候，加个索引、优化一条 SQL、调整一下硬件配置，效果可能更直接。

可用性说明：本文发布于 2021 年 6 月，距今已超过五年。文中涉及的软件版本、接口、下载地址、命令参数和操作界面可能已经发生变化，部分方案在当前环境下可能失效。请结合官方最新文档核对后再操作，生产环境使用前务必先行验证。

版权声明: 本文首发于 指尖魔法屋-数据库迁移：单机不够用了之后（https://blog.thinkmoon.cn/post/133-migration-single-to-distributed-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章