通过上一节的分析我们知道，主从建连之后会一系列握手操作，这里面最关键的一步就是从库向主库发送 PSYNC 命令，其中会携带从库当前的 Replication ID 和 Replication Offset。这里紧接上文，继续介绍从库对 PSYNC 响应的处理。

当主库返回 +CONTINUE 响应的时候，表示进行部分同步，从库会直接进入 REPL_STATE_CONNECTED 状态，主从握手的流程也就结束了，后续会进入正常的主从复制流程。

当主库返回 +FULLRESYNC 响应时，从库就要准备与主库进行全量同步了，下面是从库需要做的准备工作。

  1. 从库首先会创建一个名为 temp-{秒级时间戳}.{进程ID}.rdb 的临时 RDB 文件，然后将这个文件名称以及对应的文件描述符记录到 redisServer.repl_transfer_tmpfile 字段和 repl_transfer_fd 字段中。

  2. 监听主从连接上的可读事件，等待主库发送 RDB 数据，相应的回调为 readSyncBulkPayload() 函数。

  3. 最后，从库会将 redisServer.repl_state 状态切换为 REPL_STATE_TRANSFER，表示从库开始读取主库返回的 RDB 文件。

全量同步

处理完 +FULLRESYNC 响应之后，从库进入了 REPL_STATE_TRANSFER 状态，正如前面说的，这个状态的主库向从库传输 RDB 数据的过程。从库在收到主库传输的 RDB 数据时，会触发 readSyncBulkPayload() 回调函数进行处理，该函数的核心流程如下图所示：

image.png

准备工作

下面我们结合上面的流程图，展开分析一下 readSyncBulkPayload() 回调函数的实现。

首先，从库会阻塞读取主库返回的第一行数据并进行解析，该行数据可能有 $<count> 和 $EOF:<40 bytes> 两种格式。

  • $<count> 中，$ 是固定前缀，count 是后续跟着的 RDB 数据的字节数，该值会被记录到 redisServer.repl_transfer_size 字段中，在从库后续读取的时候，会使用 redisServer.repl_transfer_read 字段记录已读字节数，两者相等则表示 RDB 数据读取完毕。

  • $EOF:<40 bytes> 中，“$EOF:” 这五个字符是个固定前缀，后面紧跟的 40 个字节为一个结束符，该结束符会被记录到 eofmark 这个静态局部变量中，在从库后续每次读取 RDB 数据时，都会使用 lastbytes 静态局部变量来记录读取到的最后 40 个字节，当 lastbytes 与 eofmark 相等时，则表示 RDB 数据读取完毕。

通过第一行返回值确认 RDB 的结束方式之后，从库会根据 repl-diskless-load 配置决定是否进行无磁盘同步，该配置项有下面三个可选值。

  • disabled：禁用无磁盘同步的功能，从库需要将收到的 RDB 先持久化到磁盘，然后再加载到自己的内存中。

  • on-empty-db：在从库中没有任何数据的时候，才使用无磁盘同步的功能。此时，从库不会将 RDB 数据持久化到磁盘，而是直接一边读取 RDB 数据，一边将数据恢复到内存中。

  • swapdb：从库会创建一个全新的、空的 redisDb 实例，来加载 RDB 中的数据，而当前从库正在使用的 redisDb 不会发生变化。如果全量同步失败，从库直接丢弃新 redisDb 实例即可，继续使用原来的 redisDb。如果 RDB 数据加载成功，从库会用新 redisDb 替换原 redisDb，这样就可以保证无磁盘同步失败的时候，从库原有数据不丢失。

    在后面还会分析到一种特殊情况，从库在加载 RDB 同步数据的过程中，是可以解析对外提供服务的，此时从库就是依赖原 redisDb 中的数据没有被覆盖这一特性。注意，因为从库需要额外维护一个 redisDb 实例，所以从库需要有充足的内存，否则会导致 RDB 加载失败。

