数据库迁移实践笔记

数据库迁移 零停机迁移 数据一致性 运维实践
作者: 醉月思 发布日期: 2021-10-19 分类: 技术实践

去年单机 MySQL 磁盘用到 85%，备份窗口也越拉越长。业务不能停，只能边跑边迁：先上主从，再分库分表。双写、对账、切流量这几步，哪一步算错了都得能回滚。

为什么需要迁移

性能问题

单机数据库扛不住了，需要升级架构。

扩展性问题

数据量太大，需要分库分表。

技术栈问题

从 MySQL 迁移到 PostgreSQL，或者从关系型数据库到 NoSQL。

停机迁移

简单场景

对于小数据量、容忍短时间停机的场景。

# 1. 停止应用
systemctl stop myapp

# 2. 导出数据
mysqldump -u root -p mydb > backup.sql

# 3. 导入到新库
mysql -u root -p newdb < backup.sql

# 4. 修改应用配置，连接新库
# 5. 启动应用
systemctl start myapp

# 6. 验证数据

优势：简单，不会出现数据不一致 劣势：需要停机，业务受影响

零停机迁移

双写方案

应用同时写新旧两个数据库。

class DualWriteDatabase:
    def __init__(self, old_db, new_db):
        self.old_db = old_db
        self.new_db = new_db
    
    def insert(self, table, data):
        # 并发写入两个库
        with ThreadPoolExecutor(max_workers=2) as executor:
            executor.submit(self.old_db.insert, table, data)
            executor.submit(self.new_db.insert, table, data)
    
    def update(self, table, id, data):
        with ThreadPoolExecutor(max_workers=2) as executor:
            executor.submit(self.old_db.update, table, id, data)
            executor.submit(self.new_db.update, table, id, data)
    
    def delete(self, table, id):
        with ThreadPoolExecutor(max_workers=2) as executor:
            executor.submit(self.old_db.delete, table, id)
            executor.submit(self.new_db.delete, table, id)
    
    def get(self, table, id):
        # 读优先新库
        data = self.new_db.get(table, id)
        if data:
            return data
        return self.old_db.get(table, id)

流程：

  1. 部署双写版本
  2. 验证新旧库数据一致
  3. 切换读流量到新库
  4. 验证业务正常
  5. 下线旧库

数据同步方案

使用工具同步数据。

使用 gh-ost：

# 安装 gh-ost
go install github.com/github/gh-ost@latest

# 在线迁移表
gh-ost \
  --max-load=Threads_running=25 \
  --critical-load=Threads_running=1000 \
  --chunk-size=1000 \
  --throttle-control-replicas="..." \
  --database=mydb \
  --table=users \
  --alter="ENGINE=InnoDB,ADD COLUMN age INT" \
  --allow-on-master \
  --execute

使用 pt-online-schema-change：

# 安装 Percona Toolkit
apt-get install percona-toolkit

# 在线迁移表
pt-online-schema-change \
  --alter="ENGINE=InnoDB,ADD COLUMN age INT" \
  --critical-load=Threads_running=100 \
  --max-load=Threads_running=25 \
  --execute \
  D=mydb,t=users

分库分表迁移

垂直分库

按业务拆分数据库。

单库: mydb
    ├── users
    ├── orders
    ├── payments
    └── logs

垂直分库:
    ├── user_db (users)
    ├── order_db (orders, payments)
    └── log_db (logs)

# 路由逻辑
class DatabaseRouter:
    def __init__(self):
        self.user_db = connect('user_db')
        self.order_db = connect('order_db')
        self.log_db = connect('log_db')
    
    def get_connection(self, table):
        if table.startswith('user'):
            return self.user_db
        elif table.startswith('order') or table.startswith('payment'):
            return self.order_db
        elif table.startswith('log'):
            return self.log_db
        raise ValueError(f"Unknown table: {table}")

水平分表

按数据量拆分表。

单表: users (1000 万行)

水平分表:
    ├── users_0
    ├── users_1
    ├── users_2
    └── users_3

class ShardingRouter:
    def __init__(self, num_shards=4):
        self.shards = [connect(f'users_{i}') for i in range(num_shards)]
    
    def get_shard(self, user_id):
        # 根据 user_id 分片
        return user_id % len(self.shards)
    
    def get_user(self, user_id):
        shard = self.get_shard(user_id)
        return self.shards[shard].get_user(user_id)
    
    def insert_user(self, user):
        user_id = user['id']
        shard = self.get_shard(user_id)
        self.shards[shard].insert_user(user)

