Redis 复制机制详解 | 运维进阶

twt企业IT社区 2021-11-08

365

Redis复制技术是实现Redis哨兵、集群高可用部署和Redis高扩展性的基石。Redis从实例通过复制主实例使得主从之间数据达到最终一致性，复制过程包括复制初始化、数据同步和命令传播三个阶段。

复制初始化

Redis实例收到replicaof {MASTER_IP} {MASTER_PORT}命令（老版本中使用slaveof命令）后与主实例进行连接并与主实例同步数据。执行replicaof命令后，Redis实例会进行如下操作：

保存主实例信息
主从之间建立socket连接
给主实例发送ping命令，等待主实例回复pong
进行权限验证
使用replconf命令给主实例发送自身的监听端口、ip地址（如果设置了announce-ip）及自身支持的capa等信息
与主实例同步数据
主实例将接到的写命令持续传播至从实例

如果该节点之前有连接的其他主实例，则该命令会执行切主操作：

断开与旧主实例复制关系
与新主实例建立复制关系
删除从实例当前所有数据
对新主实例进行复制操作

数据同步

当从实例首次连接主实例、执行了切主操作或者网络出现闪断后，从实例会与主实例进行数据同步。Redis中数据同步包括全量同步和部分同步两种，涉及到SYNC和PSYNC两个命令。

SYNC命令

在Redis2.8之前，Redis只支持SYNC命令。当收到从实例发送的SYNC命令时，主实例与从实例进行全量同步，其操作流程如下：

初始化：

从实例初次连接或者断线后重连主实例
主、从之间完成鉴权、保存相互监听端口、地址信息等操作

全量复制

从实例发送SYNC命令
主实例fork出子进程执行bgsave生成RDB文件
主实例向从发送RDB文件
主实例在发送RDB文件期间正常处理客户端命令，并将写命令添加到从实例的输出缓存中，在发送完RDB文件后，将该部分缓存命令发送给从实例
从实例清空数据
从实例加载主发送的RDB文件和缓存命令
全量复制完成

PSYNC命令

在Redis2.8之前，主从之间即使短暂的网络闪断也会触发全量复制，全量复制对于Redis来说是一种高IO、高内存占用、高用时的操作。为了解决该问题，在Redis2.8版本中引入了部分复制机制和其相应的命令PSYNC，其命令的格式为psync {RUN_ID} {OFFSET}。Redis使用复制偏移量、复制积压缓冲区和运行id（psync2中改为复制id）三部分来支持该命令。

复制偏移量

主实例和从实例双方都会分别维护一个复制偏移量，用以记载当前实例发送/接受到的命令的长度

主实例每次向从实例传播N个字节的数据就将自己的复制偏移量增加N
从实例每收到主实例传播来的N个数据，就将自己的复制偏移量增加N

从实例每秒钟使用replconf ack {OFFSET}命令向主实例上报自己的复制偏移量，通过对比主从实例的复制偏移量，可以判断主从实例是否处于一致的状态。

复制积压缓冲区

复制积压缓冲区是主实例维护的一个固定长度（参数repl-backlog-size设置，默认为1M）的先进先出队列，当主实例进行命令传播时，不仅会将写命令同步给从实例，同时也会把写命令写入复制积压缓冲区。当从实例重新连接主实例时，主实例通过比较自身和从实例的复制偏移量，判断从实例缺少的数据是否还在复制积压缓冲区中，以此来决定对从实例进行全量同步还是部分同步。

主实例运行ID

主实例运行ID是一个长度为40位的16进制字符串，用于唯一标识一个Redis节点，每次在节点启动均会重新生成。从实例保存主实例的运行id标识自己复制的是哪个主实例，主实例判断psync命令中的run_id是否与自身的相同，若不相同则进行全量复制。

PSYNC2

Redis在4.0版本之前使用psync命令解决了主从短暂掉线后必须进行全量复制的问题。但是在以下两个常见的运维场景仍会发生全量复制：

从实例重启后丢失了以前run_id和复制偏移量
发生了主从切换，主实例的run_id发生了变化

Redis 4.0对PSYNC命令进行了升级，解决在上述两种场景下必然引起全量复制的问题。PYSNC2的命令格式为PSYNC {REPL_ID} {OFFSET}，其中REPL_ID复制ID，用以替代原先PSYNC1命令中的主实例run_id。

针对从实例重启的场景

在Redis关闭保存RDB文件时，会将当前实例的repl_id和offset保存到RDB文件中，并在启动后从RDB文件加载回来。

int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
    ...
    
        if (rdbSaveAuxFieldStrStr(rdb,"repl-id",server.replid)
            == -1) return -1;
        if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",server.master_repl_offset)
            == -1) return -1;
    ...
    return 1;
}

复制

int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
    ...
        else if (type == RDB_OPCODE_AUX) {
            ...
            else if (!strcasecmp(auxkey->ptr,"repl-id")) {
                if (rsi && sdslen(auxval->ptr) == CONFIG_RUN_ID_SIZE) {
                    memcpy(rsi->repl_id,auxval->ptr,CONFIG_RUN_ID_SIZE+1);
                    rsi->repl_id_is_set = 1;
                }
            } else if (!strcasecmp(auxkey->ptr,"repl-offset")) {
                if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
            }
            ...
       }
    ...
}

复制

针对主从切换的场景

每个Redis实例均会保存两个repli_id及它们的复制偏移量，可使用info replication命令查看：

127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:1
slave0:ip=172.16.0.205,port=6379,state=online,offset=392936296,lag=0
master_replid:2ac88837548c56c416f41fd0daf994d97a924dbe
master_replid2:7bf82d2358bcf03c22c1447fc7ee5f75527f8673
master_repl_offset:392936440
second_repl_offset:5156450
...

复制

master_replid和master_repl_offset是当前master的复制id和复制偏移量，master_replid2和second_repl_offset是本实例上一个master的复制id和对应的复制偏移量。在发生主从切换的时候，从实例发送给主实例的的repli_id会与主实例当前保存的master_replid2字段值相同，继而可以进行部分同步。

int masterTryPartialResynchronization(client *c) {
    char *master_replid = c->argv[1]->ptr;
    ...
    if (strcasecmp(master_replid, server.replid) &&
        (strcasecmp(master_replid, server.replid2) ||
         psync_offset > server.second_replid_offset))
    {
        ...
        goto need_full_resync;
    }

    /* We still have the data our slave is asking for? */
    if (!server.repl_backlog ||
        psync_offset < server.repl_backlog_off ||
        psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen))
    {
        ...
        goto need_full_resync;
    }

    ...
    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 (connWrite(c->conn,buf,buflen) != buflen) {
        freeClientAsync(c);
        return C_OK;
    }
    psync_len = addReplyReplicationBacklog(c,psync_offset);
    ...
}

复制