渐进式解析 Redis 源码 - 复制 replication

阅读量

复制 作为现代数据服务基本必备的功能，保证 数据的冗余备份，从而保障服务高可用和高可靠性，当然 Redis 这么伟大的服务也不例外

本章可能有点儿绕，做好心理准备… ￣ω￣=

Redis 复制( replicate ) 流程的核心在于 状态机 中状态的流转，连接各个逻辑运算，我根据 Redis ver4.0.11 的源码逻辑画了一个简单的 master-slave 状态机流转示意图，由于一般 从库的挂载都是由 slave 发起的，所以 slave 放在左边，大家可以先看一眼，我们后续进行每个步骤的详细讲解
复制流程

Redis 复制模块 由于有些方法里的代码量比较大而且有点儿绕，所以我们分流程在文字详解的时候一块解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

状态机流转

发起建立 master-slave 关系

发起 master-slave 关系有几种方式，不过我们通常是在 conf 里面进行配置

  • redis.conf 文件中配置 slaveof [masterip] [masterport] 选项，当然还是需要在服务启动的时候指定这个 redis.conf 配置
  • redis-server 命令启动服务的时候指定参数 --slaveof [masterip] [masterport]
  • client 模式下通过 slaveof [masterip] [masterport] 命令执行绑定主服务器

上面三种方式的主从复制，都是由 从服务器 主动执行的，最终命令使用的是 slaveof 命令的函数逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
/* replication.c: line 1992 */
void slaveofCommand(client *c) {
    // slaveof 命令不允许在 cluster 集群模式下使用
    if (server.cluster_enabled) {
        addReplyError(c,"SLAVEOF not allowed in cluster mode.");
        return;
    }
    // 如果是 SLAVEOF NO ONE 命令，使得这个从属服务器关闭复制功能，并从从属服务器转变回主服务器，原来同步所得的数据集不会被丢弃，利用『 SLAVEOF NO ONE 不会丢弃同步所得数据集』这个特性，可以在主服务器失败的时候，将从属服务器用作新的主服务器，从而实现无间断运行
    if (!strcasecmp(c->argv[1]->ptr,"no") &&
        !strcasecmp(c->argv[2]->ptr,"one")) {
        // 如果之前有主服务器
        if (server.masterhost) {
            // 关闭复制，重置主服务器，将当前 服务器升级为 主服务器
            replicationUnsetMaster();
            // 获取 client 信息，并以 sds 格式返回，方便打印日志
            sds client = catClientInfoString(sdsempty(),c);
            serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')", client);
            sdsfree(client);
        }
    } else {
    // slaveof [masterip] [masterport] 命令
    
        long port;
        // 获取端口号
        if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != C_OK))
        return;
        // 如果已经存在 主服务器配置，而且跟 此次命令的参数一致，那么就直接返回，不进行重复处理了
        if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr) && server.masterport == port) {
            serverLog(LL_NOTICE,"SLAVE OF would result into synchronization with the master we are already connected with. No operation performed.");
            addReplySds(c,sdsnew("+OK Already connected to specified master\r\n"));
            return;
        }
        // 设置主服务器地址和端口
        replicationSetMaster(c->argv[1]->ptr, port);
        // 获取 client 信息，并以 sds 格式返回，方便打印日志
        sds client = catClientInfoString(sdsempty(),c);
        serverLog(LL_NOTICE,"SLAVE OF %s:%d enabled (user request from '%s')", server.masterhost, server.masterport, client);
        sdsfree(client);
    }
    addReply(c,shared.ok);
}

我们还要看一下设置主服务器地址和端口的函数，同时也会有状态机的一部分，状态设置为 REPL_STATE_CONNECT

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/* replication.c: line 1927 */
void replicationSetMaster(char *ip, int port) {
    int was_master = server.masterhost == NULL;
    
    sdsfree(server.masterhost);
    // 设置 主服务器地址和端口
    server.masterhost = sdsnew(ip);
    server.masterport = port;
    // 如果已经设置过其他主服务器则释放
    if (server.master) {
        freeClient(server.master);
    }
    // 解除所有客户端的阻塞状态
    disconnectAllBlockedClients();
    // 关闭所有从服务器的连接，强制从服务器进行重新同步操作
    disconnectSlaves();
    // 取消复制操作
    cancelReplicationHandshake();
    // 保存 master 信息至缓存
    if (was_master) replicationCacheMasterUsingMyself();
    // 状态机状态设置为 REPL_STATE_CONNECT
    server.repl_state = REPL_STATE_CONNECT;
}

当前状态机状态为 REPL_STATE_CONNECT

master-slave 连接建立

slaveof 操作为一个异步操作，由 从服务发起 后直接返回响应结果，随后由 时间事件 回调函数 serverCron() 里面逻辑发起 调用 replicationCron() 函数

1
2
3
/* server.c: line 1144 */
// 周期性执行复制的任务
run_with_period(1000) replicationCron();

replicationCron() 每秒会被执行一次，如果发现 有主服务器信息 而且 状态机状态为 REPL_STATE_CONNECT，那么就会发起建立连接

1
2
3
4
5
6
7
8
/* replication.c: line 2530 */
// 摘自 replicationCron()
if (server.repl_state == REPL_STATE_CONNECT) {
    serverLog(LL_NOTICE,"Connecting to MASTER %s:%d", server.masterhost, server.masterport);
    if (connectWithMaster() == C_OK) {
        serverLog(LL_NOTICE,"MASTER <-> SLAVE sync started");
    }
}

上述逻辑中可以看到，调用 connectWithMaster()函数 以非阻塞的方式连接主服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/* replication.c: line 1855 */
int connectWithMaster(void) {
    int fd;
    // 连接主服务器
    fd = anetTcpNonBlockBestEffortBindConnect(NULL,server.masterhost,server.masterport,NET_FIRST_BIND_ADDR);
    // 如果连接失败
    if (fd == -1) {
        serverLog(LL_WARNING,"Unable to connect to MASTER: %s", strerror(errno));
        return C_ERR;
    }
    // 为连接句柄建立读写事件监听，并设定 syncWithMaster() 回调函数，方便后续主从的数据交流
    if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) == AE_ERR)
    {
        // 建立事件监听失败
        close(fd);
        serverLog(LL_WARNING,"Can't create readable event for SYNC");
        return C_ERR;
    }
    // 最近一次读取到 RDB 数据的时间，这里还有一层意思，避免超时被干掉
    server.repl_transfer_lastio = server.unixtime;
    // 主从连接套接字
    server.repl_transfer_s = fd;
    // 状态机状态变更
    server.repl_state = REPL_STATE_CONNECTING;
    return C_OK;
}

当前状态机状态为 REPL_STATE_CONNECTING

