Redis数据迁移实战：从自建到云托管（阿里云／腾讯云）的平滑过渡

第一章：迁移规划与策略选型

1.1 迁移核心挑战分析

挑战	描述	影响
数据一致性	迁移过程中源库仍在写入，需保证数据不丢失。	业务正确性
服务可用性	尽可能减少停机时间，实现平滑切换。	用户体验、收入
性能影响	迁移过程对源库和网络的压力。	源库稳定性
兼容性	自建Redis版本与云服务版本的特性差异。	功能可用性

1.2 迁移方案对比选型

以下是三种主流迁移方案的深度对比，其决策流程可概括为：

方案	原理	适用场景	预估停机时间	复杂度
1. 停机文件迁移	关闭源库 → 备份RDB → 上传到云 → 启动云库	数据量极大，可接受较长停机时间（小时级）	高（数小时）	低
2. 直连同步迁移	使用工具（如redis-shake）直连源库和云库，全量+增量同步	数据量适中（< 50GB），允许分钟级停机	低（分钟级）	中
3. 持续增量迁移	先做一次全量同步，然后持续同步增量数据，最后切换	数据量大，要求秒级停机，业务连续性要求极高	极低（秒级）	高

推荐结论：

• 大多数场景：选择方案二（直连同步），平衡了复杂度和停机时间。
• 海量数据场景：选择方案三（持续增量），但需要投入更多运维资源。
• 测试/非核心业务：可选择方案一（停机文件），操作简单。

---

第二章：迁移前准备

2.1 环境检查与兼容性验证

2.1.1 版本与特性检查

# 1. 检查源Redis信息
redis-cli -h <source-host> -p <source-port> info server
# 重点关注：redis_version, arch_bits, multiplexing_api, gcc_version# 2. 检查数据量
redis-cli -h <source-host> -p <source-port> info keyspace
# 示例输出：db0:keys=1000000,expires=200000,avg_ttl=3600000# 3. 检查云Redis兼容性（以阿里云为例）
# 登录阿里云控制台，确认目标实例的Redis版本 >= 源实例版本
# 检查是否禁用了一些命令（如KEYS, FLUSHALL）

2.1.2 网络连通性测试

# connectivity_test.py
import redis
import socketdef test_connectivity(host, port):try:# 测试TCP连接sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.settimeout(5)result = sock.connect_ex((host, port))sock.close()if result == 0:print(f"TCP连接成功: {host}:{port}")else:print(f"TCP连接失败: {host}:{port}")# 测试Redis连接r = redis.Redis(host=host, port=port, socket_timeout=5)r.ping()print(f"Redis连接成功: {host}:{port}")return Trueexcept Exception as e:print(f"连接测试失败 {host}:{port} - {e}")return False# 测试源库和云库的连接性
source_host = "192.168.1.100"
source_port = 6379
target_host = "r-xxx.redis.rds.aliyuncs.com"  # 云Redis连接地址
target_port = 6379test_connectivity(source_host, source_port)
test_connectivity(target_host, target_port)

2.2 云环境准备

2.2.1 创建云Redis实例

# 阿里云CLI创建Redis实例示例
aliyun rds CreateDBInstance \--EngineVersion "5.0" \--DBInstanceClass "redis.master.small.default" \--DBInstanceStorage 20 \--InstanceName "production-redis-migration" \--SecurityIPList "192.168.1.100/32" \  # 源服务器IP--VpcId "vpc-xxx" \--VSwitchId "vsw-xxx" \--ZoneId "cn-hangzhou-h"

2.2.2 网络配置

# 配置云Redis白名单，允许源服务器访问
# 阿里云控制台操作路径：
# 1. 登录Redis控制台
# 2. 选择目标实例
# 3. 数据安全性 -> 白名单设置 -> 添加源服务器IP# 如果需要通过VPN/专线连接
# 配置VPC对等连接或VPN网关

2.3 迁移工具准备

2.3.1 安装Redis-Shake

