Python+香港VPS：3招缩短数据库迁移停机时间

在香港VPS上做数据库迁移，最让运维头疼的就是停机时间——去年帮某电商客户迁订单库时，因迁移耗时过长，直接影响了大促期间的订单写入，损失了近20万销售额。其实通过Python脚本优化迁移流程，完全能把停机时间从几小时压缩到几十分钟。以下分享三个亲测有效的技巧。

第一步：用Python做迁移前「体检」，避免中途翻车

很多人迁移时急着「开干」，结果中途遇到版本不兼容、字段缺失等问题，反而拉长停机时间。去年处理的一个案例里，源库用的是PostgreSQL 12，目标库装成了11，迁移时索引语法不兼容，导致数据写入失败，光回滚就花了1个多小时。

用Python写个检查脚本，能提前规避这类问题。脚本核心是连接源库和目标库，对比版本号、表结构、索引信息。比如检查版本兼容性的代码：

import psycopg2

def check_compatibility():
    try:
        # 连接源库
        source_conn = psycopg2.connect(
            database="source_db", user="source_user",
            password="source_password", host="source_host", port="source_port"
        )
        source_cur = source_conn.cursor()
        source_cur.execute("SELECT version();")
        source_version = source_cur.fetchone()[0]

        # 连接目标库（香港VPS实例）
        target_conn = psycopg2.connect(
            database="target_db", user="target_user",
            password="target_password", host="香港VPS公网IP", port="5432"
        )
        target_cur = target_conn.cursor()
        target_cur.execute("SELECT version();")
        target_version = target_cur.fetchone()[0]

        if source_version.split()[1] != target_version.split()[1]:
            raise Exception(f"版本不匹配：源库{source_version}，目标库{target_version}")

        # 检查表结构是否一致（示例）
        source_cur.execute("SELECT column_name FROM information_schema.columns WHERE table_name='orders'")
        source_columns = {row[0] for row in source_cur.fetchall()}
        target_cur.execute("SELECT column_name FROM information_schema.columns WHERE table_name='orders'")
        target_columns = {row[0] for row in target_cur.fetchall()}
        if source_columns != target_columns:
            raise Exception(f"表结构差异：源库字段{source_columns- target_columns}，目标库字段{target_columns- source_columns}")

    except psycopg2.Error as e:
        print(f"检查失败：{str(e)}")
    finally:
        source_cur.close()
        source_conn.close()
        target_cur.close()
        target_conn.close()

check_compatibility()

这段脚本跑通后，相当于给迁移买了「保险」——去年帮教育平台迁用户信息库时，就是靠它提前发现目标库少了「注册来源」字段，临时修改表结构后再迁移，全程没耽误业务。

第二步：增量同步代替全量迁移，停机仅补最后10分钟数据

全量迁移大库（比如100GB以上）耗时久，中途业务还在产生新数据，容易导致迁移后数据不一致。这时候用Python做增量同步更聪明：先全量拷贝一次，之后实时监控源库变更（比如通过更新时间字段），把新数据同步到目标库。真正停机时，只需要同步最后几分钟的增量数据。

以下是简化的增量同步脚本逻辑：

import psycopg2
import time

def incremental_sync():
    source_conn = psycopg2.connect(...)  # 源库连接信息
    target_conn = psycopg2.connect(...)  # 香港VPS目标库连接信息
    source_cur = source_conn.cursor()
    target_cur = target_conn.cursor()

    last_sync_time = None
    while True:
        # 查询源库新增/修改的数据
        if last_sync_time:
            source_cur.execute("""
                SELECT * FROM orders 
                WHERE update_time > %s
                ORDER BY update_time ASC
            """, (last_sync_time,))
        else:
            source_cur.execute("SELECT * FROM orders ORDER BY update_time ASC")  # 首次全量

        rows = source_cur.fetchall()
        if rows:
            # 批量写入目标库（比逐条插入快3-5倍）
            insert_sql = "INSERT INTO orders VALUES (%s,%s,%s,%s) ON CONFLICT (order_id) DO UPDATE SET ..."
            target_cur.executemany(insert_sql, rows)
            target_conn.commit()
            last_sync_time = rows[-1][-1]  # 假设最后一列是update_time

        time.sleep(5)  # 每5秒检查一次，平衡延迟和资源消耗

    # 实际使用时需添加异常处理和退出逻辑

之前帮物流企业迁运单库时，源库每天新增20万条数据，用这个方法后，全量迁移花了2小时，之后每5秒同步一次增量，正式切换时只需要同步最后8分钟的3000条数据，停机时间从原本的3小时压缩到15分钟。

第三步：多线程并行迁移，3张表1小时干完3小时的活

如果数据库有多个独立表（比如用户表、订单表、商品表），用Python多线程并行迁移能大幅提升效率。之前迁一个包含5张表的系统库，串行迁移需要5小时，用多线程后2小时就完成了。

具体实现时，给每个表分配一个线程，同时从源库读取数据并写入目标库。需要注意：线程数别超过香港VPS的CPU核心数（比如4核VPS建议开4个线程），避免资源竞争拖慢速度。

示例代码：

import threading
import psycopg2

def migrate_table(table_name):
    try:
        # 每个线程独立连接数据库，避免共享连接导致的锁竞争
        source_conn = psycopg2.connect(...)
        target_conn = psycopg2.connect(...)
        source_cur = source_conn.cursor()
        target_cur = target_conn.cursor()

        # 读取源表数据
        source_cur.execute(f"SELECT * FROM {table_name}")
        rows = source_cur.fetchall()

        # 批量写入目标表
        insert_sql = f"INSERT INTO {table_name} VALUES ({','.join(['%s']*len(rows[0]))})"
        target_cur.executemany(insert_sql, rows)
        target_conn.commit()

    except Exception as e:
        print(f"表{table_name}迁移失败：{str(e)}")
    finally:
        source_cur.close()
        source_conn.close()
        target_cur.close()
        target_conn.close()

# 启动多线程迁移
tables_to_migrate = ["users", "orders", "products", "addresses", "coupons"]
threads = []
for table in tables_to_migrate:
    thread = threading.Thread(target=migrate_table, args=(table,))
    threads.append(thread)
    thread.start()

# 等待所有线程完成
for thread in threads:
    thread.join()

需要注意的是，如果表之间有外键依赖（比如订单表关联用户表），建议先迁主表（用户表），再迁从表（订单表），避免外键约束导致写入失败。

总结下来，用Python优化香港VPS数据库迁移的核心就三点：迁移前用脚本做兼容性检查避坑，迁移中用增量同步减少停机补数据量，迁移时用多线程并行提升速度。实际操作时，根据数据库大小（小库用并行，大库用增量）和业务中断容忍度（高敏感业务优先增量同步）灵活组合这些方法，基本能把停机时间控制在30分钟以内。