我们回到 readSyncBulkPayload() 继续分析。如果没有开启无磁盘同步的功能，从库会将主库发来的 RDB 数据写入到前面创建的 RDB 临时文件中，该过程有几个点需要说明一下。

  • 这里使用的是非阻塞读取方式。这个时候从库还没有开始加载 RDB 数据，还是可以继续对外提供服务的，所以不能一直阻塞在这里读取 RDB 数据。

  • 每次读取到数据时，从库会使用 redisServer.repl_transfer_lastio 字段记录当前时间戳。在 serverCron() 中会定时检查该字段，来判断 RDB 传输是否有问题，如果发现长时间阻塞，就会进行终止此次传输，重新建立。

  • 在读取过程中，从库使用 redisServer.repl_transfer_last_fsync_off 字段记录已经刷盘的字节数，通过计算该字段与 redisServer.repl_transfer_read 的差值，可以计算出未刷盘的数据大小，当未刷盘数据超过 8M，从库会执行一次刷盘操作，防止大量数据一起刷盘导致磁盘性能出现尖刺。

如果开启无磁盘加载能力的话，就无需上述持久化流程。

接下来，readSyncBulkPayload() 会是针对 swapdb 策略进行单独处理，这里会为 swapdb 策略创建一组新的 redisDb 实例，这部分逻辑位于 disklessLoadInitTempDb() 函数中，这个函数就是创建多个空的 redisDb，然后组成一个 redisDb 的数组，如下图所示：

image.png

如果没有使用 swapdb 策略，从库会将内存中的数据全部清空掉，这个清空逻辑在 emptyDb() 函数内，其中支持同步和异步两种清空方式，由 replica-lazy-flush 配置项指定，默认为 no，也就是同步模式。如果使用异步方式，会向 lazy free 后台线程提交一个任务，该任务会执行 lazyfreeFreeDatabase() 函数逐个释放 db->dict 和 expires 两个字典中的全部键值对。

加载 RDB 数据

完成上述准备工作之后，从库正式开始使用 RDB 恢复数据。这里会根据是否使用了无磁盘同步模式，进入不同的分支：

image.png

我们先来看没有使用无磁盘同步的处理分支，从库首先要结束掉后台正在执行的 RDB 持久化，因为从库已经从主库那边同步到了最新的数据并生成了临时 RDB 文件，此时的 RDB 持久化已经没有意义了。

然后，从库会执行一次刷盘操作，将从主库那边同步到的 RDB 数据，全部刷到磁盘上的临时 RDB 文件中。随后，将 RDB 临时文件重命名为一个正式的 RDB 文件，文件名由 dbfilename 配置项指定。

接下来，从库就会加载这个最新的 RDB 文件，这过程中会调用 rdbLoadRio() 函数恢复从库数据，加载 RDB 恢复数据的过程，其实就是 RDB 持久化的逆过程，就不再多说了。

在加载结束之后，从库就可以根据配置决定是否删除磁盘上的 RDB 文件，要进行删除需要满足两方面的配置：一个是 rdb-del-sync-files 配置项为 true，另一个是从库没有开启任何持久化操作，也就是 RDB 和 AOF 两种持久化都没有开启。

下面我们再来看无磁盘同步的处理分支。该分支中就无需在磁盘上生成 RDB 文件刷盘了，而是将主从连接转换为阻塞模式，然后调用 rdbLoadRio() 函数从中读取 RDB 数据，恢复从库数据。虽然rdbLoadRio() 函数的核心逻辑基本是 RDB 持久化的逆过程，但是还是有些细节需要特殊说明一下。

  • rdbLoadRio() 函数底层依赖传入的 rio 实例读取 RDB，rio 屏蔽了底层数据来源的不同。例如，无磁盘同步时，使用的是 rioConnIO 实例，也就是将主从连接作为数据源；依赖磁盘文件同步时，使用的是 rioFileIO 实例，也就是将 RDB 文件作为数据源。

  • 在读取 RDB 时，从库会将 rio->update_cksum 指针指向 rdbLoadProgressCallback() 函数，其中除了计算校验和之外，还将定时向主库发送一个换行符，作为心跳消息，让主库知道从库一直在线。

  • 加载 RDB 是一个耗时比较长的操作，此时虽然不一定能执行读命令（从库一般是只读模式），但是有些配置命令、查询元信息命令是可以执行的，所以在 rdbLoadProgressCallback() 函数中还会定时处理收到的客户端命令。这部分逻辑封装在 processEventsWhileBlocked() 函数中，其中连续调用 4 次 aeProcessEvents() 处理读取网络连接上发来的客户端命令，并在执行完成之后返回响应。