ping-pong 的发送/响应

刚才在上一个步骤中建立了与 主服务器的连接，而且建立了读写事件，所以会触发 AE_WRITABLE 事件，触发回调函数 syncWithMaster()，发送 PING 给主服务器，并且等待 PONG 响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/* replication.c: line 1604 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_CONNECTING) {
    serverLog(LL_NOTICE,"Non blocking connect for SYNC fired the event.");
    // 因为后续只需要等待 PONG 响应即可，所以只需要后续的 PONG 读操作即可
    aeDeleteFileEvent(server.el,fd,AE_WRITABLE);
    // 状态机为 REPL_STATE_RECEIVE_PONG等待响应状态
    server.repl_state = REPL_STATE_RECEIVE_PONG;
    // 写方式 发送 PING 命令
    err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PING",NULL);
    if (err) goto write_error;
    return;
}

主服务器在接到 PING 信息后 会返回 +PONG\r\n，触发 从服务器写事件，调用 之前我们新建的从服务器读事件，回调 syncWithMaster() 进行处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/* replication.c: line 1618 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_RECEIVE_PONG) {
    // 读取返回的 PONG 结果信息
    err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
    // 一般会返回结果 +PONG，如果需要 认证则会返回 -NOAUTH，旧版本也可能会返回 -ERR operation not permitted
    if (err[0] != '+' &&
        strncmp(err,"-NOAUTH",7) != 0 &&
        strncmp(err,"-ERR operation not permitted",28) != 0)
    {
        serverLog(LL_WARNING,"Error reply to PING from master: '%s'",err);
        sdsfree(err);
        goto error;
    } else {
        serverLog(LL_NOTICE, "Master replied to PING, replication can continue...");
    }
    sdsfree(err);
    // 状态机为 REPL_STATE_SEND_AUTH
    server.repl_state = REPL_STATE_SEND_AUTH;
}

当前状态机状态为 REPL_STATE_SEND_AUTH

auth 权限认证

上一个状态的逻辑还没有 return，所以还会继续在 syncWithMaster() 函数中进行处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/* replication.c: line 1642 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_AUTH) {
    // 如果设置的有认证信息
    if (server.masterauth) {
        // 写方式 发送认证信息
        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",server.masterauth,NULL);
        if (err) goto write_error;
        // 状态机状态直接为 REPL_STATE_RECEIVE_AUTH 等待认证结果
        server.repl_state = REPL_STATE_RECEIVE_AUTH;
        return;
    } else {
        // 状态机状态为 REPL_STATE_SEND_PORT 发送端口信息给主服务器
        server.repl_state = REPL_STATE_SEND_PORT;
    }
}

根据上述逻辑，如果发送了认证信息，那么我们就需要等待 主服务器的认证结果返回，当前状态为 REPL_STATE_RECEIVE_AUTH，主服务器接到认证信息后，返回结果，从服务读取结果，继续处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/* replication.c: line 1654 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_RECEIVE_AUTH) {
    // 读取返回的 认证结果信息
    err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
    // 成功信息前缀一般为 +，前缀为 - 为返回错误
    if (err[0] == '-') {
        serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);
        sdsfree(err);
        goto error;
    }
    sdsfree(err);
    // 认证成功，状态机状态为 REPL_STATE_SEND_PORT 发送端口信息给主服务器
    server.repl_state = REPL_STATE_SEND_PORT;
}

当前状态机状态为 REPL_STATE_SEND_PORT

ip/port 发送

上一个状态中我们看到状态机状态转为发送 端口信息的时候 没有 return，所以继续运算处理，发送 端口信息，状态码状态改为 REPL_STATE_RECEIVE_PORT

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/* replication.c: line 1667 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_PORT) {
    // 获取端口号
    sds port = sdsfromlonglong(server.slave_announce_port ?
        server.slave_announce_port : server.port);
    // 写方式 发送端口信息
    err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
            "listening-port",port, NULL);
    sdsfree(port);
    if (err) goto write_error;
    sdsfree(err);
    // 状态机状态为 REPL_STATE_RECEIVE_PORT 接受端口发送后的响应结果
    server.repl_state = REPL_STATE_RECEIVE_PORT;
    return;
}

我们注意到上面发送端口信息的时候，使用了一个 REPLCONF listening-port [port]，主服务器读取到信息以后，调用了 replconfCommand() 命令函数进行处理，提取端口信息，保存至 c->slave_listening_port 中，然后响应 +OK\r\n成功状态信息，通过 fd 句柄响应给 从服务器，从服务器读取主服务返回结果，状态码状态改为 REPL_STATE_SEND_IP

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/* replication.c: line 1680 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_RECEIVE_PORT) {
    // 读取主服务器响应信息
    err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
    // 成功信息前缀一般为 +，前缀为 - 为返回错误
    if (err[0] == '-') {
        serverLog(LL_NOTICE,"(Non critical) Master does not understand REPLCONF listening-port: %s", err);
    }
    sdsfree(err);
    // 状态机状态为 REPL_STATE_SEND_IP 发送IP地址
    server.repl_state = REPL_STATE_SEND_IP;
}

从服务器发送完端口号并且正确收到主服务器的回复后，状态机状态为 REPL_STATE_SEND_IP，紧接着 syncWithMaster() 函数执行发送IP的代码，发送IP和发送端口号过程几乎一致，发送完成以后状态机状态修改为 REPL_STATE_RECEIVE_IP

1
2
3
4
5
6
7
8
9
10
11
12
13
/* replication.c: line 1701 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_IP) {
    // 写方式 发送 IP 信息
    err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
            "ip-address",server.slave_announce_ip, NULL);
    if (err) goto write_error;
    sdsfree(err);
    // 状态机状态为 REPL_STATE_RECEIVE_IP 接收IP地址响应信息
    server.repl_state = REPL_STATE_RECEIVE_IP;
    return;
}

发送 IP 地址信息给主服务器， 我们可以看到，跟发送 port 的时候一样逻辑，主服务器接收到了信息，调用 replconfCommand() 命令函数进行处理，提取 IP 信息，保存至 c->slave_ip 中，从服务器读取主服务返回结果，状态码状态改为 REPL_STATE_SEND_CAPA

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/* replication.c: line 1711 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_RECEIVE_IP) {
    // 读取主服务器响应信息
    err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
    // 成功信息前缀一般为 +，前缀为 - 为返回错误
    if (err[0] == '-') {
        serverLog(LL_NOTICE,"(Non critical) Master does not understand REPLCONF ip-address: %s", err);
    }
    sdsfree(err);
    // 状态机状态为 REPL_STATE_SEND_CAPA 发送 capa 能力信息
    server.repl_state = REPL_STATE_SEND_CAPA;
}

当然如果从服务器里面没有保存需要传递给主服务的 IP 地址的时候，就直接跳转到下一个步骤

1
2
3
4
5
6
7
/* replication.c: line 1693 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_IP && server.slave_announce_ip == NULL)
{
    server.repl_state = REPL_STATE_SEND_CAPA;
}

当前状态机状态为 REPL_STATE_SEND_CAPA，发送 从服务器的数据处理能力（capability），当前版本通常代表为 能否解析出 RDB 文件的 EOF流 格式

capability 能力发送

发送 capability 能力（当前版本通常代表为 能否解析出 RDB 文件的 EOF流 格式）跟上面的 port 信息基本一致，主要为 发送 从服务的数据处理能力，而服务器接收到信息以后使用 replconfCommand() 函数进行提取，赋值 c->slave_capa，然后回复 +OK\r\n，从服务状态接收回复信息后状态机变更为 REPL_STATE_SEND_PSYNC

1
2
3
4
5
6
7
8
9
10
11
12
13
/* replication.c: line 1729 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_CAPA) {
    // 写方式 发送 capa 能力信息，当前版本一般为 eof 支持
    err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
            "capa","eof","capa","psync2",NULL);
    if (err) goto write_error;
    sdsfree(err);
    // 状态机状态为 REPL_STATE_RECEIVE_CAPA 接收 capa 能力响应信息
    server.repl_state = REPL_STATE_RECEIVE_CAPA;
    return;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/* replication.c: line 1739 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_RECEIVE_CAPA) {
    // 读取发送 capa 后的 主服务器响应信息
    err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
    // 成功信息前缀一般为 +，前缀为 - 为返回错误
    if (err[0] == '-') {
        serverLog(LL_NOTICE,"(Non critical) Master does not understand REPLCONF capa: %s", err);
    }
    sdsfree(err);
    // 状态机状态为 REPL_STATE_SEND_PSYNC 发送 psync命令信息
    server.repl_state = REPL_STATE_SEND_PSYNC;
}

当前状态机状态为 REPL_STATE_SEND_PSYNC 发送同步信息给主服务器

psync 响应

从这里开始跟前面的各个状态机状态就不太一致了，而且比较绕 就体现在这个步骤

状态机处于 REPL_STATE_SEND_PSYNC 状态的时候，从服务器会尝试向主服务器发起同步请求，一般请求分为 全量同步 和 增量同步；首次一般都是 全量同步，而全量后的普通同步 和 由于网络等原因断开后重连的同步 会选择增量同步

秉着高效的原理，Redis v2.8 版本的时候引入了 PSYNC，主从可以增量同步，这样当主从链接短时间中断恢复后，无需做完整的RDB完全同步这种重量级操作

所以 从服务器在连接上主服务器后 首先尝试的是 增量同步，因为 有可能是 断线后重连的情况，如果判断发现不是重连的情况不能进行 增量同步，就进行一次全量同步

接状态机的上个步骤状态REPL_STATE_SEND_PSYNC

1
2
3
4
5
6
7
8
9
10
11
12
13
/* replication.c: line 1756 */
// 摘自 syncWithMaster()
if (server.repl_state == REPL_STATE_SEND_PSYNC) {
    // 先尝试发起全量同步
    if (slaveTryPartialResynchronization(fd,0) == PSYNC_WRITE_ERROR) {
        err = sdsnew("Write error sending the PSYNC command.");
        goto write_error;
    }
    // 状态机状态为 REPL_STATE_RECEIVE_PSYNC 接收 psync命令信息
    server.repl_state = REPL_STATE_RECEIVE_PSYNC;
    return;
}

