数据库分库分表策略详解及数据迁移方案实战-源码库

数据库分库分表策略详解及数据迁移方案实战：从单表千万到分布式架构的平滑演进

大家好，作为一名经历过多次数据库架构升级的老兵，今天想和大家深入聊聊分库分表这个经典话题。记得第一次面对单表数据量突破千万时，查询性能明显下降，那种焦虑感至今记忆犹新。经过多个项目的实战积累，我总结出了一套相对成熟的分库分表策略和数据迁移方案，希望能帮助正在面临同样挑战的你。

为什么需要分库分表？

当单表数据量达到千万级别，或者并发连接数超过数据库承载能力时，我们就需要考虑分库分表了。我经历过的一个电商项目，用户表在三年内积累了近两千万数据，即使加了索引，复杂查询的响应时间也经常超过3秒。更糟糕的是，数据库服务器的CPU使用率长期在80%以上，严重影响了业务稳定性。

分库分表的核心目标很明确：

提升查询性能，降低单表数据量
提高系统可用性，避免单点故障
支持水平扩展，应对业务增长

分库分表策略选择

在具体实施前，我们需要根据业务特点选择合适的策略。根据我的经验，主要有以下几种常用方案：

1. 水平分表

这是最常用的方式，按照某种规则将数据分布到多个结构相同的表中。比如按用户ID取模：

-- 假设分4张表
table_0 (user_id % 4 = 0)
table_1 (user_id % 4 = 1)  
table_2 (user_id % 4 = 2)
table_3 (user_id % 4 = 3)

这种方案的优点是实现简单，数据分布均匀。但有个坑需要注意：如果后续需要扩容，重新分片的成本很高。

2. 按时间分表

对于日志、订单等有时效性的数据，按时间分表是个不错的选择：

orders_2023_01
orders_2023_02
orders_2023_03

我曾经在一个金融项目中采用这种方案，配合数据归档策略，既保证了近期数据的查询性能，又控制了总数据量。

3. 垂直分库

将不同业务模块的数据拆分到不同的数据库中：

user_db    -- 用户相关表
order_db   -- 订单相关表  
product_db -- 商品相关表

这种方案能有效降低单库压力，但跨库联查会变得复杂，需要在应用层处理。

数据迁移实战步骤

确定了分片策略后，最关键的就是数据迁移了。这里我分享一个经过验证的平滑迁移方案：

第一步：双写准备

在开始迁移前，先改造代码支持双写。这是保证数据一致性的关键：

// 示例代码 - 数据写入层
public class UserService {
    public void createUser(User user) {
        // 写入原表
        userMapper.insertToOldTable(user);
        
        // 同时写入新分片表
        String tableSuffix = getTableSuffix(user.getId());
        userMapper.insertToShardTable(user, tableSuffix);
    }
}

注意要先开启双写，再开始数据迁移，这样可以避免迁移期间的数据丢失。

第二步：数据迁移

使用批处理方式迁移历史数据，建议在业务低峰期进行：

#!/bin/bash
# 数据迁移脚本示例
for i in {0..99}; do
    # 分批迁移，每批10000条
    mysql -h old_db -e "SELECT * FROM users WHERE id % 100 = $i LIMIT 10000" | 
    mysql -h new_db -e "LOAD DATA LOCAL INFILE '/dev/stdin' INTO TABLE users_$((i % 4))"
    
    echo "迁移批次 $i 完成"
    sleep 1  # 控制迁移速度，避免对线上造成影响
done

这里有个经验：迁移过程中要实时监控数据库性能，如果发现明显影响，立即调整迁移速度。

第三步：数据校验

迁移完成后，必须进行数据校验。我通常会用脚本来对比关键指标：

import pymysql

# 对比原表和新表的数据量
def check_data_count():
    old_conn = pymysql.connect(host='old_db')
    new_conn = pymysql.connect(host='new_db')
    
    old_count = old_conn.cursor().execute("SELECT COUNT(*) FROM users")
    new_count = 0
    for i in range(4):
        count = new_conn.cursor().execute(f"SELECT COUNT(*) FROM users_{i}")
        new_count += count
    
    assert old_count == new_count, "数据量不一致！"
    print("数据校验通过")