小伙伴们这里需要关注一下 redisServer.loading 和 async_loading 两个字段，loading 表示当前从库处于加载 RDB 数据的状态，async_loading 表示的是 loading 状态下的一个特例，它的含义是在 swapdb 模式下，当前从库中的数据与加载的 RDB 数据 Replication ID 一致。也就是，从库与主库是同一个数据集，只不过从库数据集版本落后于主库，所以在 loading 过程中，从库是可以用旧版本的数据继续提供服务的。

在 readSyncBulkPayload() 开始加载 RDB 数据开始之前，会执行 startLoading() 函数，将 loading 以及 async_loading 这两个字段置为 1，在结束 RDB 加载之后，从库会调用 stopLoading() 函数，再将它们重置为 0。下面是 readSyncBulkPayload() 函数中相关的代码片段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
void readSyncBulkPayload(connection *conn) {
    ... // 省略其他逻辑
    if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
        if (memcmp(server.replid, server.master_replid, CONFIG_RUN_ID_SIZE) == 0) {
            //设置async_loading=1需要在swapdb模式下，而且主从的Replication ID一致
            asyncLoading = 1;
        }
        ...
    }
    ...
    startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION, asyncLoading);
    ...// 下面才是前面描述的依赖磁盘或者磁盘的同步流程
}

void startLoading(size_t size, int rdbflags, int async) {
    server.loading = 1;
    if (async == 1) server.async_loading = 1;
    ... // 省略其他逻辑
}

从库如果在 loading 过程中收到了客户端发来的命令，会根据命令特性以及 loading、async_loading 字段决定是否拒绝该命令，像 GET 等读取命令，都是可以在异步加载状态下执行的，这部分检查逻辑的代码片段位于 processCommand() 函数中，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
int processCommand(client *c) {
    ... // 省略分析命令特性的逻辑，命令有ok-loading特新的，例如AUTH命令、CONF命令，
        // 在能在loading状态下执行    
    if (server.loading && !server.async_loading && is_denyloading_command) {
        rejectCommand(c, shared.loadingerr);
        return C_OK;
    }
    // 针对异步加载的检查，像GET等读取命令，都是可以在异步加载状态下执行的
    if (server.async_loading && is_deny_async_loading_command) {
        rejectCommand(c,shared.loadingerr);
        return C_OK;
    }
    ... // 省略其他逻辑
}

善后处理

在加载 RDB 数据顺利完成之后，从库如果使用 swapdb 模式，就可以执行 swapMainDbWithTempDb() 函数，进行 redisDb 替换了，之后，从库就可以用新数据对外提供服务了。如果 RDB 加载过程出现异常，从库会释放掉 RDB 恢复出来的 redisDb，使用原 redisDb 继续对外提供服务。

完成 RDB 加载之后，从库会初始化后续与主库交互的 client 实例，也就是 redisServer.master 这个字段，相关实现位于 replicationCreateMasterClient() 函数，其核心逻辑如下。

  • 将主从连接上可读事件的回调设置为 readQueryFromClient() 函数，之后由这个函数来处理主库发来的日志。

  • 在 redisServer.master 这个 client 的 flags 字段中，添加 CLIENT_MASTER 标记。从库会根据此标记区分 client 是客户端对应的 client 实例，还是主库对应的 client 实例，这样就可以针对主库 client 进行一些特殊处理了。

  • 之前处理 +FULLRESYNC 响应的时候，是将主库返回的 Replication ID 和 Replication Offset 暂存到 redisServer.master_replid 和 server.master_initial_offset 字段中。现在会将这两个值拷贝到 redisServer.master 中的 replid、reploff 字段，以及 redisServer.replid 和 master_repl_offset 字段中，供后续同步使用。

接下来，从库会将 repl_state 状态切换为 REPL_STATE_CONNECTED 状态，整个全量同步流程就结束了。

从库还会初始化自己的主从复制缓冲区，它主要有两个作用：一个是支持 PSYNC2 的优化，因为 PSYNC2 优化中需要从库自己维护一个主从复制缓冲区，这样在它升级成主库的时候，才能为其他从库提供部分复制的能力；另一个作用支持 Redis 的多级从库模式（Sub-Slave 模式），也就是从库下面还可以再挂一个从库，如下图所示。

image.png

上图中的 Slave 相对于 Sub-Slave 来说，就是 Master 相对于 Slave 的角色，所以，主从复制缓冲区的相关介绍，放到后面分析主库视角下主从复制实现的小节中进行介绍。