我们接下来看一下 尝试发送 全量同步的函数 slaveTryPartialResynchronization()，需要注意的是 第二个参数传递的是0，代表 写命令，我们 摘取 写逻辑部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
/* replication.c: line 1422 */
// 摘自 slaveTryPartialResynchronization()
/* Writing half */
if (!read_reply) {
    // 初始化master_initial_offset 为 -1，标记当前 run_id 和 offset偏移量 无效；后面进行全量同步的时候会根据响应值进行设置
    server.master_initial_offset = -1;
    // 如果缓存不为空，则可以进行增量同步，证明为 断开重连的
    if (server.cached_master) {
        // 从服务器在主服务器里的标识 runid
        psync_replid = server.cached_master->replid;
        // 复制偏移量
        snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
        // 记录日志
        serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset);
    } else {
        // 全量同步
        // 记录日志
        serverLog(LL_NOTICE,"Partial resynchronization not possible (no cached master)");
        // 从服务器在主服务器里的标识 runid
        psync_replid = "?";
        // 复制偏移量
        memcpy(psync_offset,"-1",3);
    }
    // 发送 PSYNC 命令给主服务器，同时传递 runid 和 复制偏移量 信息
    reply = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PSYNC",psync_replid,psync_offset,NULL);
    // 如果发送失败，则记录日志并 删除 读事件，返回错误
    if (reply != NULL) {
        serverLog(LL_WARNING,"Unable to send PSYNC to master: %s",reply);
        sdsfree(reply);
        aeDeleteFileEvent(server.el,fd,AE_READABLE);
        return PSYNC_WRITE_ERROR;
    }
    // 返回 等待回复标识PSYNC_WAIT_REPLY，调用逻辑会将 read_reply 设置为1，然后再次调用该函数，执行下面的读部分
    return PSYNC_WAIT_REPLY;
}