# 下载redis-shake（选择最新版本）
wget https://github.com/alibaba/RedisShake/releases/download/v3.0.0/redis-shake-linux-amd64.tar.gz
tar -zxvf redis-shake-linux-amd64.tar.gz
cd redis-shake# 或者从源码编译
git clone https://github.com/alibaba/RedisShake.git
cd RedisShake
make

2.3.2 配置迁移任务

// redis-shake.conf
{"source": {"type": "standalone","address": "192.168.1.100:6379","password": "source_password","tls": false},"target": {"type": "standalone", "address": "r-xxx.redis.rds.aliyuncs.com:6379","password": "cloud_password","tls": true},"parallel": 32,                   // 并发数"scan.key_number": 1000,          // 每次扫描key数量"scan.special.cloud": true,        // 云Redis优化"scan.rate_limit": 20,             // 限流20MB/s"log": {"level": "info","filename": "redis-shake.log"},"metric": {"enable": true,"prometheus": true,"port": 9999}
}

---

第三章：迁移实战操作

3.1 方案二实施：直连同步迁移

3.1.1 全量数据迁移

# 1. 启动redis-shake进行全量迁移
./redis-shake -type sync -conf redis-shake.conf# 2. 监控迁移进度
# 通过prometheus指标监控：http://localhost:9999/metrics
# 查看日志：tail -f redis-shake.log# 3. 验证数据一致性
# 在源库和目标库分别执行以下命令对比
redis-cli -h source-host info keyspace
redis-cli -h target-host info keyspace# 4. 抽样验证特定key
redis-cli -h source-host get "sample:key:1"
redis-cli -h target-host get "sample:key:1"

3.1.2 增量数据同步与切换

# 1. 配置应用双写（在迁移期间）
# 在应用代码中同时写入源库和目标库
public class DualWriteRedis {private RedisTemplate primaryRedis; // 源库private RedisTemplate secondaryRedis; // 目标库public void set(String key, String value) {// 先写主库primaryRedis.opsForValue().set(key, value);// 异步写备库CompletableFuture.runAsync(() -> {try {secondaryRedis.opsForValue().set(key, value);} catch (Exception e) {log.error("双写失败", e);}});}
}# 2. 停止增量同步并切换
# 观察redis-shake日志，确认增量同步延迟为0
# 停止应用向源库的写入# 3. 最后一批增量数据同步
# 等待redis-shake完成最后的数据同步# 4. 切换应用配置
# 修改应用配置，将Redis连接指向云实例

3.2 方案三实施：持续增量迁移

3.2.1 使用Redis原生复制

# 1. 在源Redis生成RDB文件
redis-cli -h source-host save  # 或者 bgsave# 2. 将RDB文件上传到云Redis
# 云Redis通常提供数据导入功能
# 阿里云控制台：实例管理 -> 备份与恢复 -> 数据导入# 3. 配置主从复制
# 在源Redis配置中允许云Redis连接
# source Redis config: replicaof no one → 保持为主库# 4. 在云Redis上配置为源Redis的从库
redis-cli -h cloud-host -a cloud_password replicaof source-host 6379# 5. 监控复制状态
redis-cli -h cloud-host info replication
# 查看：master_link_status:up, master_sync_in_progress:0

3.2.2 切换流程

# switchover.py - 自动化切换脚本
import redis
import timedef perform_switchover(source_host, source_port, source_passwd,target_host, target_port, target_passwd):# 连接源库和目标库source_r = redis.Redis(host=source_host, port=source_port, password=source_passwd)target_r = redis.Redis(host=target_host, port=target_port, password=target_passwd)try:# 1. 停止源库写入source_r.config_set('readonly', 'yes')print("源库已设置为只读")# 2. 等待最后的数据同步time.sleep(30)  # 等待30秒确保最后数据同步# 3. 检查复制延迟replication_info = target_r.info('replication')if replication_info['master_sync_in_progress'] == 0 and \replication_info['master_link_status'] == 'up':print("复制状态正常，可以切换")else:raise Exception("复制状态异常，不能切换")# 4. 提升云库为主库target_r.replicaof('no', 'one')print("云库已提升为主库")# 5. 应用切换逻辑（更新DNS、配置中心等）update_dns_config(target_host)print("DNS配置已更新")return Trueexcept Exception as e:print(f"切换失败: {e}")# 回滚：恢复源库写入source_r.config_set('readonly', 'no')print("已回滚：源库恢复读写")return False# 执行切换
perform_switchover('192.168.1.100', 6379, 'source_pass','r-xxx.redis.rds.aliyuncs.com', 6379, 'cloud_pass')