最后，如果当前 RDB 传输使用的是结束符的方式，也就是主库第一行返回 $EOF:<40 bytes> 的格式，这里会立刻给主库发送一条 REPLCONF ACK {Replication Offset} 命令，通知其 RDB 加载结束，主库会更新从库的相关状态。其实，即使这里不发送 ACK 命令，从库也会在之后周期性执行的 replicationCron() 中，给主库发送自己的 Replication Offset。

CONNECTED 状态

从库进入 CONNECTED 状态之后，就会在主从复制连接上监听主库发来的命令，在监听到可读事件的时候，回调的是 readQueryFromClient() 函数，从而实现部分同步以及正常主从复制的流程。

readQueryFromClient() 函数在第 30 讲《内核解析篇：Redis 读取与请求核心》里面已经详细分析过了，这里就不再重复里面的实现了，只是带小伙伴们简单浏览一下 Redis IO 多线程模式下，从库执行主从复制命令与执行普通客户端命令上的差异。

下面先来回顾一下一条客户端命令执行的流程：在从库收到客户端发来命令的时候，会触发主线程调用 readQueryFromClient() 函数处理可读事件，其中会先给对应 client 添加 CLIENT_PENDING_READ 标记表示延迟读取，并添加到 redisServer.clients_pending_read 队列中。主线程后续会在 beforeSleep() 函数中，将 clients_pending_read 队列中的 client 分配给 IO 线程，由 IO 线程从 client 中读取并解析命令。IO 线程读取完命令之后，会将命令缓存在 client->argv 数组中，同时会在 client 中设置 CLIENT_PENDING_COMMAND 标记，来表示有待执行的命令。最后，主线程在等待全部 IO 线程解析完命令之后，一并执行所有命令。

在主从复制场景中，从库处理主库发来的命令基本与上述处理客户端命令的流程类似，但是有些许差异需要特别说明。

不使用 IO 多线程

从库在处理主库命令的时候，并不会进行 IO 线程处理，而是全部由主线程读取、解析并执行。判断是否延迟读取的逻辑位于 postponeClientRead() 函数中，如下所示，之所以不使用延迟加载是为了减少主从复制的延迟。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
void readQueryFromClient(connection *conn) {
    // 是否由IO线程进行读取和解析
    if (postponeClientRead(c)) return; 
    ... // 省略其他逻辑
}

int postponeClientRead(client *c) {
    if (server.io_threads_active && // 开启多线程模式
        server.io_threads_do_reads && // I/O线程可以执行读取操作
        !ProcessingEventsWhileBlocked && // 处于加载RDB或AOF的流程中
        // 检查client的状态标记位，如果是主库对应的client，也就是redisServer.master这
        // 个client实例，其flags字段中一定包含CLIENT_MASTER标记
        !(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_BLOCKED)) &&
        // 检查io_threads_op全局变量
        io_threads_op == IO_THREADS_OP_IDLE){
         c->flags |= CLIENT_PENDING_READ;
        // 将client追加到clients_pending_read列表中，等待I/O线程去处理
        listAddNodeHead(server.clients_pending_read,c);
        c->pending_read_list_node = listFirst(server.clients_pending_read);
        return 1;
    }else {
        return 0; // 由主线程立即处理
    }
}

Sub-Slave 模式

前面提到 Redis 主从复制是可以组成链状结构的，如下图所示。

image.png

从库在复制主库的同时，还需要将复制来的命令传播到 Sub-Slave 从库，所以从库执行完命令之后，不会立刻释放 client->querybuf 缓冲区的空间，而是将其中已经执行的命令，拷贝到从库的主从复制缓冲区，后续会发送给 Sub-Slave。

这部分逻辑位于 commandProcessed() 函数中，其中涉及到 redisServer.master 这个 client 中的三个关键字段。

  • 第一个是 read_reploff 字段，它用来记录当前从库从主库那里读取了多少个字节的命令，这些命令从库已经执行了一部分，有一部分还未执行。
  • 第二个是 reploff 字段，它用来记录了从库当前已经执行了多少主库发来的命令，单位也是字节，也就是从库的 Replication Offset。
  • 最后一个字段是 repl_applied 字段，它用来记录当前 querybuf 中有多少命令已经被执行，querybuf 缓冲区中 repl_applied 之前的数据，都可以清除掉了。