上面逻辑可以看到 从服务器发送了一个 PSYNC ? -1 信息给 主服务器，代表申请全量同步，随后 主服务器调用 syncCommand() 命令函数进行处理，这个部分的逻辑稍微有点儿复杂，不像 之前几个步骤，交互比较单一，我们需要特地进行解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
/* replication.c: line 627 */
void syncCommand(client *c) {
    // 如果 client 是从服务器，忽略
    if (c->flags & CLIENT_SLAVE) return;
    // 如果当前也是 从服务器，并且跟 主服务器没有连接，发送错误，直接返回
    if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) {
        addReplySds(c,sdsnew("-NOMASTERLINK Can't SYNC while not connected with my master\r\n"));
        return;
    }
    // 如果 client 缓冲区里还有数据，则不能执行
    if (clientHasPendingReplies(c)) {
        addReplyError(c,"SYNC and PSYNC are invalid with pending output");
        return;
    }
    // 记录日志
    serverLog(LL_NOTICE,"Slave %s asks for synchronization", replicationGetSlaveName(c));
    
    // 如果执行的命令为 psync
    if (!strcasecmp(c->argv[0]->ptr,"psync")) {
        // 尝试部分同步，如果成功证明不需要全量了
        if (masterTryPartialResynchronization(c) == C_OK) {
            // 可以执行PSYNC命令，则将接受PSYNC命令的个数加1
            server.stat_sync_partial_ok++;
            // 不需要全量同步了，直接返回
            return;
        } else {
        // 需要全量同步
            // 传递的命令参数
            char *master_replid = c->argv[1]->ptr;
            // 从服务器传递 ？强制全量同步，所以不能执行部分同步，所以 部分同步失败 +1
            if (master_replid[0] != '?') server.stat_sync_partial_err++;
        }
    } else {
    // sync 命令
        c->flags |= CLIENT_PRE_PSYNC;
    }
    // 全部同步统计+1
    server.stat_sync_full++;
    // 状态为 从服务器等待 BGSAVE 开始
    c->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
    // 执行 sync 后是否关闭 TCP_NODELAY
    if (server.repl_disable_tcp_nodelay)
        // 启用nagle算法
        anetDisableTcpNoDelay(NULL, c->fd);
    // 保存主服务传递过来的 RDB 文件 fd，设置为 -1
    c->repldbfd = -1;
    // 标识 client 为一个从服务器
    c->flags |= CLIENT_SLAVE;
    // 将client 添加到 从服务器列表中
    listAddNodeTail(server.slaves,c);
    // 如果从服务器列表只有一个元素而且 backlog 为空，证明刚初始化
    if (listLength(server.slaves) == 1 && server.repl_backlog == NULL) {
        // 初始化 replication id，主要是用来为后续 master-slave 模式分配唯一 id
        changeReplicationId();
        // 清除辅助 replication id，这种情况有可能发生在 在完全重新同步之后开始新的复制历史记录时
        clearReplicationId2();
        // 初始化 backlog
        createReplicationBacklog();
    }
    // 情况1：BGSAVE 已经正在执行了，且是同步到磁盘上
    if (server.rdb_child_pid != -1 &&
        server.rdb_child_type == RDB_CHILD_TYPE_DISK)
    {
        client *slave;
        listNode *ln;
        listIter li;
        listRewind(server.slaves,&li);
        // 遍历从服务器列表
        while((ln = listNext(&li))) {
            slave = ln->value;
            // 如果有从服务器已经创建子进程执行写RDB操作，等待完成，那么退出循环
            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) break;
        }
        // 对于这个从服务器，我们检查它是否具有触发当前BGSAVE操作的能力，这个也就是之前我们传递的 复制能力
        if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) {
            // 将slave的输出缓冲区所有内容拷贝给c的所有输出缓冲区中
            copyClientOutputBuffer(c,slave);
            // 设置全量重同步从服务器的状态，设置部分重同步的偏移量
            replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);
            serverLog(LL_NOTICE,"Waiting for end of BGSAVE for SYNC");
        } else {
            // 从服务能力不符合条件
            serverLog(LL_NOTICE,"Can't attach the slave to the current BGSAVE. Waiting for next BGSAVE for SYNC");
        }
    // 情况2：BGSAVE 已经正在执行了，且是无盘同步，直接写到 socket 中
    } else if (server.rdb_child_pid != -1 &&
               server.rdb_child_type == RDB_CHILD_TYPE_SOCKET)
    {
        // 虽然有子进程在执行写 RDB，但是它直接写到 socket 中，不能复制，所以等待下次执行 BGSAVE
        serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC");
    // 情况3：BGSAVE 没有在执行
    } else {
        // 如果服务器支持无盘同步
        if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF)) {
            // 无盘同步复制的子进程被创建在 replicationCron() 中，因为想等待更多的从服务器可以到来而延迟，为了更多 slave 共享
            if (server.repl_diskless_sync_delay)
                serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC");
        } else {
        // 服务器不支持无盘复制
            // 如果没有正在执行 BGSAVE，且没有进行写 AOF文件，则开始为复制执行 BGSAVE，并且是将 RDB 文件写到磁盘上
            if (server.aof_child_pid == -1) {
                startBgsaveForReplication(c->slave_capa);
            } else {
                serverLog(LL_NOTICE,
                    "No BGSAVE in progress, but an AOF rewrite is active. BGSAVE for replication delayed");
            }
        }
    }
    return;
}

然后紧接着开始 复制执行BGSAVE 操作，也就印证我们最上方流程图里面的函数 startBgsaveForReplication()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
/* replication.c: line 563 */
int startBgsaveForReplication(int mincapa) {
    int retval;
    // 是否直接写 socket
    int socket_target = server.repl_diskless_sync && (mincapa & SLAVE_CAPA_EOF);
    listIter li;
    listNode *ln;
    // 日志
    serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s", socket_target ? "slaves sockets" : "disk");
    rdbSaveInfo rsi, *rsiptr;
    // RDB 的保存信息
    rsiptr = rdbPopulateSaveInfo(&rsi);
    // RDB 保存信息有效
    if (rsiptr) {
        // socket 保存文件
        if (socket_target)
            retval = rdbSaveToSlavesSockets(rsiptr);
        else
        // RDB 文件保存到磁盘上
            retval = rdbSaveBackground(server.rdb_filename,rsiptr);
    } else {
        serverLog(LL_WARNING,"BGSAVE for replication: replication information not available, can't generate the RDB file right now. Try later.");
        retval = C_ERR;
    }
    // BGSAVE 执行错误，将等待全量同步的从服务器从从服务器链表中删除，打印发生错误，立即关闭连接
    if (retval == C_ERR) {
        serverLog(LL_WARNING,"BGSAVE for replication failed");
        listRewind(server.slaves,&li);
        // 遍历从服务器链表
        while((ln = listNext(&li))) {
            client *slave = ln->value;
            // 将等待全量同步的从服务器从从服务器点链表中删除
            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
                slave->flags &= ~CLIENT_SLAVE;
                listDelNode(server.slaves,ln);
                addReplyError(slave, "BGSAVE failed, replication can't continue");
                // 立即关闭 client 连接
                slave->flags |= CLIENT_CLOSE_AFTER_REPLY;
            }
        }
        return retval;
    }
    // 如果是直接写到socket中，rdbSaveToSlavesSockets()已经会设置从服务器为全量复制，否则直接写到磁盘上
    if (!socket_target) {
        listRewind(server.slaves,&li);
        // 遍历从服务器链表
        while((ln = listNext(&li))) {
            client *slave = ln->value;
            // 设置等待全量同步的从服务器的状态
            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
                replicationSetupSlaveForFullResync(slave, getPsyncInitialOffset());
            }
        }
    }
    // 刷新脚本的缓存
    if (retval == C_OK) replicationScriptCacheFlush();
    return retval;
}

