Redis学习实践（三） -- 主从复制和集群

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// redis/src/server.h
struct redisServer {
    /* Replication (slave) */
    ...
    // 主库主机名
    char *masterhost;               /* Hostname of master */
    // 主库端口号
    int masterport;                 /* Port of master */
    // 超时时间
    int repl_timeout;               /* Timeout after N seconds of master idle */
    // 从库上用来和主库连接的客户端
    client *master;     /* Client that is master for this slave */
    // 从库上缓存的主库信息
    client *cached_master; /* Cached master to be reused for PSYNC. */
    ...
    // 从库的复制状态机
    int repl_state;          /* Replication status if the instance is a slave */
    ...
};

1
2
3
4
5
6
7
8
9
10
11
12
// redis/src/server.c
struct redisCommand redisCommandTable[] = {
    ...
    {"psync",syncCommand,3,
     "admin no-script",
     0,NULL,0,0,0,0,0,0},
    ...
    {"replicaof",replicaofCommand,3,
     "admin no-script ok-stale",
     0,NULL,0,0,0,0,0,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
// redis/src/replication.c
int slaveTryPartialResynchronization(connection *conn, int read_reply) {
    ...
    // 写部分
    if (!read_reply) {
        server.master_initial_offset = -1;
        if (server.cached_master) {
            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)");
            // 第一次发送psync，命令为：`psync ? -1`
            psync_replid = "?";
            memcpy(psync_offset,"-1",3);
        }
        reply = sendSynchronousCommand(SYNC_CMD_WRITE,conn,"PSYNC",psync_replid,psync_offset,NULL);
        ...
        return PSYNC_WAIT_REPLY;
    }
    // 读部分
    // 接收应答
    reply = sendSynchronousCommand(SYNC_CMD_READ,conn,NULL);
    ...
    // 对应答进行判断处理
    // FULLRESYNC 表示主库响应类型为 全量复制
    if (!strncmp(reply,"+FULLRESYNC",11)) {
        ...
    }
    if (!strncmp(reply,"+CONTINUE",9)) {
        ...
    }
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// redis/src/replication.c
int replicationSetupSlaveForFullResync(client *slave, long long offset) {
    char buf[128];
    int buflen;

    slave->psync_initial_offset = offset;
    slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END;
    server.slaveseldb = -1;
    if (!(slave->flags & CLIENT_PRE_PSYNC)) {
        buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
                          server.replid,offset);
        if (connWrite(slave->conn,buf,buflen) != buflen) {
            freeClientAsync(slave);
            return C_ERR;
        }
    }
    return C_OK;
}

1
2
3
4
5
6
7
8
9
10
11
// redis/src/server.h
struct redisServer {
    ...
    /* Replication (master) */
    char replid[CONFIG_RUN_ID_SIZE+1];  /* My current replication ID. */
    char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
    // 记录当前的最新写操作在 repl_backlog_buffer 中的位置
    long long master_repl_offset;   /* My current replication offset */
    long long second_replid_offset; /* Accept offsets up to this for replid2. */
    ...
};

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
// redis/src/server.c
int main(int argc, char **argv) {
    ...
    // 检查是否是哨兵模式启动
    server.sentinel_mode = checkForSentinelMode(argc,argv);
    // 初始化服务配置，里面会把 redisCommandTable 中定义的命令添加到 server.commands 中
    initServerConfig();
    ...
    if (server.sentinel_mode) {
        // 初始化配置项，默认启动端口是 26379
        // #define REDIS_SENTINEL_PORT 26379
        initSentinelConfig();
        // 哨兵服务初始化
        // 其中会清空所有上述新增的redis命令，并设置 sentinelcmds 里的命令到commands
        initSentinel();
    }
    ...
    // 非哨兵模式
    if (!server.sentinel_mode) {
        serverLog(LL_WARNING,"Server initialized");
        ...
        moduleLoadFromQueue();
        ACLLoadUsersAtStartup();
        // 线程初始化
        InitServerLast();
        // 加载 RDB or AOF 数据到内存
        loadDataFromDisk();
        ...
    } else {
        InitServerLast();
        // 哨兵服务启动
        sentinelIsRunning();
        if (server.supervised_mode == SUPERVISED_SYSTEMD) {
            redisCommunicateSystemd("STATUS=Ready to accept connections\n");
            redisCommunicateSystemd("READY=1\n");
        }
    }
    ...
    redisSetCpuAffinity(server.server_cpulist);
    // 调整oom_score_adj，不让系统OOM
    setOOMScoreAdj(-1);
    // 事件循环
    aeMain(server.el);
    aeDeleteEventLoop(server.el);
    return 0;
}

// 检查是否是以哨兵模式启动
int checkForSentinelMode(int argc, char **argv) {
    int j;

    // 方式1：./redis-sentinel 方式启动
    if (strstr(argv[0],"redis-sentinel") != NULL) return 1;
    for (j = 1; j < argc; j++)
        // 方式2：./redis-server --sentinel 方式启动
        if (!strcmp(argv[j],"--sentinel")) return 1;
    return 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
38
// redis/src/sentinel.c
struct redisCommand sentinelcmds[] = {
    {"ping",pingCommand,1,"",0,NULL,0,0,0,0,0},
    {"sentinel",sentinelCommand,-2,"",0,NULL,0,0,0,0,0},
    {"subscribe",subscribeCommand,-2,"",0,NULL,0,0,0,0,0},
    {"unsubscribe",unsubscribeCommand,-1,"",0,NULL,0,0,0,0,0},
    {"psubscribe",psubscribeCommand,-2,"",0,NULL,0,0,0,0,0},
    {"punsubscribe",punsubscribeCommand,-1,"",0,NULL,0,0,0,0,0},
    {"publish",sentinelPublishCommand,3,"",0,NULL,0,0,0,0,0},
    {"info",sentinelInfoCommand,-1,"",0,NULL,0,0,0,0,0},
    {"role",sentinelRoleCommand,1,"ok-loading",0,NULL,0,0,0,0,0},
    {"client",clientCommand,-2,"read-only no-script",0,NULL,0,0,0,0,0},
    {"shutdown",shutdownCommand,-1,"",0,NULL,0,0,0,0,0},
    {"auth",authCommand,2,"no-auth no-script ok-loading ok-stale fast",0,NULL,0,0,0,0,0},
    {"hello",helloCommand,-2,"no-auth no-script fast",0,NULL,0,0,0,0,0}
};

void initSentinel(void) {
    unsigned int j;

    /* Remove usual Redis commands from the command table, then just add
     * the SENTINEL command. */
    // 清空所有的常规redis命令
    dictEmpty(server.commands,NULL);
    for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) {
        int retval;
        struct redisCommand *cmd = sentinelcmds+j;

        // 添加 sentinelcmds 里的命令到commands
        retval = dictAdd(server.commands, sdsnew(cmd->name), cmd);
        ...
    }

    /* Initialize various data structures. */
    sentinel.current_epoch = 0;
    sentinel.masters = dictCreate(&instancesDictType,NULL);
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// redis/src/sentinel.c
void sentinelIsRunning(void) {
    ...
    // 向每个被监听的主节点发送事件信息
    sentinelGenerateInitialMonitorEvents();
}

void sentinelGenerateInitialMonitorEvents(void) {
    dictIterator *di;
    dictEntry *de;

    di = dictGetIterator(sentinel.masters);
    while((de = dictNext(di)) != NULL) {
        sentinelRedisInstance *ri = dictGetVal(de);
        // 发送事件，通过下面的日志，可看到发送的事件信息：+monitor master mymaster 127.0.0.1 6380 quorum 2
        sentinelEvent(LL_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum);
    }
    dictReleaseIterator(di);
}

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
// redis/src/sentinel.c
void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
    // 1、尝试和断连的实例重新建立连接
    sentinelReconnectInstance(ri);
    // 2、向实例发送 PING、INFO、PUBLISH 等命令
    sentinelSendPeriodicCommands(ri);
    ...
    // 3、检查监听的实例是否主观下线
    sentinelCheckSubjectivelyDown(ri);

    /* Masters and slaves */
    if (ri->flags & (SRI_MASTER|SRI_SLAVE)) {
        /* Nothing so far. */
    }

    /* Only masters */
    // 4、针对监听的主节点，进行客观下线检查判断及故障切换等操作
    if (ri->flags & SRI_MASTER) {
        // 4.1、针对监听的主节点，检查其是否客观下线
        sentinelCheckObjectivelyDown(ri);
        // 4.2、判断是否要启动故障切换
        if (sentinelStartFailoverIfNeeded(ri))
            // 若需要切换，则获取其他哨兵实例对主节点状态的判断，并向其他哨兵发送 is-master-down-by-addr 命令，发起 Leader 选举
            sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED);
        // 4.3、执行故障切换
        sentinelFailoverStateMachine(ri);
        // 4.4、再次获取其他哨兵实例对主节点状态的判断
        sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// redis/src/sentinel.c
typedef struct sentinelRedisInstance {
    // 记录哨兵对主节点主观下线的判断结果
    int flags;      /* See SRI_... defines */
    char *name;     /* Master name from the point of view of this sentinel. */
    char *runid;    /* Run ID of this instance, or unique ID if is a Sentinel.*/
    ...
    // 记录了哨兵和主节点间的两个连接，分别对应用来发送命令的连接 cc 和用来发送 Pub/Sub 消息的连接 pc
    instanceLink *link; /* Link to the instance, may be shared for Sentinels. */
    ...
    // 保存了监听同一主节点的其他哨兵实例
    dict *sentinels;    /* Other sentinels monitoring the same master. */
    // 判断主节点为客观下线需要的哨兵数量
    unsigned int quorum;/* Number of sentinels that need to agree on failure. */
    ...
    // 哨兵对Leader投票的结果，leaderID
    char *leader;
    // 哨兵对Leader投票的结果，leader纪元
    uint64_t leader_epoch; /* Epoch of the 'leader' field. */
    // 记录故障切换的状态
    int failover_state; /* See SENTINEL_FAILOVER_STATE_* defines. */
    ...
} sentinelRedisInstance;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// redis/src/sentinel.c
char *sentinelGetLeader(sentinelRedisInstance *master, uint64_t epoch) {
    ...
    // voters 是所有哨兵的个数，max_votes 是获得的票数
    // 赞成票的数量必须是超过半数以上的哨兵个数
    voters_quorum = voters/2+1;
    // 如果赞成票数不到半数的哨兵个数或者少于quorum阈值，那么Leader就为NULL
    if (winner && (max_votes < voters_quorum || max_votes < master->quorum))
        winner = NULL;

    // 确定最终的Leader
    winner = winner ? sdsnew(winner) : NULL;
    ...
    return winner;
}

1
2
3
4
5
6
7
8
9
10
11
// redis/src/server.h
struct redisServer {
    ...
    /* Cluster */
    int cluster_enabled;      /* Is cluster enabled? */
    mstime_t cluster_node_timeout; /* Cluster node timeout. */
    char *cluster_configfile; /* Cluster auto-generated config file name. */
    // 集群
    struct clusterState *cluster;  /* State of the cluster */
    ...
};

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
// redis/src/cluster.h
#define CLUSTER_SLOTS 16384

// 每个集群节点对应一个 clusterNode
typedef struct clusterNode {
    ...
    // 位图（bitmap），标记该节点负责哪些slots
    unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
    ...
} clusterNode;

// 整个集群结构
typedef struct clusterState {
    ...
    int size;             /* Num of master nodes with at least one slot */
    dict *nodes;          /* Hash table of name -> clusterNode structures */
    dict *nodes_black_list; /* Nodes we don't re-add for a few seconds. */
    // 当前节点负责的 slot 正在迁往哪个节点
    clusterNode *migrating_slots_to[CLUSTER_SLOTS];
    // 当前节点正在从哪个节点迁入某个 slot
    clusterNode *importing_slots_from[CLUSTER_SLOTS];
    // 16384 个 slot 分别是由哪个节点负责的
    clusterNode *slots[CLUSTER_SLOTS];
    // 每个slot存储的key数量
    uint64_t slots_keys_count[CLUSTER_SLOTS];
    // rax字典树，记录 slot 和 key 的对应关系，可以通过它快速找到 slot 上有哪些 keys
    rax *slots_to_keys;
    ...
} clusterState;

1
2
3
4
5
6
7
// redis/src/cluster.c
clusterNode *createClusterNode(char *nodename, int flags);
int clusterAddNode(clusterNode *node);
void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void clusterReadHandler(connection *conn);
void clusterSendPing(clusterLink *link, int type);
...