下面这张示意图，可以清晰地描述这三个字段的关系：

image.png

下面是 commandProcessed() 函数的核心逻辑：

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
int processCommandAndResetClient(client *c) {
    ... // 省略其他准备逻辑
    // 正常执行命令
    if (processCommand(c) == C_OK) { 
        // commandProcessed()函数会在命令正常执行之后，进行一些善后工作
        commandProcessed(c);
    }
    ... // 省略其他逻辑
}

void commandProcessed(client *c) {
    long long prev_offset = c->reploff;
    if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
        // querybuf缓冲区中qb_pos这个位置之前的命令都是已经在processCommand()中
        // 执行的命令
        c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
    }
    ...// 省略其他逻辑
    if (c->flags & CLIENT_MASTER) {
        // 计算此次要发送给SubSlave的命令占了多少字节
        long long pplied = c->reploff - prev_offset; 
        if (applied) { 
            // 如果执行了命令，就会通过replicationFeedSlavesFromMasterStream()将
            // 这些命令写入到复制缓冲区，并发给下游的Sub-Slave
            replicationFeedStreamFromMasterStream(c->querybuf+c->repl_applied,
                applied);
            c->repl_applied += applied;
        }
    }
}

replicationFeedStreamFromMasterStream() 其中做了两件事，一个是将此次执行的主库命令写入到主从复制缓冲区，另一个是开始监听 Sub-Slave 对应的 client，后续发生可写事件的时候，当前从库就会把主从复制缓冲区中对应的命令发到 Sub-Slave，实现 Sub-Slave 这种多级从库的复制效果。这两个逻辑与主库向从库发送命令的逻辑一样，在后续介绍主从复制主库视角下的逻辑时，再展开详细分析。

下面通过一张图简单总结各个缓冲区的数据流转过程，如下所示：

image.png

命令执行限制

在前文介绍全量同步的时候提到过，从库加载 RDB 过程会导致部分命令无法正常执行，我们也介绍了 redisServer.loading 以及 async_loading 的作用。除此之外，从库执行命令时还有其他的一些限制，这里我们补充说明一下。

首先，从库在与主库断开连接之后，从库中的数据可能已经与主库不一致了，从库会根据 replica-serve-stale-data 配置决定是否能继续执行客户端的命令，相关的代码位于 processCommand() 函数中，判断如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
int processCommand(client *c) {
    ... // 省略其他代码
    
    if (server.masterhost // 当前Redis实例是从库
        && server.repl_state != REPL_STATE_CONNECTED  // 主从复制连接已经断开
        && server.repl_serve_stale_data == 0 // 从库是否支持提供旧数据
        && is_denystale_command) // 命令是否支持在旧数据集上执行
    {
        rejectCommand(c, shared.masterdownerr);
        return C_OK;
    }
    ... // 下面才是对loading以及async_loading的检查
}

其次，在绝大多数场景中，为了保证主从一致性，从库只能执行主库发来的数据写命令，从客户端的角度来看，从库就是 Read-Only 的。只有极特殊场景下，我们才希望能够向从库单独写入数据，此时，可以通过 replica-read-only 配置来修改从库的 Read-Only 特性。控制能不能直接向从库写入数据的逻辑在 processCommand() 函数中，相关片段如下：

1
2
3
4
5
6
7
8
9
10
11
int processCommand(client *c) {
    ... // 省略其他代码
    if (server.masterhost  // 当前Redis实例是从库
        && server.repl_slave_ro  // 检查replica-read-only配置
        && !obey_client    // 当前客户端为连接主库的client或者是回放AOF日志的client
        && is_write_command // 命令是否会修改数据
    ){
        rejectCommand(c, shared.roslaveerr);
        return C_OK;
    }
}

总结

在这一节中，我们依旧站在从库的角度来分析 Redis 主从同步的内容。首先，我们介绍了主从握手完成之后，从库进行一次全量同步的核心流程；然后介绍了进入 CONNECTED 正常主从复制状态之后，从库如何同步主库的命令，同时也介绍了从库执行普通客户端命令与执行主库命令的关键区别。

从下一节开始，我们将站在主库的视角，分析 Redis 主从复制的内容，其中还会涉及到 Redis 6.0 提出的无磁盘同步、Redis 7.0 提出的共享缓冲区等一系列优化点的介绍。