我们看到上面逻辑，rdbSaveBackground()函数 RDB 文件保存完毕，而且会调用 replicationSetupSlaveForFullResync() 函数给从服务器发送 +FULLRESYNC 命令，而且要发送 主服务器的运行 id server.runid 和 主服务器的全局复制偏移量 server.master_repl_offset

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/* replication.c: line 418 */
int replicationSetupSlaveForFullResync(client *slave, long long offset) {
    char buf[128];
    int buflen;
    // 设置全量重同步的偏移量
    slave->psync_initial_offset = offset;
    // 设置从服务器复制状态，开始累计差异数据
    slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END;
    // 将 slaveseldb 设置为-1，是为了强制发送一个 select 命令在复制流中
    server.slaveseldb = -1;
    
    // 如果从服务器的状态是 CLIENT_PRE_PSYNC，则表示是 Redis v2.8 之前的版本，则不将这些信息发送给从服务器，因为在 2.8 之前只支持 SYNC 的全量复制同步，而在之后的版本提供了部分的重同步
    if (!(slave->flags & CLIENT_PRE_PSYNC)) {
        // 将全量复制的信息写给从服务器
        buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n", server.replid,offset);
        if (write(slave->fd,buf,buflen) != buflen) {
            freeClientAsync(slave);
            return C_ERR;
        }
    }
    return C_OK;
}

至此主服务器端完成了 RDB 保存以及响应 从服务器完成信息，随后就轮到主服务器 将 缓存的数据发送给从服务器

发送缓冲区数据

上个步骤我们看到 主服务已经完成了 RDB 保存而且响应了从服务器，那么又到了状态机逻辑部分了

1
2
3
4
/* replication.c: line 1773 */
// 摘自 syncWithMaster()
psync_result = slaveTryPartialResynchronization(fd,1);

这次调用 slaveTryPartialResynchronization() 函数后面的参数传递的为 1，走读逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
/* replication.c: line 1452 */
// 摘自 slaveTryPartialResynchronization()
/* Reading half */
// 读方式获取 主服务器返回的响应数据
reply = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
// 主服务器为了保持连接的状态，可能会在接收到PSYNC命令后发送一个空行，但是还是会返回 PSYNC_WAIT_REPLY 状态，下次再读取，重走当前逻辑
if (sdslen(reply) == 0) {
    sdsfree(reply);
    return PSYNC_WAIT_REPLY;
}
// 删除可读事件，方便后面的处理
aeDeleteFileEvent(server.el,fd,AE_READABLE);
// 接受到的是 +FULLRESYNC ，表示进行一次全量同步
if (!strncmp(reply,"+FULLRESYNC",11)) {
    char *replid = NULL, *offset = NULL;
    // 提取 runid 和 偏移量
    replid = strchr(reply,' ');
    if (replid) {
        replid++;
        offset = strchr(replid,' ');
        if (offset) offset++;
    }
    // 如果 runid 和 偏移量 出现问题，那么就 报错、重置
    if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) {
        serverLog(LL_WARNING, "Master replied with wrong +FULLRESYNC syntax.");
        memset(server.master_replid,0,CONFIG_RUN_ID_SIZE+1);
    } else {
        // 保存 偏移量 和 runid
        memcpy(server.master_replid, replid, offset-replid-1);
        server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
        server.master_initial_offset = strtoll(offset,NULL,10);
        serverLog(LL_NOTICE,"Full resync from master: %s:%lld", server.master_replid, server.master_initial_offset);
    }
    // 清空 缓存的主服务器结构没用了，因为要执行全量了
    replicationDiscardCachedMaster();
    sdsfree(reply);
    return PSYNC_FULLRESYNC;
}

返回 PSYNC_FULLRESYNC 状态后 syncWithMaster() 继续往下走

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
/* replication.c: line 1810 */
// 摘自 syncWithMaster()
// 执行到这里，状态为 PSYNC_FULLRESYNC 或 PSYNC_NOT_SUPPORTED
// 准备一个合适临时文件用来写入和保存主服务器传来的RDB文件数据
while(maxtries--) {
    // 设置文件名
    snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
    // 可 读 写 执行 权限设置文件
    dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
    if (dfd != -1) break;
    sleep(1);
}
// 临时文件打开失败
if (dfd == -1) {
    serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
    goto error;
}
// 监听一个 fd 的读事件，并设置该事件的处理程序为 readSyncBulkPayload() 函数
if (aeCreateFileEvent(server.el,fd,AE_READABLE,readSyncBulkPayload,NULL) == AE_ERR)
{
    serverLog(LL_WARNING,
        "Can't create readable event for SYNC: %s (fd=%d)",strerror(errno),fd);
    goto error;
}
// 状态机状态更改为正从主服务器接受 RDB 文件
server.repl_state = REPL_STATE_TRANSFER;
// 初始化 RDB 文件大小
server.repl_transfer_size = -1;
// 已读大小
server.repl_transfer_read = 0;
// 最近一次执行的 fsync 偏移量
server.repl_transfer_last_fsync_off = 0;
// 传输 RDB 文件的临时句柄
server.repl_transfer_fd = dfd;
// 最近一次读取 RDB 文件内容的时间
server.repl_transfer_lastio = server.unixtime;
// 保存 RDB 文件临时文件名
server.repl_transfer_tmpfile = zstrdup(tmpfile);
return;

从上面看出主服务器发送 RDB 文件后 从服务器触发可读事件执行 readSyncBulkPayload() 函数，这个函数就会把主服务器发来的数据读到一个缓冲区中，然后将缓冲区的数据写到刚才打开的临时文件中，接着要载入到从服务器的数据库中，最后同步到磁盘中