---

第四章：迁移后验证与优化

4.1 数据一致性验证

4.1.1 全量校验工具

# redis_verify.py
import redis
import jsonclass RedisDataVerifier:def __init__(self, source_config, target_config):self.source_client = redis.Redis(**source_config)self.target_client = redis.Redis(**target_config)def verify_key(self, key):"""验证单个key的一致性"""source_type = self.source_client.type(key).decode()target_type = self.target_client.type(key).decode()if source_type != target_type:return False, f"类型不一致: {source_type} vs {target_type}"if source_type == 'string':source_val = self.source_client.get(key)target_val = self.target_client.get(key)return source_val == target_val, "字符串值不一致"elif source_type == 'hash':source_val = self.source_client.hgetall(key)target_val = self.target_client.hgetall(key)return source_val == target_val, "哈希值不一致"# 其他数据类型类似...return True, "一致"def sample_verify(self, sample_size=1000):"""抽样验证"""all_keys = self.source_client.keys('*')sampled_keys = random.sample(all_keys, min(sample_size, len(all_keys)))results = []for key in sampled_keys:is_consistent, message = self.verify_key(key)results.append({'key': key.decode(),'consistent': is_consistent,'message': message})return results# 使用示例
verifier = RedisDataVerifier({'host': '192.168.1.100', 'password': 'source_pass'},{'host': 'cloud-host', 'password': 'cloud_pass'}
)results = verifier.sample_verify(1000)
consistent_count = sum(1 for r in results if r['consistent'])
print(f"一致性比例: {consistent_count/len(results)*100:.2f}%")

4.2 性能基准测试

4.2.1 使用redis-benchmark

# 迁移前在源库测试
redis-benchmark -h 192.168.1.100 -p 6379 -a source_password \-t set,get -n 100000 -c 50 -d 1000# 迁移后在云库测试
redis-benchmark -h r-xxx.redis.rds.aliyuncs.com -p 6379 -a cloud_password \-t set,get -n 100000 -c 50 -d 1000# 对比结果指标：
# - 每秒请求数（Requests per second）
# - 延迟分布（Latency percentile）

4.2.2 自定义性能测试脚本

# performance_test.py
import redis
import time
import statisticsdef test_performance(host, port, password, key_count=10000):client = redis.Redis(host=host, port=port, password=password)# 测试写入性能write_times = []for i in range(key_count):start = time.time()client.set(f'test:key:{i}', 'x' * 100)  # 100字节的值write_times.append(time.time() - start)# 测试读取性能read_times = []for i in range(key_count):start = time.time()client.get(f'test:key:{i}')read_times.append(time.time() - start)# 清理测试数据for i in range(key_count):client.delete(f'test:key:{i}')return {'write_avg': statistics.mean(write_times) * 1000,  # 毫秒'write_p95': statistics.quantiles(write_times, n=20)[18] * 1000,'read_avg': statistics.mean(read_times) * 1000,'read_p95': statistics.quantiles(read_times, n=20)[18] * 1000}# 对比源库和云库性能
source_perf = test_performance('192.168.1.100', 6379, 'source_pass')
cloud_perf = test_performance('r-xxx.redis.rds.aliyuncs.com', 6379, 'cloud_pass')print("源库性能:", source_perf)
print("云库性能:", cloud_perf)
print("性能差异:", {k: cloud_perf[k] - source_perf[k] for k in source_perf})

---

第五章：云Redis优化与监控

5.1 云Redis特定配置

5.1.1 阿里云Redis优化

