Redis 主从复制怎样进行同步
主从复制概述
Redis 主从复制是 Redis 高可用架构的基础，通过将主节点（Master）的数据同步到从节点（Slave），实现数据备份、读写分离和故障转移。复制过程分为全量同步和增量同步两个阶段。
主从复制的核心优势：
- 数据安全性: 多副本保证数据不丢失
- 读写分离: 主节点写，从节点读，提升性能
- 高可用性: 主节点故障时可快速切换
- 负载均衡: 多个从节点分担读压力
复制建立流程
连接建立阶段
当从节点启动复制时，会经历以下步骤：
实际应用场景示例：
// 配置主从复制的伪代码示例
func setupReplication(masterAddr, slaveAddr string) error {
// 1. 从节点连接主节点
conn, err := net.Dial("tcp", masterAddr)
if err != nil {
return fmt.Errorf("连接主节点失败: %v", err)
}
// 2. 发送 SLAVEOF 命令
command := fmt.Sprintf("SLAVEOF %s %s",
strings.Split(masterAddr, ":")[0],
strings.Split(masterAddr, ":")[1])
// 3. 认证（如果需要）
if password != "" {
_, err = conn.Write([]byte(fmt.Sprintf("AUTH %s\r\n", password)))
if err != nil {
return fmt.Errorf("认证失败: %v", err)
}
}
return nil
}
数据同步准备
在正式同步之前，主从节点需要交换关键信息：
从节点发送的信息：
listening-port: 从节点监听端口ip-address: 从节点 IP 地址capa: 支持的复制能力（如 psync2、eof）
主节点的响应：
- 记录从节点信息到连接列表
- 准备复制偏移量和运行 ID
- 检查复制积压缓冲区状态
PSYNC 协议详解
PSYNC 命令格式
Redis 使用 PSYNC 协议来决定使用全量同步还是增量同步：
PSYNC 命令的三种情况：
- 首次复制:
PSYNC ? -1 - 重连复制:
PSYNC <runid> <offset> - 版本不支持: 降级为
SYNC命令
// PSYNC 协议处理伪代码
func handlePSYNC(runID string, offset int64) string {
currentRunID := getCurrentRunID()
// 情况1: 首次连接或 runID 不匹配
if runID == "?" || runID != currentRunID {
return fmt.Sprintf("+FULLRESYNC %s 0", currentRunID)
}
// 情况2: 检查是否可以增量同步
if canPartialSync(offset) {
return "+CONTINUE"
}
// 情况3: 偏移量过旧，需要全量同步
return fmt.Sprintf("+FULLRESYNC %s 0", currentRunID)
}
func canPartialSync(offset int64) bool {
// 检查偏移量是否在积压缓冲区范围内
backlogStart := getCurrentOffset() - getBacklogSize()
return offset >= backlogStart
}
主节点响应类型
主节点会根据情况返回三种响应：
1. +FULLRESYNC: 执行全量同步
+FULLRESYNC <runid> <offset>
2. +CONTINUE: 执行增量同步
+CONTINUE
3. -ERR: 错误情况
-ERR Can't PSYNC with 2.6.x masters
全量同步过程
全量同步适用于首次复制或数据差异过大的情况：
全量同步的关键步骤
步骤1: RDB 快照生成
主节点通过 BGSAVE 命令在后台生成 RDB 文件，这个过程不会阻塞主节点继续处理客户端请求。
步骤2: 命令缓存 在 RDB 生成期间，主节点会将所有写命令记录到复制积压缓冲区中。
步骤3: RDB 传输 RDB 文件生成完成后，主节点将文件内容发送给从节点。
步骤4: 数据加载 从节点清空现有数据，然后加载 RDB 文件恢复数据状态。
步骤5: 增量补齐 从节点加载完 RDB 后，主节点发送积压缓冲区中的命令，确保数据完全同步。
// 全量同步处理伪代码
func performFullSync(slave *SlaveConnection) error {
// 1. 生成 RDB 快照
rdbFile, err := generateRDBSnapshot()
if err != nil {
return fmt.Errorf("生成RDB失败: %v", err)
}
// 2. 发送 RDB 文件
err = sendRDBFile(slave, rdbFile)
if err != nil {
return fmt.Errorf("发送RDB失败: %v", err)
}
// 3. 发送积压缓冲区命令
commands := getBacklogCommands(slave.offset)
for _, cmd := range commands {
err = slave.SendCommand(cmd)
if err != nil {
return fmt.Errorf("发送命令失败: %v", err)
}
}
return nil
}
增量同步机制
增量同步通过复制积压缓冲区实现，避免了重复传输大量数据：
积压缓冲区工作原理
缓冲区特性：
- 环形结构: 固定大小的环形缓冲区，默认 1MB
- FIFO 机制: 新命令覆盖最旧的命令
- 偏移量标记: 每个命令都有唯一的偏移量
增量同步条件：
- 从节点的复制偏移量在积压缓冲区范围内
- 主从节点的
runid匹配 - 网络连接正常
// 增量同步处理伪代码
func performPartialSync(slave *SlaveConnection) error {
startOffset := slave.replicationOffset
endOffset := getCurrentOffset()
// 检查偏移量范围
if !isOffsetInBacklog(startOffset) {
return errors.New("偏移量超出积压缓冲区范围")
}
// 获取增量命令
commands := getCommandsInRange(startOffset+1, endOffset)
// 发送增量命令
for _, cmd := range commands {
err := slave.SendCommand(cmd)
if err != nil {
return fmt.Errorf("发送增量命令失败: %v", err)
}
slave.replicationOffset++
}
return nil
}
type BacklogBuffer struct {
buffer []byte
size int64
offset int64 // 当前写入位置的偏移量
}
func (b *BacklogBuffer) AddCommand(cmd []byte) {
// 环形缓冲区写入逻辑
pos := b.offset % b.size
copy(b.buffer[pos:], cmd)
b.offset += int64(len(cmd))
}
心跳和偏移量管理
心跳机制
主从节点通过定期心跳维持连接状态：
心跳的双重作用：
- 连接检测: 确保主从连接正常
- 偏移量同步: 追踪数据同步进度
偏移量管理
偏移量是 Redis 复制的核心概念，用于追踪同步进度：
// 偏移量管理伪代码
type ReplicationState struct {
masterOffset int64 // 主节点当前偏移量
slaveOffset int64 // 从节点已确认偏移量
lag int64 // 同步延迟
}
func updateReplicationOffset(slave *SlaveConnection, ackOffset int64) {
slave.lastAckTime = time.Now()
slave.replicationOffset = ackOffset
// 计算同步延迟
slave.lag = getCurrentOffset() - ackOffset
// 检查是否需要告警
if slave.lag > maxAllowedLag {
logWarning("从节点同步延迟过大: %d", slave.lag)
}
}
func handleReplconfAck(slave *SlaveConnection, offset string) {
ackOffset, err := strconv.ParseInt(offset, 10, 64)
if err != nil {
logError("无效的偏移量格式: %s", offset)
return
}
updateReplicationOffset(slave, ackOffset)
}
断线重连机制
重连流程
网络断开后的重连过程展示了 Redis 复制的健壮性：
重连的关键点：
- runid 验证: 确保连接的是同一个主节点
- 偏移量检查: 判断是否可以增量同步
- 超时处理: 防止无限重连消耗资源
断线重连优化
// 断线重连处理伪代码
func handleReconnection(slave *SlaveConnection) error {
maxRetries := 3
retryInterval := time.Second * 5
for attempt := 1; attempt <= maxRetries; attempt++ {
// 尝试重新连接
conn, err := establishConnection(slave.address)
if err != nil {
logWarning("重连尝试 %d/%d 失败: %v", attempt, maxRetries, err)
time.Sleep(retryInterval)
continue
}
// 发送 PSYNC 命令
psyncCmd := fmt.Sprintf("PSYNC %s %d", slave.runID, slave.offset)
response, err := sendCommand(conn, psyncCmd)
if err != nil {
conn.Close()
continue
}
// 处理响应
switch {
case strings.HasPrefix(response, "+CONTINUE"):
return performPartialSync(slave)
case strings.HasPrefix(response, "+FULLRESYNC"):
return performFullSync(slave)
default:
return fmt.Errorf("未知的PSYNC响应: %s", response)
}
}
return fmt.Errorf("重连失败，已达到最大重试次数")
}
性能优化和最佳实践
复制性能优化
关键优化策略：
- 积压缓冲区调优
# 增大积压缓冲区 ，减少全量同步概率
repl-backlog-size 64mb
repl-backlog-ttl 3600
- 网络优化
# 禁用 TCP Nagle 算法，减少延迟
tcp-nodelay yes
# 设置合适的发送缓冲区
repl-backlog-size 16mb
- RDB 优化
# 使用无磁盘复制（适用于快速网络）
repl-diskless-sync yes
repl-diskless-sync-delay 5
监控指标
重要的复制监控指标：
// 关键监控指标
type ReplicationMetrics struct {
// 连接状态
SlaveConnected bool
SlaveState string // "wait_bgsave", "send_bulk", "online"
SlaveLag int64 // 同步延迟（秒）
// 数据同步
MasterReplOffset int64 // 主节点偏移量
SlaveReplOffset int64 // 从节点偏移量
ReplBacklogSize int64 // 积压缓冲 区大小
// 性能指标
ReplOutputBuffer int64 // 输出缓冲区大小
SyncFullCount int64 // 全量同步次数
SyncPartialOk int64 // 增量同步成功次数
SyncPartialErr int64 // 增量同步失败次数
}
func collectReplicationMetrics() *ReplicationMetrics {
// 通过 INFO replication 命令获取指标
info := execCommand("INFO replication")
return parseReplicationInfo(info)
}
故障处理
常见问题及解决方案：
-
同步延迟过大
- 检查网络带宽和延迟
- 优化主节点写入性能
- 增大复制缓冲区
-
频繁全量同步
- 增大
repl-backlog-size - 检查网络稳定性
- 监控从节点重启情况
- 增大
-
复制中断
- 检查
repl-timeout设置 - 监控网络连接状态
- 验证防火墙配置
- 检查
Redis 主从复制通过精心设计的 PSYNC 协议和积压缓冲区机制，在保证数据一致性的同时，最大化了复制效率。理解这些机制有助于在生产环境中更好地配置和监控 Redis 集群。