数据一致性

数据校验

迁移完成后，需要校验数据一致性。

def compare_tables(old_table, new_table):
    old_data = old_table.all()
    new_data = new_table.all()
    
    if len(old_data) != len(new_data):
        print(f"Row count mismatch: {len(old_data)} vs {len(new_data)}")
        return False
    
    for old_row, new_row in zip(old_data, new_data):
        if old_row != new_row:
            print(f"Data mismatch: {old_row} vs {new_row}")
            return False
    
    return True

# 定期校验
schedule.every().day.at("02:00").do(lambda: compare_tables(old_table, new_table))

数据修复

发现不一致后，需要修复数据。

def repair_data(old_table, new_table):
    old_data = old_table.all()
    
    for old_row in old_data:
        new_row = new_table.get(old_row['id'])
        if not new_row or new_row != old_row:
            # 插入或更新
            new_table.upsert(old_row)
            print(f"Repaired: {old_row['id']}")

踩过的坑

坑一：数据结构不兼容

新旧数据库的字段类型不一样。

解决：迁移前做好字段映射，统一数据类型。

def transform_data(old_data):
    return {
        'id': old_data['id'],
        'name': old_data['name'],
        'age': int(old_data['age']),  # 类型转换
        'created_at': parse_date(old_data['created_at']),  # 格式转换
    }

坑二：外键约束

迁移时外键约束导致失败。

解决：暂时关闭外键约束，迁移后再打开。

-- 关闭外键约束
SET FOREIGN_KEY_CHECKS = 0;

-- 执行迁移
...

-- 打开外键约束
SET FOREIGN_KEY_CHECKS = 1;

坑三：索引不同步

新库的索引没建好，查询很慢。

解决：迁移前规划好索引，迁移后验证。

-- 创建索引
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_created_at ON users(created_at);

-- 验证索引
EXPLAIN SELECT * FROM users WHERE email = '[email protected]';

坑四：双写失败

双写时，一个库成功了，另一个失败了。

解决：记录失败的操作，定期重试。

class DualWriteDatabase:
    def __init__(self, old_db, new_db):
        self.old_db = old_db
        self.new_db = new_db
        self.failed_writes = []
    
    def insert(self, table, data):
        old_success = self.safe_insert(self.old_db, table, data)
        new_success = self.safe_insert(self.new_db, table, data)
        
        if not (old_success and new_success):
            # 记录失败的操作
            self.failed_writes.append({
                'table': table,
                'data': data,
                'timestamp': datetime.now()
            })
    
    def retry_failed(self):
        for operation in self.failed_writes:
            try:
                self.insert(operation['table'], operation['data'])
                self.failed_writes.remove(operation)
            except Exception as e:
                print(f"Retry failed: {e}")

迁移检查清单

迁移前

  • 评估数据量和迁移时间
  • 备份原数据库
  • 设计迁移方案
  • 准备回滚方案
  • 通知相关方

迁移中

  • 监控数据库性能
  • 监控应用错误率
  • 验证数据一致性
  • 准备随时回滚

迁移后

  • 验证业务功能
  • 监控性能指标
  • 清理临时数据
  • 更新文档
  • 总结经验

写在最后

数据库迁移这东西，不是技术问题，是工程问题。

准备好了：

  • 备份方案
  • 回滚方案
  • 监控方案
  • 验证方案

失败了，可能的原因：

  • 测试不充分
  • 监控不到位
  • 回滚不及时
  • 沟通不充分

迁移之前先问自己几个问题：

  • 数据量多大？
  • 业务能容忍多长时间停机？
  • 有没有回滚方案？
  • 团队有没有经验？

不是所有迁移都需要零停机，有时候简单粗暴的停机迁移反而更可靠。

这次数据库迁移花了一个月，从单机到主从，再到分库分表。迁移完成后，数据库 QPS 从 2000 提升到 10000，响应时间从 500ms 降到 50ms。虽然中间遇到过数据不一致、索引丢失等问题，但最终都解决了。

版权声明: 本文首发于 指尖魔法屋-数据库迁移实践笔记（https://blog.thinkmoon.cn/post/64-database-migration-zero-downtime-consistency/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

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