# 通过阿里云CLI调整参数
aliyun rds ModifyParameter \--DBInstanceId r-xxx \--Parameters '[{"Name":"maxmemory-policy", "Value":"volatile-lru"}]'# 重要参数建议：
# - maxmemory-policy: volatile-lru (避免OOM)
# - hash-max-ziplist-entries: 512 (优化内存使用)
# - slowlog-log-slower-than: 10000 (10ms慢查询阈值)

5.1.2 腾讯云Redis优化

# 腾讯云API调整参数
tccli redis ModifyInstanceParams \--InstanceId crs-xxx \--InstanceParams '[{"Name":"maxmemory-policy", "CurrentValue":"volatile-lru"}]'

5.2 监控与告警配置

5.2.1 云监控集成

# aliyun_monitoring.yaml
metrics:- metric_name: MemoryUsagestatistics: Averageperiod: 300threshold: 85action: alert- metric_name: QPSstatistics: Maximumperiod: 60threshold: 10000action: alert- metric_name: ConnectedClientsstatistics: Averageperiod: 300threshold: 5000action: alertalerts:- name: "HighMemoryUsage"condition: "MemoryUsage > 85% for 2 periods"action: - "notify:ops-team"- "auto-scale:memory"- name: "HighQPS"condition: "QPS > 10000 for 1 period"action: - "notify:dev-team"- "investigate:hotkeys"

---

第六章：回滚与应急方案

6.1 回滚流程设计

6.2 自动化回滚脚本

# rollback_plan.py
class MigrationRollback:def __init__(self, source_config, target_config):self.source_client = redis.Redis(**source_config)self.target_client = redis.Redis(**target_config)def prepare_rollback(self):"""准备回滚：确保源Redis数据最新"""# 1. 停止云Redis写入self.target_client.config_set('readonly', 'yes')# 2. 从云Redis同步最新数据到源Redis# 这里需要根据数据量选择方案：# - 小数据量：直接使用redis-shake反向同步# - 大数据量：使用RDB文件恢复# 3. 验证数据一致性if self.verify_data_consistency():return Trueelse:return Falsedef execute_rollback(self):"""执行回滚"""if self.prepare_rollback():# 1. 切换应用回源Redisupdate_app_config(self.source_config)# 2. 恢复源Redis写入self.source_client.config_set('readonly', 'no')# 3. 通知监控系统notify_monitoring('ROLLBACK_COMPLETED')return Truereturn False# 回滚执行
rollback = MigrationRollback({'host': '192.168.1.100', 'password': 'source_pass'},{'host': 'cloud-host', 'password': 'cloud_pass'}
)if not rollback.execute_rollback():alert_ops_team("回滚失败，需要手动干预")

---

第七章：总结与最佳实践

7.1 迁移检查清单

阶段	检查项	状态	负责人
迁移前	版本兼容性验证	☑️	DBA
	网络连通性测试	☑️	运维
	数据备份完成	☑️	DBA
	迁移方案评审通过	☑️	全体
迁移中	全量数据同步完成	☑️	DBA
	增量同步延迟监控	☑️	运维
	业务流量监控	☑️	开发
迁移后	数据一致性验证	☑️	QA
	性能基准测试	☑️	运维
	监控告警验证	☑️	运维
	文档更新完成	☑️	全体

7.2 云Redis成本优化建议

1. 实例规格选择：

# 根据内存使用量选择合适规格
# 预留20-30%的内存缓冲空间
required_memory = current_usage * 1.3# 根据QPS选择规格
# 云Redis不同规格有对应的QPS上限

2. 存储分离策略：

# 冷热数据分离
# 热点数据放在云Redis
# 冷数据可归档到更便宜的存储（如OSS）
def get_data(key):# 先查Redisvalue = redis_client.get(key)if value is None:# Redis没有，查归档存储value = archive_storage.get(key)if value:# 回种Redis，设置较短TTLredis_client.setex(key, 3600, value)return value

3. 弹性伸缩配置：

# 配置自动扩缩容规则
# 基于QPS或内存使用率触发扩容
aliyun rds CreateScalingRule \--DBInstanceId r-xxx \--RuleType "memory" \--MetricName "MemoryUsage" \--Threshold 80 \--ScalingValue "+20%"