但是主服务器同步 RDB 文件的逻辑是肿么样的呢，除了 最上面的流程图上，我们还没有具体提到过；现在主要难点在于 主服务器如何触发事件将 RDB 文件发往 从服务器，我们从 时间事件 逻辑上找到了对应逻辑，serverCron()->backgroundSaveDoneHandler()->backgroundSaveDoneHandlerDisk()->updateSlavesWaitingBgsave() 调用完成以后，至此 RDB 文件保存完毕，接下来 调用 sendBulkToSlave() 函数将 RDB 文件写入到 fd 中，触发从服务器的读事件，从服务器调用 readSyncBulkPayload() 函数，来将 RDB文件 的数据载入数据库中，我们看一下 sendBulkToSlave() 都干了啥

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
/* replication.c: line 870 */
void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
    client *slave = privdata;
    UNUSED(el);
    UNUSED(mask);
    char buf[PROTO_IOBUF_LEN];
    ssize_t nwritten, buflen;
    // 发送 RDB 文件 之前先发送 "$<length>\r\n"，表示即将发送的 RDB 文件大小
    if (slave->replpreamble) {
        nwritten = write(fd,slave->replpreamble,sdslen(slave->replpreamble));
        if (nwritten == -1) {
            serverLog(LL_VERBOSE,"Write error sending RDB preamble to slave: %s",
                strerror(errno));
            freeClient(slave);
            return;
        }
        // 更新已经写到网络的字节数
        server.stat_net_output_bytes += nwritten;
        // 保留未写的字节，删除已写的字节
        sdsrange(slave->replpreamble,nwritten,-1);
        // 如果已经写完了，则释放 replpreamble
        if (sdslen(slave->replpreamble) == 0) {
            sdsfree(slave->replpreamble);
            slave->replpreamble = NULL;
        } else {
            return;
        }
    }
    // 将文件指针移动到刚才发送replpreamble的下一个字节，准备写回复
    lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
    // 将 repldbfd 读出 RDB文件 中的内容保存在buf中
    buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN);
    // 保存错误
    if (buflen <= 0) {
        serverLog(LL_WARNING,"Read error sending DB to slave: %s", (buflen == 0) ? "premature EOF" : strerror(errno));
        freeClient(slave);
        return;
    }
    // 将保存 RDB文件 数据的 buf 写到从服务器中
    if ((nwritten = write(fd,buf,buflen)) == -1) {
        if (errno != EAGAIN) {
            serverLog(LL_WARNING,"Write error sending DB to slave: %s",
                strerror(errno));
            freeClient(slave);
        }
        return;
    }
    // 更新从服务器读取主服务器传来的 RDB文件 的字节数
    slave->repldboff += nwritten;
    // 更新服务器已经写到网络的字节数
    server.stat_net_output_bytes += nwritten;
    // 如果写入完成，从网络读到的大小等于文件大小
    if (slave->repldboff == slave->repldbsize) {
        // 关闭 RDB文件 描述符
        close(slave->repldbfd);
        slave->repldbfd = -1;
        // 删除等待从服务器的文件可读事件
        aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
        // 将从服务器置于在线状态
        putSlaveOnline(slave);
    }
}

写完成后，又一次取消监听文件可写事件，等待下一次发送缓冲区数据时再监听触发，并且调用 putSlaveOnline() 函数将从服务器 client 的复制状态设置为 SLAVE_STATE_ONLINE，表示已经发送 RDB文件 完毕，发送缓存更新

