笔者通过一个 Redis 数据库迁移的例子，介绍了迁移脚本的执行思路。

作者：马文斌，MySQL/Redis 爱好者~

爱可生开源社区出品，原创内容未经授权不得随意使用，转载请联系小编并注明来源。

本文约 500 字，预计阅读需要 2 分钟。

业务背景

最近因业务需要将集群中的 Redis db8 迁移到另一个库中。不需要把全部的 Redis 都迁移过去，只需要迁移 db8 到新服务器的 db15。

大概情况是这样，如图：

当然也探索了一些开源工具看能否实现，最后效果都不太理想，于是自己手撸一个脚本来迁移吧。

migrate_redis.py

以下是脚本内容。

import redis import time # 定义 redis1 和 redis 主机信息 redis1_host = '192.168.1.1' redis1_port = 6579 redis1_db = 8 redis1_password = 'xxxx' redis2_host = '192.168.1.1' redis2_port = 6579 redis2_db = 15 redis2_password = 'xxxx' # 连接 redis1 和 redis2 并验证密码 redis1 = redis.StrictRedis(host=redis1_host, port=redis1_port, db=redis1_db, password=redis1_password) redis2 = redis.StrictRedis(host=redis2_host, port=redis2_port, db=redis2_db, password=redis2_password) # 设置每次批量迁移的数据量 batch_size = 1000 # 为进度跟踪初始化变量 keys_processed = 0 start_time = time.time() # 使用 SCAN 批量获取 key cursor = '0' total_keys = len(redis1.keys('*')) while cursor != 0: cursor, keys = redis1.scan(cursor, count=batch_size) for key in keys: key_data = redis1.dump(key) redis2.restore(key, 0, key_data, replace=True) keys_processed += 1 # 每 1000 个 key 打印一次进度 if keys_processed % batch_size == 0 or keys_processed == total_keys: elapsed_time = time.time() - start_time keys_per_second = batch_size / elapsed_time estimated_remaining_time = (total_keys - keys_processed) / keys_per_second print(f"Processed {keys_processed}/{total_keys} keys. " f"Elapsed Time: {elapsed_time:.2f} seconds. " f"Estimated Remaining Time: {estimated_remaining_time:.2f} seconds for the next 1000 keys.") # 为下一批次重置变量 start_time = time.time() print("Data migration completed.") 

输出效果

每隔 1000 个 key 打印一次输出，并评估剩余迁移时间。

Processed 1000/3592 keys. Elapsed Time: 16.46 seconds. Estimated Remaining Time: 42.67 seconds for the next 1000 keys. Processed 2000/3592 keys. Elapsed Time: 16.96 seconds. Estimated Remaining Time: 27.01 seconds for the next 1000 keys. Processed 3000/3592 keys. Elapsed Time: 17.03 seconds. Estimated Remaining Time: 10.08 seconds for the next 1000 keys. Processed 3592/3592 keys. Elapsed Time: 9.81 seconds. Estimated Remaining Time: 0.00 seconds for the next 1000 keys. Data migration completed. Process finished with exit code 0 

迁移完之后检查

运行 info 命令检查：

源库 3592 个 key，目标库也是 3592 个 key, 迁移完成，收工！

更多技术文章，请访问：https://opensource.actionsky.com/

关于 SQLE

SQLE 是一款全方位的 SQL 质量管理平台，覆盖开发至生产环境的 SQL 审核和管理。支持主流的开源、商业、国产数据库，为开发和运维提供流程自动化能力，提升上线效率，提高数据质量。

SQLE 获取

类型	地址
版本库	https://github.com/actiontech/sqle
文档	https://actiontech.github.io/sqle-docs/
发布信息	https://github.com/actiontech/sqle/releases
数据审核插件开发文档	https://actiontech.github.io/sqle-docs/docs/dev-manual/plugins/howtouse