从服务器可读以后，触发函数 readSyncBulkPayload() 进行 RDB 文件读取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
/* replication.c: line 1106 */
void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[4096];
    ssize_t nread, readlen;
    off_t left;
    UNUSED(el);
    UNUSED(privdata);
    UNUSED(mask);
    
    // 用于保存EOF标记的静态变量和从服务器接收的最后40个字节：当它们匹配时，表示传输结束
    static char eofmark[CONFIG_RUN_ID_SIZE];
    static char lastbytes[CONFIG_RUN_ID_SIZE];
    static int usemark = 0;
    // 即使 repl_transfer_size 等于 -1，我们仍然从主服务器读取一个长度的回复，该长度是 RDB文件 的大小
    if (server.repl_transfer_size == -1) {
        // 从 fd 中同步读取数据到 buf 中，读取1024个字节
        if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout*1000) == -1) {
            serverLog(LL_WARNING, "I/O error reading bulk count from MASTER: %s", strerror(errno));
            goto error;
        }
        // 读出的数据出错
        if (buf[0] == '-') {
            serverLog(LL_WARNING,
                "MASTER aborted replication with an error: %s",
                buf+1);
            goto error;
        } else if (buf[0] == '\0') {
        // 读出了一个'\0'
            // 只是读出了一个作用和 PING 相同的字符，是为了测试连接是否中断,所以更新最近交互的时间戳，直接返回
            server.repl_transfer_lastio = server.unixtime;
            return;
        } else if (buf[0] != '$') {
        // 读出的长度按照协议格式是 $<length>\r\n，所以第一个字符不是'$'就出错
            serverLog(LL_WARNING,"Bad protocol from MASTER, the first byte is not '$' (we received '%s'), are you sure the host and port are right?", buf);
            goto error;
        }
        
        // 这有两种可能的情况：一种是通常的 $<count> 格式；另一种是用于主服务器事先不知道RDB文件的大小的无盘传输，格式是 $EOF:<40 bytes delimiter>
        // 在文件结尾处，传送了分界符。 分界符长度随机，可以忽略与实际文件内容的冲突概率，如果读出了无盘传输的流格式"EOF:"
        if (strncmp(buf+1,"EOF:",4) == 0 && strlen(buf+5) >= CONFIG_RUN_ID_SIZE) {
            usemark = 1;
            // 将40字节长的分隔符保存到 eofmark 静态数组中
            memcpy(eofmark,buf+5,CONFIG_RUN_ID_SIZE);
            // 将保存从服务器接收的最后40个字节的数组初始化为0
            memset(lastbytes,0,CONFIG_RUN_ID_SIZE);
            // 同步期间从主服务器读到的RDB的大小设置为0，因为是无盘传输的方式
            server.repl_transfer_size = 0;
            serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: receiving streamed RDB from master");
        } else {
        // 从RDB文件读，读出RDB文件长度
            usemark = 0;
            // 从主服务器读到的 RDB文件 的大小
            server.repl_transfer_size = strtol(buf+1,NULL,10);
            serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: receiving %lld bytes from master", (long long) server.repl_transfer_size);
        }
        return;
    }
    // 读数据
    // usemark在无盘传输读取长度是被设置为1
    if (usemark) {
        // 计算数据大小，也就是读的长度
        readlen = sizeof(buf);
    } else {
    // 从RDB文件读
        // 计算读的长度
        left = server.repl_transfer_size - server.repl_transfer_read;
        readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
    }
    // 从 RDB文件 中读到 buf
    nread = read(fd,buf,readlen);
    if (nread <= 0) {
        serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s", (nread == -1) ? strerror(errno) : "connection lost");
        cancelReplicationHandshake();
        return;
    }
    // 更新从网络读的字节数
    server.stat_net_input_bytes += nread;
    // EOF标志是否到达
    int eof_reached = 0;
    if (usemark) {
        // 将所读的 nread 长度的后40字节拷贝到 lastbytes 数组中
        if (nread >= CONFIG_RUN_ID_SIZE) {
            memcpy(lastbytes,buf+nread-CONFIG_RUN_ID_SIZE,CONFIG_RUN_ID_SIZE);
        } else {
        // 读到的不足40字节，那就补足40字节
            int rem = CONFIG_RUN_ID_SIZE-nread;
            memmove(lastbytes,lastbytes+nread,rem);
            memcpy(lastbytes+rem,buf,nread);
        }
        // 比较是否匹配，如果相同表示传输结束，到达EOF结尾
        if (memcmp(lastbytes,eofmark,CONFIG_RUN_ID_SIZE) == 0) eof_reached = 1;
    }
    // 更新最近一次读到RDB文件内容的时间
    server.repl_transfer_lastio = server.unixtime;
    // 将 buf 中的数据写到临时 RDB文件 中，只是写到系统内核的缓冲区中，等待sync
    if (write(server.repl_transfer_fd,buf,nread) != nread) {
        serverLog(LL_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", strerror(errno));
        goto error;
    }
    // 更新同步期间从主服务器读到的RDB文件的总量
    server.repl_transfer_read += nread;
    // 如果是无盘同步，且到达了EOF
    if (usemark && eof_reached) {
        // 删除后40字节
        if (ftruncate(server.repl_transfer_fd,
            server.repl_transfer_read - CONFIG_RUN_ID_SIZE) == -1)
        {
            serverLog(LL_WARNING,"Error truncating the RDB file received from the master for SYNC: %s", strerror(errno));
            goto error;
        }
    }
    // 定期将文件写入磁盘，避免阻塞在IO上，Redis同步缓冲区到磁盘上，并不是每次写到同步，而是当写够8M大小，调用 sync_file_range 函数，一次性的冲刷数据，这样大大提高IO的性能
    if (server.repl_transfer_read >=
        server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
    {
        off_t sync_size = server.repl_transfer_read - server.repl_transfer_last_fsync_off;
        rdb_fsync_range(server.repl_transfer_fd, server.repl_transfer_last_fsync_off, sync_size);
        // 更新最近一个执行fsync的偏移量
        server.repl_transfer_last_fsync_off += sync_size;
    }
    // 从 RDB文件 中读这种类型的同步，是否完成
    if (!usemark) {
        if (server.repl_transfer_read == server.repl_transfer_size)
            eof_reached = 1;
    }
    // 已经完成，则将临时文件该为"dump.rdb"
    if (eof_reached) {
        int aof_is_enabled = server.aof_state != AOF_OFF;
        if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
            serverLog(LL_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
            cancelReplicationHandshake();
            return;
        }
        serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Flushing old data");
        if(aof_is_enabled) stopAppendOnly();
        // 将旧数据库清空
        signalFlushedDb(-1);
        // 调用 replicationEmptyDbCallback 清空所有数据库
        emptyDb(-1, server.repl_slave_lazy_flush ? EMPTYDB_ASYNC : EMPTYDB_NO_FLAGS, replicationEmptyDbCallback);
        // 先删除之前的可读事件，因为载入RDB会设置监听读事件
        aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
        serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
        // 将 RDB文件 载入数据库
        if (rdbLoad(server.rdb_filename,&rsi) != C_OK) {
            serverLog(LL_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
            cancelReplicationHandshake();
            if (aof_is_enabled) restartAOF();
            return;
        }
        // 释放临时文件和临时文件的fd
        zfree(server.repl_transfer_tmpfile);
        close(server.repl_transfer_fd);
        // 为主服务器设置一个client
        replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db);
        server.repl_state = REPL_STATE_CONNECTED;
        server.repl_down_since = 0;
        memcpy(server.replid,server.master->replid,sizeof(server.replid));
        server.master_repl_offset = server.master->reploff;
        clearReplicationId2();
        if (server.repl_backlog == NULL) createReplicationBacklog();
        serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
        // 如果设置了AOF持久化，那么重启AOF持久化功能，并强制生成新数据库的AOF文件
        if (aof_is_enabled) restartAOF();
    }
    return;
// 错误处理，取消复制的握手状态
error:
    cancelReplicationHandshake();
    return;
}

至此全量同步完成 .

命令传播机制

那么以后主服务器有写命令执行，主从的数据又不一致了，那么就需要一个 命令传播机制，传播的时候会通过 propagate() 函数调用 replicationFeedSlaves() ，会将执行的命令以协议的传输格式写到从服务器 client 的输出缓冲区中，这就是为什么主服务器会将从服务器 client 的输出缓冲区发送到从服务器，也会添加到 server.repl_backlog 中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
/* replication.c: line 173 */
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int j, len;
    char llstr[LONG_STR_SIZE];
    // 如果当前就位 从服务器，直接返回
    if (server.masterhost != NULL) return;
    // 如果没有backlog且没有从服务器，直接返回
    if (server.repl_backlog == NULL && listLength(slaves) == 0) return;
    serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL));
    // 如果当前从服务器使用的数据库不是目标的数据库，则要生成一个 select命令
    if (server.slaveseldb != dictid) {
        robj *selectcmd;
        // 0 <= id < 10 ，可以使用共享的select命令对象
        if (dictid >= 0 && dictid < PROTO_SHARED_SELECT_CMDS) {
            selectcmd = shared.select[dictid];
        } else {
        // 否则自行按照协议格式构建select命令对象
        
            int dictid_len;
            dictid_len = ll2string(llstr,sizeof(llstr),dictid);
            selectcmd = createObject(OBJ_STRING,
                sdscatprintf(sdsempty(),
                "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
                dictid_len, llstr));
        }
        // 将select 命令添加到backlog中
        if (server.repl_backlog) feedReplicationBacklogWithObject(selectcmd);
        // 发送给从服务器
        listRewind(slaves,&li);
        // 遍历所有的从服务器
        while((ln = listNext(&li))) {
            client *slave = ln->value;
            // 从服务器状态为等待 BGSAVE 的开始，因此跳过回复，遍历下一个从服务器
            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
            // 添加 select 命令到当前从服务器的回复中
            addReply(slave,selectcmd);
        }
        // 释放临时对象
        if (dictid < 0 || dictid >= PROTO_SHARED_SELECT_CMDS)
            decrRefCount(selectcmd);
    }
    // 设置当前从服务器使用的数据库 ID
    server.slaveseldb = dictid;
    // 将命令写到backlog中
    if (server.repl_backlog) {
        char aux[LONG_STR_SIZE+3];
        // 将参数个数构建成协议标准的字符串
        aux[0] = '*';
        len = ll2string(aux+1,sizeof(aux)-1,argc);
        aux[len+1] = '\r';
        aux[len+2] = '\n';
        // 添加到backlog中
        feedReplicationBacklog(aux,len+3);
        // 遍历所有的参数
        for (j = 0; j < argc; j++) {
            // 返回参数对象的长度
            long objlen = stringObjectLen(argv[j]);
            // 构建成协议标准的字符串，并添加到 backlog 中
            aux[0] = '$';
            len = ll2string(aux+1,sizeof(aux)-1,objlen);
            aux[len+1] = '\r';
            aux[len+2] = '\n';
            feedReplicationBacklog(aux,len+3);
            feedReplicationBacklogWithObject(argv[j]);
            feedReplicationBacklog(aux+len+1,2);
        }
    }
    // 将命令写到每一个从服务器中
    listRewind(slaves,&li);
    // 遍历从服务器链表
    while((ln = listNext(&li))) {
        client *slave = ln->value;
        // 从服务器状态为等待BGSAVE的开始，因此跳过回复，遍历下一个服务器
        if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
        // 将命令写给正在等待初次 SYNC 的从服务器（所以这些命令在输出缓冲区中排队，直到初始SYNC完成），或已经与主服务器同步
        // 添加回复的长度
        addReplyMultiBulkLen(slave,argc);
        // 将所有的参数列表添加到从服务器的输出缓冲区
        for (j = 0; j < argc; j++)
            addReplyBulk(slave,argv[j]);
    }
}

增量同步

之前上次我们也提到了，中途断开的情况，比如 突然断电等，连接上以后，这个时候就可以尝试增量同步，如果可以增量，那么就不用动用全量这个重型操作

部分重同步在复制的过程中，相当于之前发送 PSYNC 命令的部分，其他所有的部分都要进行，他只是主服务器回复从服务器的命令不同，回复 +CONTINUE 则执行部分重同步，回复 +FULLRESYNC 则执行全量同步

那么主服务器如何发现从服务器掉线了呢，那这里就依赖与其他数据库主从一样的 心跳机制，在主从服务器建立连接后，他们之间都维护长连接并彼此发送心跳命令，其实主从服务器彼此都有心跳机制，各自模拟成对方的客户端进行通信；

  • 主服务器心跳机制的频率由 server.repl_ping_slave_period 默认10秒全局变量 控制
  • 从服务每个 1秒 发送 REPLCONF ACK <offset> 信息，像主服务器报告当前复制偏移量

另外，我们之前反复提到了一个 全局变量 server.repl_backlog，代表 复制积压缓冲区；复制积压缓冲区 是一个大小为1M的循环队列，主服务器在命令传播时，不仅会将命令发送给所有的从服务器，还会将命令写入 复制积压缓冲区中，也就是说，复制积压缓冲区 最多可以备份1M大小的数据，如果主从服务器断线时间过长，复制积压缓冲区 的数据会被新数据覆盖，那么当从主从中断连接起，主服务器接收到的数据超过1M大小，那么从服务器就无法进行部分重同步，只能进行全量复制

刚才我们介绍的 syncCommand() 函数中，调用 masterTryPartialResynchronization() 函数会进行尝试部分重同步，在我们之前分析的第一次全量同步时，该函数会执行失败，然后返回 syncCommand()函数执行全量同步，而在进行恢复主从连接后，则会进行部分重同步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
/* replication.c: line 447 */
int masterTryPartialResynchronization(client *c) {
    long long psync_offset, psync_len;
    char *master_replid = c->argv[1]->ptr;
    char buf[128];
    int buflen;
    
    // 从参数对象中获取psync_offset
    if (getLongLongFromObjectOrReply(c,c->argv[2],&psync_offset,NULL) !=
       C_OK) goto need_full_resync;
       
    // 主服务器的 runid 是否和从服务器执行 PSYNC 的参数提供的 runid 相同，如果 runid 发生了改变，则主节点是一个不同的实例，那么就不能进行继续执行原有的复制进程
    if (strcasecmp(master_replid, server.replid) &&
        (strcasecmp(master_replid, server.replid2) ||
         psync_offset > server.second_replid_offset))
    {
        // 如果从节点的 runid 是"?"，表示想要强制进行一个全量同步
        if (master_replid[0] != '?') {
            if (strcasecmp(master_replid, server.replid) &&
                strcasecmp(master_replid, server.replid2))
            {
                serverLog(LL_NOTICE,"Partial resynchronization not accepted: Replication ID mismatch (Slave asked for '%s', my replication IDs are '%s' and '%s')", master_replid, server.replid, server.replid2);
            } else {
                serverLog(LL_NOTICE,"Partial resynchronization not accepted: Requested offset for second ID was %lld, but I can reply up to %lld", psync_offset, server.second_replid_offset);
            }
        } else {
            serverLog(LL_NOTICE,"Full resync requested by slave %s", replicationGetSlaveName(c));
        }
        goto need_full_resync;
    }
    // 如果 psync_offset 小于 repl_backlog_off，说明 backlog 所备份的数据的已经太新了，有一些数据被覆盖，则需要进行全量复制；如果 psync_offset 大于 (server.repl_backlog_off + server.repl_backlog_histlen)，表示当前 backlog 的数据不够全，则需要进行全量复制
    if (!server.repl_backlog ||
        psync_offset < server.repl_backlog_off ||
        psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen))
    {
        serverLog(LL_NOTICE, "Unable to partial resync with slave %s for lack of backlog (Slave request was: %lld).", replicationGetSlaveName(c), psync_offset);
        if (psync_offset > server.master_repl_offset) {
            serverLog(LL_WARNING,
                "Warning: slave %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c));
        }
        goto need_full_resync;
    }
    
    // 执行到这则可以进行部分重同步
    // 设置 client 状态为从节点
    c->flags |= CLIENT_SLAVE;
    // 设置复制状态为在线，此时 RDB文件 传输完成，发送差异数据
    c->replstate = SLAVE_STATE_ONLINE;
    // 设置从服务器收到 ack 的时间
    c->repl_ack_time = server.unixtime;
    // slave 向master 发送 ack 标志设置为0
    c->repl_put_online_on_ack = 0;
    // 将当前 client 加入到从服务器链表中
    listAddNodeTail(server.slaves,c);
    // 向从节点发送 +CONTINUE
    if (c->slave_capa & SLAVE_CAPA_PSYNC2) {
        buflen = snprintf(buf,sizeof(buf),"+CONTINUE %s\r\n", server.replid);
    } else {
        buflen = snprintf(buf,sizeof(buf),"+CONTINUE\r\n");
    }
    if (write(c->fd,buf,buflen) != buflen) {
        freeClientAsync(c);
        return C_OK;
    }
    // 将 backlog 的数据发送从节点
    psync_len = addReplyReplicationBacklog(c,psync_offset);
    serverLog(LL_NOTICE, "Partial resynchronization request from %s accepted. Sending %lld bytes of backlog starting from offset %lld.", replicationGetSlaveName(c), psync_len, psync_offset);
    // 计算延迟值小于 min-slaves-max-lag 的从节点的个数
    refreshGoodSlavesCount();
    return C_OK;
need_full_resync:
    return C_ERR;
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-replication.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！