渐进式解析 Redis 源码 - 哨兵 sentinel

阅读量

刚好紧接上一章 的主从架构，我们这次来分享另外一种 原生高可用方案，就是 sentinel 哨兵 架构

主从架构优势比较明显，可以 保障数据的高可用，从服务器可以扩展主服务器的读能力，有效应对大并发操作

但是，如果主服务器发生故障的时候，需要手动将一个从服务器升级为主服务器，同时需要通知业务方变更配置，并且需要让其它从服务器去复制新主服务器节点，整个过程需要人为干预，比较繁琐；另外还有一个就是，如果没有升级 Redis 到 2.8 以上，则 可能因为 全量同步问题 造成大量资源消耗

所以 Redis 社区推出了一种原生的高可用解决方案，就是 Redis sentinel 架构，由一个或多个 sentinel 实例 组成的 sentinel 系统可以监控任意多个主服务器，以及这些主服务器属下的所有从服务器，并在被监控的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务，实现 Redis 架构的 监控、通知 以及 自动故障转移

由于我们这篇文章主要着重 源码解析，所以关于 Redis sentinel 哨兵架构的部署 诸位可以自行查找一些资料进行实验，我们假定 大家都已经能熟练使用 Redis sentinel 架构

哨兵 sentinel 的使命

常见的 Redis sentinel 架构如下：
Redis sentinel 架构

由上图可见，主要由 sentinel 集群 和 Redis 数据服务器 组成，sentinel 集群对 所有 Redis 数据服务器进行监控，同时 sentinel 集群中 各个哨兵 服务器又相互监控，保证 哨兵监控集群的可用；不过一般 一个稳健的 Redis Sentinel 集群，应该使用至少 三个 Sentinel 实例，并且保证讲这些实例放到 不同的机器 上，甚至不同的 物理区域

Redis sentinel 哨兵的职责主要在于 主服务器存活检测、主从运行状态监控、自动故障转移、主从切换 等，针对 Redis 数据服务器 的功能主要有如下几点：

  • 监控 不断定期检查 主从数据服务器 的存活状况以及运行状态
  • 通知 如果 监控目标服务器出现问题，通过 API 向 管理员 或者 其他应用程序 发送通知
  • 自动故障转移 如果主服务器 下线而不能正常工作的时候，sentinel 会开始执行自动故障转移 操作，从 下线主服务器 下属的 从服务器中 选举出一个作为 主服务器，并且将 其他 从服务器 指向 新的主服务器；不过 下线判断 也分情况，默认 sentinel 会对其监控的 其他 sentinel 节点 和 数据服务器节点 进行每秒一次的 PING 命令检测
    • 主观下线 SDOWN 如果 监控的服务器节点 在 down-after-milliseconds 设置的毫秒时效内没有响应检测，则会被判定为 主观下线；这个状态适用所有服务器节点
    • 客观下线 ODOWN 主服务器节点在发生故障的时候，sentinel 会通过 is-master-down-by-addr 命令向其他 sentinel 节点询问 该主服务器节点的状态，如果超过 quorum 个数的哨兵节点都认为 主服务器节点不可达，则判定为 客观下线；这个状态只会针对 主服务器节点
  • 配置提供者 在 Redis Sentinel 结构中，客户端在初始化的时候连接的是Sentinel 节点集合，从中获取主节点信息

监控通讯

通讯命令
  • sentinel 节点 和 redis 数据节点 之间
    • PING sentinel 向 redis 数据节点发送 PING 命令，检查其状态
    • INFO sentinel 向 redis 数据节点的 主服务器节点发送 INFO 命令，获取 其他从服务器节点信息
    • PUBLISH sentinel 向 redis 数据节点 __sentinel__:hello 频道 发布自己的信息及主服务器节点相关的配置
    • SUBSCRIBE sentinel 通过订阅 redis 主从服务器节点 的 __sentinel__:hello 频道，获取正在监控相同服务的其他 sentinel 节点信息
  • sentinel 节点 和 sentinel 节点 之间
    • PING sentinel 向其他 sentinel节点 发送 PING 命令，检查节点状态
    • is-master-down-by-addr 和其他 sentinel 协商 主服务器节点 的状态，如果 主服务器 处于 SDOWN 状态，则投票自动选出新的 主服务器节点
交互过程

每个 sentinel 以 每秒一次 的频率，向它所知的 主服务器节点 和 从服务器节点 以及其他 sentinel 节点 发送 PING 命令，检测其状态

如果一个 节点 或者 实例 距离 最后一次 有效回复 PING 命令的时间超过 down-after-milliseconds 设定的值，那么这个 节点 或者 实例会被 sentinel 标记为 SDOWN 主观下线

随后正在监控这个 主服务器节点 的所有 sentinel 都要以每秒一次的频率确认这个 主服务器节点 是否真正进入了 主观下线 状态，一旦有 足够数量 的 sentinel 在指定时间范围内同意这个下线判断，那么这个 服务器就要被标记为 ODOWN 客观下线，然后 sentinel 向 客观下线主服务器 的所有 从服务器 发送 INFO 命令的频率，会从 10 秒一次改为 每秒一次

当然，如果 中途当没有足够数量的 sentinel 同意下线判断，那这个 客观下线 状态就会被移除，而且当 主服务器重新对 PING 进行有效响应的时候，那 主观下线状态就会被移除

接上面，如果还是 客观下线状态，就需要进行 主服务器 选举逻辑；slave 的选举主要会评估这几个方面

  • 与master断开连接的次数
  • slave的优先级
  • 数据复制的下标（ 用来评估 slave 当前拥有多少 master 的数据 ）
  • 进程ID

源码解析

由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

启动

sentinel 在源码角度就是一个 特殊模式运行的 Redis 服务器，入口执行的依然是 server.c 的 main() 函数

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
/* server.c: line 3709 */
int main(int argc, char **argv) {
    // ...省略一部分代码
    // 判断是否为 sentinel 模式，一般启动 sentinel模式 主要是通过 redis-sentinel、redis-server --sentinel 两种方式，这里的逻辑就是判断是否这两种方式之一
    server.sentinel_mode = checkForSentinelMode(argc,argv);
    // ...省略一部分代码
    // 如果开启 sentinel 模式，初始化 哨兵配置
    if (server.sentinel_mode) {
        // 初始化节点默认配置
        initSentinelConfig();
        // 初始化节点状态
        initSentinel();
    }
    // ...省略一部分代码
    // 载入配置文件
    loadServerConfig(configfile,options);
    // ...省略一部分代码
    if (!server.sentinel_mode) {
        // 非哨兵模式，涉及 AOF 和 RDB 文件的载入以及 redis cluster 的处理等逻辑
        // ...省略一部分代码
    } else {
        // 启动哨兵模式
        sentinelIsRunning();
    }
    // ...省略一部分代码
}

通过 initSentinelConfig()、initSentinel() 两个函数进行初始化系列处理，不过 initSentinelConfig()函数 实际就是设置初始化 sentinel 端口配置，我们下面着重看一下 initSentinel() 函数

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
/* sentinel.c: line 449 */
void initSentinel(void) {
    unsigned int j;
    // 清空服务器命令表，因为 哨兵模式 并不需要这些 业务命令的处理
    dictEmpty(server.commands,NULL);
    // 循环添加 哨兵模式命令到 命令表中
    for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) {
        int retval;
        struct redisCommand *cmd = sentinelcmds+j;
        retval = dictAdd(server.commands, sdsnew(cmd->name), cmd);
        serverAssert(retval == DICT_OK);
    }
    // 初始化sentinel的各种状态
    // 哨兵的配置版本，后续我们会提到，关于 epoch 这个纪元概念可以查看这篇文章 https://blog.csdn.net/chen_kkw/article/details/82724330
    sentinel.current_epoch = 0;
    // 哨兵监控的 redis 服务器 哈希表
    sentinel.masters = dictCreate(&instancesDictType,NULL);
    // TILT 模式标识，TILT 模式其实是一个 sentinel 特殊保护性的模式；哨兵的运行，其实非常依赖于系统时间，但是当系统时间被调整，或者哨兵中的流程因为某种原因（比如负载较高、IO发生阻塞、进程被信号停止等）而被阻塞时，哨兵的行为就会变得不可预知了。于是就有了 TILT 模式，进入 TILT 模式后，哨兵只定期发送命令用于收集信息，而不采取实质性的动作，比如不会进行故障转移流程。当恢复正常30秒后，哨兵就是退出 TILT 模式
    sentinel.tilt = 0;
    // tilt 模式开始时间
    sentinel.tilt_start_time = 0;
    // 最后执行处理事件
    sentinel.previous_time = mstime();
    // 正在执行的脚本数量
    sentinel.running_scripts = 0;
    // 初始化用户脚本队列
    sentinel.scripts_queue = listCreate();
    // 哨兵通讯的 IP 地址，设置的话就是指定地址，而不是自动发现
    sentinel.announce_ip = NULL;
    // 配合上面，哨兵通讯的 IP 端口
    sentinel.announce_port = 0;
    // 故障模拟
    sentinel.simfailure_flags = SENTINEL_SIMFAILURE_NONE;
    // 是否允许哨兵在运行时修改脚本位置
    sentinel.deny_scripts_reconfig = SENTINEL_DEFAULT_DENY_SCRIPTS_RECONFIG;
    // sentinel 的 ID 设置为 0
    memset(sentinel.myid,0,sizeof(sentinel.myid));
}

刚才上面也发现了关于 配置文件载入的逻辑，调试后发现 sentinel 配置解析 主要调用的还是 sentinelHandleConfiguration()函数

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
/* sentinel.c: line 1576 */
char *sentinelHandleConfiguration(char **argv, int argc) {
    sentinelRedisInstance *ri;
    // 解析 monitor 配置，格式为 monitor <name> <host> <port> <quorum>
    // <name> 代表 监控的主服务器节点别名
    // <host> 代表 监控的主服务器节点 IP 地址
    // <port> 代表 监控的主服务器节点 端口
    // <quorum> 代表 主服务器节点 宕机判定 需要有几个 sentine 节点同意
    if (!strcasecmp(argv[0],"monitor") && argc == 5) {
        int quorum = atoi(argv[4]);
        // 如果 判定节点个数 小于等于0，提示失败
        if (quorum <= 0) return "Quorum must be 1 or greater.";
        // 创建一个主服务器节点实例，并加入到 sentinel 监控的 master 字典中
        if (createSentinelRedisInstance(argv[1],SRI_MASTER,argv[2], atoi(argv[3]),quorum,NULL) == NULL)
        {
            // 处理不同错误信息
            switch(errno) {
            case EBUSY: return "Duplicated master name.";
            case ENOENT: return "Can't resolve master instance hostname.";
            case EINVAL: return "Invalid port number";
            }
        }
    } else if (!strcasecmp(argv[0],"down-after-milliseconds") && argc == 3) {
    // 解析 down-after-milliseconds 参数，每个 sentinel 节点 都会定期向其监控对象 redis 数据节点 和 其他 sentinel 节点 发送 PING 命令，判断状态，如果设定的时间内没有正常响应，则为不可达，这个 时间设定就是 down-after-milliseconds
    // 配置格式为 down-after-milliseconds <name> <milliseconds>
    
        // 根据别名查找主节点实例
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        // 设置主节点判断主观下线的判断时间
        ri->down_after_period = atoi(argv[2]);
        // 如果设定时间 <=0 提示错误
        if (ri->down_after_period <= 0)
            return "negative or zero time parameter.";
        // 根据该 主节点的 down_after_period 字段的值 设置所有连接该主节点的 从节点 和 sentinel实例 的主观下线的判定时间
        sentinelPropagateDownAfterPeriod(ri);
    } else if (!strcasecmp(argv[0],"failover-timeout") && argc == 3) {
    // 解析 failover-timeout 参数，代表 故障转移超时时间设置，当 主节点 客观下线后需要对其进行故障转移，选举新的主节点
    // 配置格式为 failover-timeout <name> <milliseconds>
    
        // 根据别名查找主节点实例
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        // 设置实例属性
        ri->failover_timeout = atoi(argv[2]);
        if (ri->failover_timeout <= 0)
            return "negative or zero time parameter.";
   } else if (!strcasecmp(argv[0],"parallel-syncs") && argc == 3) {
   // 解析 parallel-syncs 参数，代表故障转以后每次向新的主服务器节点发起复制操作的从服务器节点个数；当 sentinel 节点做故障转移操作选举出新的主服务器节点后，原来的从服务器节点会向新的主服务器节点发起复制操作，但是需要进行限制，限制每次向新的主服务器节点发起复制操作的从服务器节点个数 就是当前配置
   // 配置格式为 parallel-syncs <name> <nums>
   
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        ri->parallel_syncs = atoi(argv[2]);
   } else if (!strcasecmp(argv[0],"notification-script") && argc == 3) {
   // 解析 notification-script 参数，代表 警告触发脚本；当故障发生的时候，比如 主客观下线，会触发设置的 警告脚本，向脚本发送响应的事件参数
   // 配置格式为 notification-script <name> <path>
   
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        if (access(argv[2],X_OK) == -1)
            return "Notification script seems non existing or non executable.";
        ri->notification_script = sdsnew(argv[2]);
   } else if (!strcasecmp(argv[0],"client-reconfig-script") && argc == 3) {
   // 解析 client-reconfig-script 参数，代表 故障转移成功以后触发的脚本，通知相关的客户端关于转移后的信息
   // 配置格式为 client-reconfig-script <name> <path>
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        if (access(argv[2],X_OK) == -1)
            return "Client reconfiguration script seems non existing or non executable.";
        ri->client_reconfig_script = sdsnew(argv[2]);
   } else if (!strcasecmp(argv[0],"auth-pass") && argc == 3) {
   // 解析 auth-pass 参数，代表监控的主服务器节点的密码验证信息
   // 配置格式为 auth-pass <name> <password>
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        ri->auth_pass = sdsnew(argv[2]);
    } else if (!strcasecmp(argv[0],"current-epoch") && argc == 2) {
    // 解析 current-epoch 参数，代表当前 sentinel 当前的纪元，也就是 版本
    // 配置格式为 current-epoch <epoch>
    
        unsigned long long current_epoch = strtoull(argv[1],NULL,10);
        // 因为后续存在 配置更新的情况，如果传递的 纪元数 大于 原本 sentinel 的记录数，则更新
        if (current_epoch > sentinel.current_epoch)
            sentinel.current_epoch = current_epoch;
    } else if (!strcasecmp(argv[0],"myid") && argc == 2) {
    // 解析 myid 参数，代表当前 sentinel 的 id
        if (strlen(argv[1]) != CONFIG_RUN_ID_SIZE)
            return "Malformed Sentinel id in myid option.";
        memcpy(sentinel.myid,argv[1],CONFIG_RUN_ID_SIZE);
    } else if (!strcasecmp(argv[0],"config-epoch") && argc == 3) {
    // 解析 config-epoch，代表 配置纪元
    // 配置格式为 config-epoch <name> <epoch>
    
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        ri->config_epoch = strtoull(argv[2],NULL,10);
        // 如果配置纪元大于当前纪元，则更新当前纪元
        if (ri->config_epoch > sentinel.current_epoch)
            sentinel.current_epoch = ri->config_epoch;
    } else if (!strcasecmp(argv[0],"leader-epoch") && argc == 3) {
    // 解析 leader-epoch 参数，代表 局部 sentinel 领导节点的 纪元数
    // 配置格式为 leader-epoch <name> <epoch>
    
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        ri->leader_epoch = strtoull(argv[2],NULL,10);
    } else if (!strcasecmp(argv[0],"known-slave") && argc == 4) {
    // 解析 known-slave 参数，代表发现的从服务器节点的 ip 和 port
    // 配置格式为 known-slave <name> <ip> <port>
    
        sentinelRedisInstance *slave;
        ri = sentinelGetMasterByName(argv[1]);
        if (!ri) return "No such master with specified name.";
        // 创建发现的 从服务器节点实例，并加入到监控字典中
        if ((slave = createSentinelRedisInstance(NULL,SRI_SLAVE,argv[2], atoi(argv[3]), ri->quorum, ri)) == NULL)
        {
            return "Wrong hostname or port for slave.";
        }
    } else if (!strcasecmp(argv[0],"known-sentinel") &&
               (argc == 4 || argc == 5)) {
    // 解析 known-sentinel 参数，代表发现的其他 sentinel 节点的信息
    
        sentinelRedisInstance *si;
        // 因为发现 sentinel 都会有 runid 配置，如果没有 runid 配置就忽略掉，格式为 known-sentinel <name> <ip> <port> [runid]
        if (argc == 5) {
            ri = sentinelGetMasterByName(argv[1]);
            if (!ri) return "No such master with specified name.";
            // 创建一个 sentinel 实例，并加入到 主 sentinel 节点的监控字典中
            if ((si = createSentinelRedisInstance(argv[4],SRI_SENTINEL,argv[2],
                        atoi(argv[3]), ri->quorum, ri)) == NULL)
            {
                return "Wrong hostname or port for sentinel.";
            }
            // 设置 sentinel 实例的 runid
            si->runid = sdsnew(argv[4]);
            // 尝试与其他 sentinel 节点共享监控对象
            sentinelTryConnectionSharing(si);
        }
    } else if (!strcasecmp(argv[0],"announce-ip") && argc == 2) {
    // 解析 announce-ip 参数，代表 哨兵通讯的 IP 地址，设置的话就是指定地址，而不是自动发现
    // 配置格式为 announce-ip <ip-address>
        if (strlen(argv[1]))
            sentinel.announce_ip = sdsnew(argv[1]);
    } else if (!strcasecmp(argv[0],"announce-port") && argc == 2) {
    // 解析 announce-port 参数，跟上面的 announce-ip 配合
    // 配置格式为 announce-port <port>
        sentinel.announce_port = atoi(argv[1]);
    } else if (!strcasecmp(argv[0],"deny-scripts-reconfig") && argc == 2) {
    // 解析 deny-scripts-reconfig，代表 是否允许哨兵在运行时修改脚本位置
    // 配置格式为 deny-scripts-reconfig <yes|no>
        if ((sentinel.deny_scripts_reconfig = yesnotoi(argv[1])) == -1) {
            return "Please specify yes or no for the "
                   "deny-scripts-reconfig options.";
        }
    } else {
        return "Unrecognized sentinel configuration statement.";
    }
    return NULL;
}

上面我们经常会提到一个创建实例函数 createSentinelRedisInstance()，其中有会传递用来表示实例类型的参数 flags ( SRI_MASTER | SRI_SLAVE | SRI_SENTINEL )；实际这个函数主要就是初始化这个创建的实例，那为什么我们还要去解析呢？主要是为了让大家熟悉实例属性

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
/* sentinel.c: line 1149 */
// 各个参数我们就不进行详细介绍了，因为多多少少我们之前都进行过讲解
sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char *hostname, int port, int quorum, sentinelRedisInstance *master) {
    sentinelRedisInstance *ri;
    sentinelAddr *addr;
    dict *table = NULL;
    char slavename[NET_PEER_ID_LEN], *sdsname;
    // 标识类型一定为 主节点、从节点、sentinel节点 其中之一
    serverAssert(flags & (SRI_MASTER|SRI_SLAVE|SRI_SENTINEL));
    // 如果实例不是 主节点类型，那么就是 从节点 或者 sentinel 节点，这样的话 传递的 主节点一定不为空
    serverAssert((flags & SRI_MASTER) || master != NULL);
    // 创建地址对象，这里会对 主机信息进行验证
    addr = createSentinelAddr(hostname,port);
    if (addr == NULL) return NULL;
    // 如果为从节点，那么命名为 IP:PORT 组合的字符串
    if (flags & SRI_SLAVE) {
        anetFormatAddr(slavename, sizeof(slavename), hostname, port);
        name = slavename;
    }
    // 由于相同的 主节点 name、相同 IP:PORT 组合的从节点、sentinel节点 可能已经存在或者被重复添加，所以一般要先获取对应类型的 字典表，从中根据 name 进行查询，是否已经存在
    // 获取主节点字典表
    if (flags & SRI_MASTER) table = sentinel.masters;
    // 获取从节点字典表
    else if (flags & SRI_SLAVE) table = master->slaves;
    // 获取sentinel节点字典表
    else if (flags & SRI_SENTINEL) table = master->sentinels;
    sdsname = sdsnew(name);
    // 根据 name 查找是否已经存在，如果已经存在就先释放刚才创建的地址对象后直接返回
    if (dictFind(table,sdsname)) {
        releaseSentinelAddr(addr);
        sdsfree(sdsname);
        errno = EBUSY;
        return NULL;
    }
    // 创建实例对象
    ri = zmalloc(sizeof(*ri));
    // 注意所有实例开始都是网络断开状态，事件循环会为他们一一建立起网络连接，下面开始给实例添加各种属性
    // 实例类型标识
    ri->flags = flags;
    // 实例 name
    ri->name = sdsname;
    // 初始 runid
    ri->runid = NULL;
    // 配置纪元
    ri->config_epoch = 0;
    // 地址对象
    ri->addr = addr;
    // 先创建一个没有网络连接的连接对象，方便后续使用
    ri->link = createInstanceLink();
    // 最后一次发送 pub 信息的时间
    ri->last_pub_time = mstime();
    // 最后一次发送 hello 的时间
    ri->last_hello_time = mstime();
    // 最近一次回复下线状态的时间
    ri->last_master_down_reply_time = mstime();
    // 被判为主观下线的时间
    ri->s_down_since_time = 0;
    // 被判为客观下线的时间
    ri->o_down_since_time = 0;
    // 实例没有响应多少毫秒后被判断为主观下线
    ri->down_after_period = master ? master->down_after_period : SENTINEL_DEFAULT_DOWN_AFTER;
    // 从节点复制操作断开时间
    ri->master_link_down_time = 0;
    // 认证密码
    ri->auth_pass = NULL;
    // INFO 命令输出的 从节点优先权
    ri->slave_priority = SENTINEL_DEFAULT_SLAVE_PRIORITY;
    // 故障转移的时候，从节点发送 slaveof <new> 命令的时间
    ri->slave_reconf_sent_time = 0;
    // 从节点的主节点地址
    ri->slave_master_host = NULL;
    // 从节点的主节点端口
    ri->slave_master_port = 0;
    // 从节点当前的主节点连接状态
    ri->slave_master_link_status = SENTINEL_MASTER_LINK_STATUS_DOWN;
    // 从节点复制偏移量
    ri->slave_repl_offset = 0;
    // 建立当前 主节点对应的 sentinel节点字典表
    ri->sentinels = dictCreate(&instancesDictType,NULL);
    // 判定主节点客观下线的投票数
    ri->quorum = quorum;
    // 故障转移时，可以同时对新主节点进行同步的从节点个数
    ri->parallel_syncs = SENTINEL_DEFAULT_PARALLEL_SYNCS;
    // 所属主节点
    ri->master = master;
    // 下属从节点字典表
    ri->slaves = dictCreate(&instancesDictType,NULL);
    // 获取 INFO 信息的时间
    ri->info_refresh = 0;
    
    // 故障转移信息
    // 如果当前是 主节点，那 leader 是执行故障转移的 sentinel 的 runid，如果当前是 sentinel 节点，那 leader 是当前这个 sentinel 选举出来的领头runid
    ri->leader = NULL;
    // leader 的纪元，也就是版本
    ri->leader_epoch = 0;
    // 当前执行故障转移的 纪元
    ri->failover_epoch = 0;
    // 故障转移状态标识
    ri->failover_state = SENTINEL_FAILOVER_STATE_NONE;
    // 故障转移状态改变的时间
    ri->failover_state_change_time = 0;
    // 故障转移开始时间
    ri->failover_start_time = 0;
    // 故障转移最大超时时间
    ri->failover_timeout = SENTINEL_DEFAULT_FAILOVER_TIMEOUT;
    // 故障转移延迟时间
    ri->failover_delay_logged = 0;
    // 晋升为新主节点的从节点实例
    ri->promoted_slave = NULL;
    // 警告触发脚本
    ri->notification_script = NULL;
    // 故障转移成功以后触发的脚本
    ri->client_reconfig_script = NULL;
    // 缓存 INFO命令的输出
    ri->info = NULL;
    // 角色信息
    // 实例角色
    ri->role_reported = ri->flags & (SRI_MASTER|SRI_SLAVE);
    // 角色变更时间
    ri->role_reported_time = mstime();
    // 最近一次从节点配置变更时间
    ri->slave_conf_change_time = mstime();
    // 加入字典
    dictAdd(table, ri->name, ri);
    return ri;
}

紧接着就是 sentinel 哨兵模式的启动了，从最上面 main() 函数中我们看到，调用了 sentinelIsRunning() 函数进行启动处理的

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
/* sentinel.c: line 480 */
void sentinelIsRunning(void) {
    int j;
    // 要求必须有配置文件
    if (server.configfile == NULL) {
        serverLog(LL_WARNING,
            "Sentinel started without a config file. Exiting...");
        exit(1);
    } else if (access(server.configfile,W_OK) == -1) {
    // 判断配置文件是否可写，因为涉及到 配置文件重写的问题
        serverLog(LL_WARNING,
            "Sentinel config file %s is not writable: %s. Exiting...", server.configfile,strerror(errno));
        exit(1);
    }
    // 如果 sentinel 在配置文件中没有添加 ID，则随机生成一个保存到文件中
    for (j = 0; j < CONFIG_RUN_ID_SIZE; j++)
        if (sentinel.myid[j] != 0) break;
        
    // 如果没有ID
    if (j == CONFIG_RUN_ID_SIZE) {
        // 通过 sha1 算法随机生成一个ID
        getRandomHexChars(sentinel.myid,CONFIG_RUN_ID_SIZE);
        // 保存到配置文件中
        sentinelFlushConfig();
    }
    serverLog(LL_WARNING,"Sentinel ID is %s", sentinel.myid);
    // sentinel 启动时，生成一个 +monitor 事件
    sentinelGenerateInitialMonitorEvents();
}

事件

哨兵的操作是由时间事件进行操作的，也就是由 serverCron() 函数发起的

1
2
3
4
5
/* server.c: line 1152 */
run_with_period(100) {
    if (server.sentinel_mode) sentinelTimer();
}

以 100毫秒的 频率执行 sentinelTimer() 事件，对 sentinel 各种事件进行周期性调度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/* sentinel.c: line 4357 */
void sentinelTimer(void) {
    // 先检查 sentinel 是否需要进入 TILT 模式，更新最近一次执行 sentinel 模式的时间
    sentinelCheckTiltCondition();
    // 对 sentinel 监控的所有主节点进行递归执行周期函数
    sentinelHandleDictOfRedisInstances(sentinel.masters);
    // 运行队列中等待的脚本，比如发送 PING、HELLO 等
    sentinelRunPendingScripts();
    // 清理已经执行的脚本，重新执行错误的脚本
    sentinelCollectTerminatedScripts();
    // 清除执行超时的脚本，等下一次周期在 sentinelCollectTerminatedScripts() 中执行
    sentinelKillTimedoutScripts();
    
    // 不断改变 Redis 定期任务的执行频率，以便使每个 sentinel 节点都不同步，这种不确定性可以避免 sentinel 在同一时间开始完全继续保持同步，当被要求进行投票时，一次又一次在同一时间进行投票，因为脑裂导致有可能没有胜选者
    server.hz = CONFIG_DEFAULT_HZ + rand() % CONFIG_DEFAULT_HZ;
}

在讲解 TILT 模式判断逻辑开始前，我们来回顾一下上面已经说到过的 TILT 模式，TILT 模式其实是一个 sentinel 特殊保护性的模式；哨兵的运行，其实非常依赖于系统时间，但是当系统时间被调整，或者哨兵中的流程因为某种原因（比如负载较高、IO发生阻塞、进程被信号停止等）而被阻塞时，哨兵的行为就会变得不可预知了。于是就有了 TILT 模式，进入 TILT 模式后，哨兵只定期发送命令用于收集信息，而不采取实质性的动作，比如不会进行故障转移流程；当有实例向 sentinel 发送 SENTINEL is-master-down-by-addr 的时候会被返回负值，因为当前情况的下线判断已经不准确了。当恢复正常30秒后，哨兵就是退出 TILT 模式

通过上面的时间事件我们可以看到，每次执行相隔 100毫秒，相当于每秒 10次，当然如果发生了异常（负载问题、IO 阻塞、进程信号停止等）或者 系统时钟发生了明显变化 等情况，sentinel 会上一周期执行的时间，与这次的时间进行对比，我们看一下具体逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/* sentinel.c: line 4345 */
void sentinelCheckTiltCondition(void) {
    mstime_t now = mstime();
    // 最近一次执行周期函数过了多久
    mstime_t delta = now - sentinel.previous_time;
    // 如果时间为 负值 或者相隔时间时间超过 2000毫秒，则进入 TILT 模式
    if (delta < 0 || delta > SENTINEL_TILT_TRIGGER) {
        // 进入 TILT 模式
        sentinel.tilt = 1;
        // TILT 模式开始时间
        sentinel.tilt_start_time = mstime();
        sentinelEvent(LL_WARNING,"+tilt",NULL,"#tilt mode entered");
    }
    // 设置最近一次 sentinel 时间处理程序的时间
    sentinel.previous_time = mstime();
}

那么接下来我们在执行周期性函数sentinelHandleDictOfRedisInstances()

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
/* sentinel.c: line 4302 */
void sentinelHandleDictOfRedisInstances(dict *instances) {
    dictIterator *di;
    dictEntry *de;
    sentinelRedisInstance *switch_to_promoted = NULL;
    // 获取监控对象实例迭代器
    di = dictGetIterator(instances);
    // 遍历监控对象实例
    while((de = dictNext(di)) != NULL) {
        // 获取实例
        sentinelRedisInstance *ri = dictGetVal(de);
        // 对指定的实例执行周期操作
        sentinelHandleRedisInstance(ri);
        // 如果当前实例为主服务器节点
        if (ri->flags & SRI_MASTER) {
            // 递归对当前实例的从节点执行周期函数
            sentinelHandleDictOfRedisInstances(ri->slaves);
            // 递归对当前实例的 sentinel 节点执行周期函数
            sentinelHandleDictOfRedisInstances(ri->sentinels);
            // 如果当前实例处于完成故障转移状态而且所有从节点已经完成了对新主节点的同步操作，那么设置 主从转换标识
            if (ri->failover_state == SENTINEL_FAILOVER_STATE_UPDATE_CONFIG) {
                switch_to_promoted = ri;
            }
        }
    }
    // 如果设置了主从转换标识
    if (switch_to_promoted)
        // 将原来的主节点从表中删除，使用晋升的新主节点替代，那么所有 从节点 和 旧主节点 都从属当前新的主节点
        sentinelFailoverSwitchToPromotedSlave(switch_to_promoted);
    // 释放迭代器
    dictReleaseIterator(di);
}

那么我们继续调试，查看周期性操作函数 sentinelHandleRedisInstance()

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
/* sentinel.c: line 4266 */
void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
    /* ========== 监控操作部分 ============ */
    // 对所有类型的实例进行操作
    // 为所有 sentinel 和 实例之间建立网络通信
    sentinelReconnectInstance(ri);
    // 定期发送 PING、PONG、PUBLISH 等命令
    sentinelSendPeriodicCommands(ri);
    /* ============== 故障检测部分 ============= */
    // 如果处于 TILT 模式，不进行故障检测
    if (sentinel.tilt) {
        // 如果 TILT 模式的时间还没到（默认 1000*30毫秒），则直接返回
        if (mstime()-sentinel.tilt_start_time < SENTINEL_TILT_PERIOD) return;
        // 超过时间，则退出 TILT 模式
        sentinel.tilt = 0;
        sentinelEvent(LL_WARNING,"-tilt",NULL,"#tilt mode exited");
    }
    // 对于各实例进行下线检测，判断是否处于主观下线状态
    sentinelCheckSubjectivelyDown(ri);
    // 主从节点不做什么操作，不过这里不太明白，也没 return，貌似像是没写完
    if (ri->flags & (SRI_MASTER|SRI_SLAVE)) {
        /* Nothing so far. */
    }
    // 主服务器节点
    if (ri->flags & SRI_MASTER) {
        // 检查是否客观下线
        sentinelCheckObjectivelyDown(ri);
        // 如果处于客观下线，则进一步进行故障转移状态判断
        if (sentinelStartFailoverIfNeeded(ri))
            // 向其他 sentinel 节点发送 SENTINEL is-master-down-by-addr 确认是否不可达，如果能从其他 sentinel 节点足够多的票数，则标记为客观下线，触发故障转移
            sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED);
        // 执行故障转移操作
        sentinelFailoverStateMachine(ri);
        // 主节点没有处于客观下线的状态，那么也要尝试发送 SENTINEL is-master-down-by-addr 给所有的 sentinel 获取回复，因为主节点如果有回复延迟等等状况，可以通过该命令，更新一些主节点状态
        sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS);
    }
}

由于在载入配置的时候，主节点实例加入到字典表的时候默认是关闭的，所以一开始上面就调用了个 sentinel 和 实例的连接函数 sentinelReconnectInstance()

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
/* sentinel.c: line 1919 */
void sentinelReconnectInstance(sentinelRedisInstance *ri) {
    // 如果实例没有断开连接，直接返回
    if (ri->link->disconnected == 0) return;
    // 如果 port == 0，则意味着 地址非法
    if (ri->addr->port == 0) return;
    instanceLink *link = ri->link;
    mstime_t now = mstime();
    // 如果最近一次重连的时间距离现在太短，小于 1秒，则直接返回
    if (now - ri->link->last_reconn_time < SENTINEL_PING_PERIOD) return;
    ri->link->last_reconn_time = now;
    // 命令模式，发送命令，跟下面的订阅连接区分
    if (link->cc == NULL) {
        // 建立连接并绑定到实例
        link->cc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
        // 连接错误，发送事件通知，并关闭连接
        if (link->cc->err) {
            sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #%s",
                link->cc->errstr);
            instanceLinkCloseConnection(link,link->cc);
        } else {
            // 重置连接属性
            link->pending_commands = 0;
            link->cc_conn_time = mstime();
            link->cc->data = link;
            // 将服务器的事件循环关联到连接的上下文中
            redisAeAttach(server.el,link->cc);
            // 设置确立连接的回调函数
            redisAsyncSetConnectCallback(link->cc, sentinelLinkEstablishedCallback);
            // 设置断开连接的回调函数
            redisAsyncSetDisconnectCallback(link->cc, sentinelDisconnectCallback);
            // 发送 AUTH 命令认证
            sentinelSendAuthIfNeeded(ri,link->cc);
            // 发送连接名字
            sentinelSetClientName(ri,link->cc,"cmd");
            // 立即发送 PING 命令
            sentinelSendPing(ri);
        }
    }
    // 订阅模式
    // 针对 主从节点建立 订阅连接
    if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && link->pc == NULL) {
        // 建立连接并绑定到实例
        link->pc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
        // 连接错误，发送事件通知，并关闭连接
        if (link->pc->err) {
            sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #%s",
                link->pc->errstr);
            instanceLinkCloseConnection(link,link->pc);
        } else {
            int retval;
            // 重置连接属性
            link->pc_conn_time = mstime();
            link->pc->data = link;
            // 将服务器的事件循环关联到连接的上下文中
            redisAeAttach(server.el,link->pc);
            // 设置确立连接的回调函数
            redisAsyncSetConnectCallback(link->pc, sentinelLinkEstablishedCallback);
            // 设置断开连接的回调函数
            redisAsyncSetDisconnectCallback(link->pc, sentinelDisconnectCallback);
            // 发送 AUTH 命令认证
            sentinelSendAuthIfNeeded(ri,link->pc);
            // 发送连接名字
            sentinelSetClientName(ri,link->pc,"pubsub");
            // 发送订阅 __sentinel__:hello 频道的命令，设置回调函数sentinelReceiveHelloMessages 处理回复，sentinelReceiveHelloMessages 是处理 pub/sub 的频道返回信息的回调函数，可以发现订阅同一 master 的 sentinel 节点
            retval = redisAsyncCommand(link->pc, sentinelReceiveHelloMessages, ri, "SUBSCRIBE %s", SENTINEL_HELLO_CHANNEL);
            // 订阅频道出错
            if (retval != C_OK) {
                // 关闭连接
                instanceLinkCloseConnection(link,link->pc);
                return;
            }
        }
    }
    // 如果已经建立了新的连接，则清除断开连接的状态，表示已经建立连接
    if (link->cc && (ri->flags & SRI_SENTINEL || link->pc))
        link->disconnected = 0;
}

命令发送

接下来我们来解析 sentinelSendPeriodicCommands() 发送监控命令函数

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
/* sentinel.c: line 2577 */
void sentinelSendPeriodicCommands(sentinelRedisInstance *ri) {
    mstime_t now = mstime();
    mstime_t info_period, ping_period;
    int retval;
    
    // 如果实例处于关闭状态，直接返回
    if (ri->link->disconnected) return;
    // 对于不是 INFO、PING、PUBLISH 等关键命令的时候，会有 SENTINEL_MAX_PENDING_COMMANDS 的限制，这样的设计是为了冗余保护措施，如果检测到超时条件，连接将被断开重新连接，每个实例已经发送未回复的命令不能超过 SENTINEL_MAX_PENDING_COMMANDS（默认100），否则直接返回
    if (ri->link->pending_commands >=
        SENTINEL_MAX_PENDING_COMMANDS * ri->link->refcount) return;
    
    // 如果主服务器节点处于了 O_DOWN 状态，那么 sentinel 由 SENTINEL_INFO_PERIOD 10秒 频率改为 默认每秒发送 INFO 命令给他的从服务器节点，因为我们需要更加密切的监控从服务器节点；如果从服务器节点和主服务器节点断开连接，我们同样也会更加频繁的监控 INFO 命令的输出，因为我们需要监控最新的断开连接时间
    
    // 如果实例为从服务器节点，而且主服务器节点故障或者主从断开复制状态
    if ((ri->flags & SRI_SLAVE) &&
        ((ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS)) ||
         (ri->master_link_down_time != 0)))
    {
        // 监控频率修改为 1秒，即上面说的每秒
        info_period = 1000;
    } else {
        // 否则默认的 10秒
        info_period = SENTINEL_INFO_PERIOD;
    }
    
    // 如果最后一次接受到 PONG 的时候比配置的 down-after-milliseconds 时间更长，并且如果 down-after-milliseconds 设置大于 1秒，则直接按照 1秒的频率 PING
    // 主观下线判断时间
    ping_period = ri->down_after_period;
    // 最小1秒的频率
    if (ping_period > SENTINEL_PING_PERIOD) ping_period = SENTINEL_PING_PERIOD;
    // 发送 INFO 命令给 主从节点
    if ((ri->flags & SRI_SENTINEL) == 0 &&
        (ri->info_refresh == 0 ||
        (now - ri->info_refresh) > info_period))
    {
        // 发送 INFO 命令
        retval = redisAsyncCommand(ri->link->cc,
            sentinelInfoReplyCallback, ri, "INFO");
        // 已经发送未回复的命令数 +1
        if (retval == C_OK) ri->link->pending_commands++;
    }
    // 如果发送和回复 PING 命令超时
    if ((now - ri->link->last_pong_time) > ping_period && (now - ri->link->last_ping_time) > ping_period/2) {
        // 立即发送一个 PING 命令，并更新 act_ping_time
        sentinelSendPing(ri);
    }
    
    // 如果发送 PUBLISH 的命令超时
    if ((now - ri->last_pub_time) > SENTINEL_PUBLISH_PERIOD) {
        // 立即发送 PUBLISH 信息给实例
        sentinelSendHello(ri);
    }
}

接上面函数里面的逻辑，INFO 命令的回调函数为 sentinelInfoReplyCallback()，进行一些简短的正确性判断和统计后，实际运算内容由 sentinelRefreshInstanceInfo() 函数 收集响应消息和处理角色变化，我们来看一下具体的逻辑

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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
/* sentinel.c: line 2004 */
void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) {
    sds *lines;
    int numlines, j;
    int role = 0;
    // 释放原来的 INFO 命令输出缓存并创建新的缓存区
    sdsfree(ri->info);
    ri->info = sdsnew(info);
    // 重置从节点复制断开时间，防止在 INFO 输出中找不到该信息
    ri->master_link_down_time = 0;
    // 一行一行的进行遍历，并把行数放到 numlines 中
    lines = sdssplitlen(info,strlen(info),"\r\n",2,&numlines);
    // 逐行处理
    for (j = 0; j < numlines; j++) {
        sentinelRedisInstance *slave;
        // 当前行内容
        sds l = lines[j];
        // 提取出 run_id，run_id 是长度为 40 的字符串，加上 "run_id:" 这几个字符串 总共长度是 47，所以大于 47长度的行 我们来进行判断，并检测 run_id 字符串
        // runid 这行的格式为 "run_id:de4a10ef73028f75e4bd9d6e345544b8ed5e9e09\r\n"
        if (sdslen(l) >= 47 && !memcmp(l,"run_id:",7)) {
            // 如果原 runid 为空，则更新为当前 l 内容中的 run_id 内容
            if (ri->runid == NULL) {
                ri->runid = sdsnewlen(l+7,40);
            } else {
            // 如果原 runid 不为空，则更新
                if (strncmp(ri->runid,l+7,40) != 0) {
                    // 事件通知
                    sentinelEvent(LL_NOTICE,"+reboot",ri,"%@");
                    // 清空原来 runid
                    sdsfree(ri->runid);
                    // 重赋值
                    ri->runid = sdsnewlen(l+7,40);
                }
            }
        }
        // 读取从服务器节点的 ip 和 port 信息
        // 老版本格式: slave0:<ip>,<port>,<state>
        // 新版本格式: slave0:ip=127.0.0.1,port=9999,...
        if ((ri->flags & SRI_MASTER) &&
            sdslen(l) >= 7 &&
            !memcmp(l,"slave",5) && isdigit(l[5]))
        {
            char *ip, *port, *end;
            // 老版本，分别定位 ip 和 port 信息
            if (strstr(l,"ip=") == NULL) {
                ip = strchr(l,':'); if (!ip) continue;
                ip++;
                port = strchr(ip,','); if (!port) continue;
                *port = '\0';
                port++;
                end = strchr(port,','); if (!end) continue;
                *end = '\0';
            } else {
            // 新版本
                ip = strstr(l,"ip="); if (!ip) continue;
                ip += 3;
                port = strstr(l,"port="); if (!port) continue;
                port += 5;
                end = strchr(ip,','); if (end) *end = '\0';
                end = strchr(port,','); if (end) *end = '\0';
            }
            // 查看新解析的 ip:port 是否已经存在于当前 主服务器节点所属中
            if (sentinelRedisInstanceLookupSlave(ri,ip,atoi(port)) == NULL) {
                // 如果不存在，则创建一个 从服务器节点实例
                if ((slave = createSentinelRedisInstance(NULL,SRI_SLAVE,ip,
                            atoi(port), ri->quorum, ri)) != NULL)
                {
                    // 发送事件通知
                    sentinelEvent(LL_NOTICE,"+slave",slave,"%@");
                    // 刷新 config
                    sentinelFlushConfig();
                }
            }
        }
        // 提取 master_link_down_since_seconds 主从节点断线时长
        if (sdslen(l) >= 32 &&
            !memcmp(l,"master_link_down_since_seconds",30))
        {
            ri->master_link_down_time = strtoll(l+31,NULL,10)*1000;
        }
        // 根据提取信息判断角色
        if (!memcmp(l,"role:master",11)) role = SRI_MASTER;
        else if (!memcmp(l,"role:slave",10)) role = SRI_SLAVE;
        // 如果是从服务器节点
        if (role == SRI_SLAVE) {
            // 获取对应主服务器 ip
            if (sdslen(l) >= 12 && !memcmp(l,"master_host:",12)) {
                if (ri->slave_master_host == NULL ||
                    strcasecmp(l+12,ri->slave_master_host))
                {
                    sdsfree(ri->slave_master_host);
                    ri->slave_master_host = sdsnew(l+12);
                    ri->slave_conf_change_time = mstime();
                }
            }
            // 获取对应主服务器 port
            if (sdslen(l) >= 12 && !memcmp(l,"master_port:",12)) {
                int slave_master_port = atoi(l+12);
                if (ri->slave_master_port != slave_master_port) {
                    ri->slave_master_port = slave_master_port;
                    ri->slave_conf_change_time = mstime();
                }
            }
            // 获取主服务器节点的连接状态
            if (sdslen(l) >= 19 && !memcmp(l,"master_link_status:",19)) {
                ri->slave_master_link_status =
                    (strcasecmp(l+19,"up") == 0) ?
                    SENTINEL_MASTER_LINK_STATUS_UP :
                    SENTINEL_MASTER_LINK_STATUS_DOWN;
            }
            // 更新从服务器节点优先级
            if (sdslen(l) >= 15 && !memcmp(l,"slave_priority:",15))
                ri->slave_priority = atoi(l+15);
            // 更新从服务器节点的复制偏移量
            if (sdslen(l) >= 18 && !memcmp(l,"slave_repl_offset:",18))
                ri->slave_repl_offset = strtoull(l+18,NULL,10);
        }
    }
    // 获取最新的 INFO 时间
    ri->info_refresh = mstime();
    sdsfreesplitres(lines,numlines);
    // 如果角色变换了，更新角色和角色变换时间
    if (role != ri->role_reported) {
        ri->role_reported_time = mstime();
        ri->role_reported = role;
        if (role == SRI_SLAVE) ri->slave_conf_change_time = mstime();
        sentinelEvent(LL_VERBOSE,
            ((ri->flags & (SRI_MASTER|SRI_SLAVE)) == role) ?
            "+role-change" : "-role-change",
            ri, "%@ new reported role is %s",
            role == SRI_MASTER ? "master" : "slave",
            ri->flags & SRI_MASTER ? "master" : "slave");
    }
    // 如果 TILT 模式，暂停
    if (sentinel.tilt) return;
    
    // 如果实例本身是主服务器节点但是 INFO 回复的却是从服务器节点，原因一般是 sentinel 判断不可达，发生了故障转移所致，这里不进行处理
    if ((ri->flags & SRI_MASTER) && role == SRI_SLAVE) {
    }
    // 如果实例本身是从服务器节点但是 INFO 回复的却是主服务器节点，那就是因为故障转移给升级了
    if ((ri->flags & SRI_SLAVE) && role == SRI_MASTER) {
        // 如果当前为被晋升的从服务器节点，并且主服务器节点在等待从服务器节点去晋升为主服务器节点
        if ((ri->flags & SRI_PROMOTED) &&
            (ri->master->flags & SRI_FAILOVER_IN_PROGRESS) &&
            (ri->master->failover_state ==
                SENTINEL_FAILOVER_STATE_WAIT_PROMOTION))
        {
            // 要确保这个从服务器节点被配置为主节点，并且将纪元设置为故障转移优先级最高的纪元，这样才能强制其他 sentinel 节点取更新他们的配置，更新旧的主节点的纪元
            ri->master->config_epoch = ri->master->failover_epoch;
            // 设置旧的主节点的故障转移状态为SENTINEL_FAILOVER_STATE_RECONF_SLAVES，该状态会向旧的主节点所属的从节点发送通知让他们向新的主节点发起复制操作
            ri->master->failover_state = SENTINEL_FAILOVER_STATE_RECONF_SLAVES;
            // 设置故障转移状态改变的时间
            ri->master->failover_state_change_time = mstime();
            // 刷新配置
            sentinelFlushConfig();
            // 事件通知
            sentinelEvent(LL_WARNING,"+promoted-slave",ri,"%@");
            // 如果开启了故障模拟标识，而且是晋升之后发生故障
            if (sentinel.simfailure_flags &
                SENTINEL_SIMFAILURE_CRASH_AFTER_PROMOTION)
                // 那么就退出程序
                sentinelSimFailureCrash();
            // 事件通知
            sentinelEvent(LL_WARNING,"+failover-state-reconf-slaves", ri->master,"%@");
            // 重新配置脚本属性，放入脚本队列
            sentinelCallClientReconfScript(ri->master,SENTINEL_LEADER, "start",ri->master->addr,ri->addr);
            // 强行发送 hello 给所有的 redis 节点 和 sentinel 实例去关联指定的主服务器节点实例
            sentinelForceHelloUpdateForMaster(ri->master);
        } else {
        // 从服务器节点晋升为了主服务器节点，但是旧的主服务器节点在发生故障转移的超时时间以后又重新上线了，因此要将旧的主服务器节点重新设置为从服务器节点
            
            // 从节点晋升为了主节点，我们要强制其重新配置成从节点，等待一些时间，然后无论如何都要让其接收新的配置，默认计算等待的时间为 8s
            mstime_t wait_time = SENTINEL_PUBLISH_PERIOD*4;
            // 当前实例已经成为主节点，且在最近的 wait_time 毫秒内，如果当前实例没有出现下线状态，并且实例的主节点实例看起来很健壮，并且距离角色更新的时间已经超过缓冲的 wait_time
            if (!(ri->flags & SRI_PROMOTED) &&
                 sentinelMasterLooksSane(ri->master) &&
                 sentinelRedisInstanceNoDownFor(ri,wait_time) &&
                 mstime() - ri->role_reported_time > wait_time)
            {
                // 发送 slaveof 命令，使其成为主节点
                int retval = sentinelSendSlaveOf(ri,
                        ri->master->addr->ip,
                        ri->master->addr->port);
                if (retval == C_OK)
                    sentinelEvent(LL_NOTICE,"+convert-to-slave",ri,"%@");
            }
        }
    }
    // 如果当前实例为从服务器节点，而且 INFO 的回复确实也是 从服务器节点，但是 主服务器节点的 ip/port 发生了改变，那么需要 从服务器节点重新指向并复制修改过 ip/port 信息的主节点
    if ((ri->flags & SRI_SLAVE) &&
        role == SRI_SLAVE &&
        (ri->slave_master_port != ri->master->addr->port ||
         strcasecmp(ri->slave_master_host,ri->master->addr->ip)))
    {
        // 故障转移超时时间
        mstime_t wait_time = ri->master->failover_timeout;
        // wait_time 时间内当前实例没有出现下线状态，而且主节点变换 ip/port 信息已经超过了 wait_time
        if (sentinelMasterLooksSane(ri->master) &&
            sentinelRedisInstanceNoDownFor(ri,wait_time) &&
            mstime() - ri->slave_conf_change_time > wait_time)
        {
            // 改变 slaveof 的目标地址
            int retval = sentinelSendSlaveOf(ri,
                    ri->master->addr->ip,
                    ri->master->addr->port);
            if (retval == C_OK)
                sentinelEvent(LL_NOTICE,"+fix-slave-config",ri,"%@");
        }
    }
    // 如果当前实例为从服务器节点，并且 INFO 传递过来的信息确实也是从节点，并且 sentinel 向当前实例发送了 slaveof 命令 或者 当前实例已经在同步新的主节点了
    if ((ri->flags & SRI_SLAVE) && role == SRI_SLAVE &&
        (ri->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG)))
    {
        // 将 SRI_RECONF_SENT 状态改为 SRI_RECONF_INPROG，因为 sentinel 向 实例发送了 slaveof 命令，但是当前实例所属主节点的地址已经和新主节点地址相同，所以，将状态给为 SRI_RECONF_INPROG，表示正在同步复制操作
        if ((ri->flags & SRI_RECONF_SENT) &&
            ri->slave_master_host &&
            strcmp(ri->slave_master_host,
                    ri->master->promoted_slave->addr->ip) == 0 &&
            ri->slave_master_port == ri->master->promoted_slave->addr->port)
        {
            ri->flags &= ~SRI_RECONF_SENT;
            ri->flags |= SRI_RECONF_INPROG;
            sentinelEvent(LL_NOTICE,"+slave-reconf-inprog",ri,"%@");
        }
        // 将状态由 SRI_RECONF_INPROG 改为 SRI_RECONF_DONE，表示同步完成
        if ((ri->flags & SRI_RECONF_INPROG) &&
            ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP)
        {
            ri->flags &= ~SRI_RECONF_INPROG;
            ri->flags |= SRI_RECONF_DONE;
            sentinelEvent(LL_NOTICE,"+slave-reconf-done",ri,"%@");
        }
    }
}

那么关于 PING 命令呢？上面我们提到了调用 sentinelSendPing() 函数，该函数的逻辑就比较简单了，发送 PING 命令，并设置回调函数 sentinelPingReplyCallback()，再加以各项数据统计等，那我们接下来看看一下 回调函数的逻辑

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
/* sentinel.c: line 2279 */
void sentinelPingReplyCallback(redisAsyncContext *c, void *reply, void *privdata) {
    sentinelRedisInstance *ri = privdata;
    instanceLink *link = c->data;
    redisReply *r;
    // 如果没有响应或者连接问题，直接返回
    if (!reply || !link) return;
    // 到这里证明已经响应，那么未回复命令个数 -1
    link->pending_commands--;
    r = reply;
    // 如果实例处于 REDIS_REPLY_STATUS 或者 REDIS_REPLY_ERROR 状态
    if (r->type == REDIS_REPLY_STATUS ||
        r->type == REDIS_REPLY_ERROR) {
        // 如果回复是 PONG、LOADING、MASTERDOWN 都属于正常响应，属于可接受的响应，则更新回复时间
        if (strncmp(r->str,"PONG",4) == 0 ||
            strncmp(r->str,"LOADING",7) == 0 ||
            strncmp(r->str,"MASTERDOWN",10) == 0)
        {
            // 更新回复时间
            link->last_avail_time = mstime();
            // 接收到信息后，设置最近发送 PING 命令时间为 0
            link->act_ping_time = 0;
        } else {
            // 如果回复是 BUSY，那么发送一个 SCRIPT KILL 命令来恢复因为执行脚本而看起来是主观下线状态的服务器
            if (strncmp(r->str,"BUSY",4) == 0 &&
                (ri->flags & SRI_S_DOWN) &&
                !(ri->flags & SRI_SCRIPT_KILL_SENT))
            {
                // 异步发送一个 script kill 命令，并设置回调函数
                if (redisAsyncCommand(ri->link->cc,
                        sentinelDiscardReplyCallback, ri,
                        "SCRIPT KILL") == C_OK)
                    // 发送成功，但是 未回复命令就要 +1 了
                    ri->link->pending_commands++;
                // 改变状态为 SRI_SCRIPT_KILL_SENT
                ri->flags |= SRI_SCRIPT_KILL_SENT;
            }
        }
    }
    // 更新最近一次收到 PING 回复的时间
    link->last_pong_time = mstime();
}

还剩一个命令就是 PUBLISH 了，其实就是向 __sentinel__:hello 频道发送信息，处理函数为 sentinelSendHello()，信息内容主要为 哨兵信息和主服务器节点信息，发送后 设置回调函数 sentinelPublishReplyCallback()

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
/* sentinel.c: line 2481 */
int sentinelSendHello(sentinelRedisInstance *ri) {
    char ip[NET_IP_STR_LEN];
    char payload[NET_IP_STR_LEN+1024];
    int retval;
    char *announce_ip;
    int announce_port;
    // 主服务器节点实例
    sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? ri : ri->master;
    // 主服务器节点的网络地址
    sentinelAddr *master_addr = sentinelGetCurrentMasterAddress(master);
    // 如果处于关闭状态，则返回
    if (ri->link->disconnected) return C_ERR;
    // 如果已经指定了地址就使用
    if (sentinel.announce_ip) {
        announce_ip = sentinel.announce_ip;
    } else {
    // 否则直接使用主服务器节点地址
        // 获取主节点地址
        if (anetSockName(ri->link->cc->c.fd,ip,sizeof(ip),NULL) == -1)
            return C_ERR;
        announce_ip = ip;
    }
    // 获取端口
    announce_port = sentinel.announce_port ?
                    sentinel.announce_port : server.port;
    // 组装 hello 信息，包括 sentinel 地址端口、runid 和 纪元版本信息，主服务器节点 别名、地址端口 和 纪元版本信息
    snprintf(payload,sizeof(payload),
        "%s,%d,%s,%llu,"
        "%s,%s,%d,%llu",
        announce_ip, announce_port, sentinel.myid,
        (unsigned long long) sentinel.current_epoch,
        master->name,master_addr->ip,master_addr->port,
        (unsigned long long) master->config_epoch);
    // 执行 PUBLISH，发送 hello 信息，并设置回调函数
    retval = redisAsyncCommand(ri->link->cc,
        sentinelPublishReplyCallback, ri, "PUBLISH %s %s",
            SENTINEL_HELLO_CHANNEL,payload);
    if (retval != C_OK) return C_ERR;
    // 未回复命令个数+1
    ri->link->pending_commands++;
    return C_OK;
}

回调函数 sentinelPublishReplyCallback() 的逻辑就比较简单了，主要为更新统计信息

主/客 观下线状态判定

上面概况中我们提到 主观下线，如果 监控的服务器节点 在 down-after-milliseconds 设置的毫秒时效内没有响应检测，则会被判定为 主观下线；这个状态适用所有服务器节点；处理函数为 sentinelCheckSubjectivelyDown()

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
/* sentinel.c: line 3435 */
void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) {
    mstime_t elapsed = 0;
    // 获取命令响应已经过去的时长
    if (ri->link->act_ping_time)
        elapsed = mstime() - ri->link->act_ping_time;
    // 实例连接断开了
    else if (ri->link->disconnected)
        elapsed = mstime() - ri->link->last_avail_time;
    // 低活跃度重新连接，即 连接超过了 1.5秒，并且之前发送了 PING 命令但是活跃度很低
    if (ri->link->cc &&
        (mstime() - ri->link->cc_conn_time) >
        SENTINEL_MIN_LINK_RECONNECT_PERIOD &&
        ri->link->act_ping_time != 0 &&
        (mstime() - ri->link->act_ping_time) > (ri->down_after_period/2) &&
        (mstime() - ri->link->last_pong_time) > (ri->down_after_period/2))
    {
        // 断开实例的连接
        instanceLinkCloseConnection(ri->link,ri->link->cc);
    }
    // 检查 订阅发布的连接是不是也是出于 低活跃度
    if (ri->link->pc &&
        (mstime() - ri->link->pc_conn_time) >
         SENTINEL_MIN_LINK_RECONNECT_PERIOD &&
        (mstime() - ri->link->pc_last_activity) > (SENTINEL_PUBLISH_PERIOD*3))
    {
        // 断开实例连接
        instanceLinkCloseConnection(ri->link,ri->link->pc);
    }
    
    // 如果定期没有回复信息 或者 sentinel 认为是主节点但是 报告是从节点
    if (elapsed > ri->down_after_period ||
        (ri->flags & SRI_MASTER &&
         ri->role_reported == SRI_SLAVE &&
         mstime() - ri->role_reported_time >
          (ri->down_after_period+SENTINEL_INFO_PERIOD*2)))
    {
        // 设置主观下线标识
        if ((ri->flags & SRI_S_DOWN) == 0) {
            sentinelEvent(LL_WARNING,"+sdown",ri,"%@");
            ri->s_down_since_time = mstime();
            ri->flags |= SRI_S_DOWN;
        }
    } else {
        // 如果已经是主观下线状态，则取消标识
        if (ri->flags & SRI_S_DOWN) {
            sentinelEvent(LL_WARNING,"-sdown",ri,"%@");
            ri->flags &= ~(SRI_S_DOWN|SRI_SCRIPT_KILL_SENT);
        }
    }
}

而 客观下线 为 主服务器节点在发生故障的时候，sentinel 会通过 is-master-down-by-addr 命令向其他 sentinel 节点询问 该主服务器节点的状态，如果超过 quorum 个数的哨兵节点都认为 主服务器节点不可达，则判定为 客观下线；这个状态只会针对 主服务器节点；操作函数为 sentinelCheckObjectivelyDown()

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
/* sentinel.c: line 3507 */
void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) {
    dictIterator *di;
    dictEntry *de;
    unsigned int quorum = 0, odown = 0;
    // 如果已经被判断为主观下线
    if (master->flags & SRI_S_DOWN) {
        // 当前 哨兵认为下线，投票1
        quorum = 1;
        di = dictGetIterator(master->sentinels);
        // 遍历所有监控该主节点的所有 sentinel
        while((de = dictNext(di)) != NULL) {
            sentinelRedisInstance *ri = dictGetVal(de);
            // 如果 sentinel 投票主观下线，投票 +1
            if (ri->flags & SRI_MASTER_DOWN) quorum++;
        }
        dictReleaseIterator(di);
        // 如果投票数够了，则认为 客观下线
        if (quorum >= master->quorum) odown = 1;
    }
    // 如果已经判定为 客观下线了
    if (odown) {
        // 如果主服务器节点还没改变状态，则修改其 客观下线状态
        if ((master->flags & SRI_O_DOWN) == 0) {
            sentinelEvent(LL_WARNING,"+odown",master,"%@ #quorum %d/%d", quorum, master->quorum);
            master->flags |= SRI_O_DOWN;
            master->o_down_since_time = mstime();
        }
    } else {
    // 如果票数不足，原本 客观下线就要改回来为 主观下线
        if (master->flags & SRI_O_DOWN) {
            sentinelEvent(LL_WARNING,"-odown",master,"%@");
            master->flags &= ~SRI_O_DOWN;
        }
    }
}

故障转移

确定客观下线后即将进行故障转移，处理函数为 sentinelFailoverStateMachine()，我们来看一下详细逻辑

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
/* sentinel.c: line 4218 */
void sentinelFailoverStateMachine(sentinelRedisInstance *ri) {
    // 要求必须是主节点
    serverAssert(ri->flags & SRI_MASTER);
    // 如果不是故障转移状态则直接返回
    if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS)) return;
    // 故障转移过程为循环一个个状态，所以这个函数会进行多次循环判断
    switch(ri->failover_state) {
        // 故障转移开始
        case SENTINEL_FAILOVER_STATE_WAIT_START:
            sentinelFailoverWaitStart(ri);
            break;
        // 选择要晋升的从服务器节点
        case SENTINEL_FAILOVER_STATE_SELECT_SLAVE:
            sentinelFailoverSelectSlave(ri);
            break;
        // 发送 slaveof no one 命令，从变主
        case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE:
            sentinelFailoverSendSlaveOfNoOne(ri);
            break;
        // 等待选择的从服务器节点变成主服务器节点，如果超时就重新选择新的从服务器节点
        case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION:
            sentinelFailoverWaitPromotion(ri);
            break;
        // 给所有从服务器发送 slaveof 通知，让他们同步 新的主服务器节点
        case SENTINEL_FAILOVER_STATE_RECONF_SLAVES:
            sentinelFailoverReconfNextSlave(ri);
            break;
    }
}

故障转移步骤是分步但是也是连续的，成功执行完一个状态，下一个时间事件顺序执行下一个状态，每一次时间事件执行一个状态，所以如果中途执行到某个状态发现主服务器节点可达了，那么故障转移结束，我们来看一下具体每个状态的逻辑

故障转移开始函数 sentinelFailoverWaitStart()

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
/* sentinel.c: line 3991 */
void sentinelFailoverWaitStart(sentinelRedisInstance *ri) {
    char *leader;
    int isleader;
    // 选举出来一个 sentinel 领导者，用来主导故障转移，并更新状态，这里的选举使用了 Raft 一致性算法，具体可以参考 http://www.cnblogs.com/mindwind/p/5231986.html
    leader = sentinelGetLeader(ri, ri->failover_epoch);
    // 当前 sentinel 节点是否就是领导节点
    isleader = leader && strcasecmp(leader,sentinel.myid) == 0;
    sdsfree(leader);
    // 如果当前 sentinel 不是领导者，而且也不是强制故障转移，那么经过简单判断就会返回
    if (!isleader && !(ri->flags & SRI_FORCE_FAILOVER)) {
        int election_timeout = SENTINEL_ELECTION_TIMEOUT;
        // 选举超时时间 最大为 配置的 failover_timeout 故障转移时间
        if (election_timeout > ri->failover_timeout)
            election_timeout = ri->failover_timeout;
        // 如果选举超时，则取消故障转移
        if (mstime() - ri->failover_start_time > election_timeout) {
            // 发送事件通知
            sentinelEvent(LL_WARNING,"-failover-abort-not-elected",ri,"%@");
            // 取消故障转移
            sentinelAbortFailover(ri);
        }
        return;
    }
    // 如果当前 sentinel 为领导者，则发送 赢取选举的 事件通知
    sentinelEvent(LL_WARNING,"+elected-leader",ri,"%@");
    // 是否指定了故障模拟，并且是 当选之后的模拟故障
    if (sentinel.simfailure_flags & SENTINEL_SIMFAILURE_CRASH_AFTER_ELECTION)
        // 退出当前节点程序
        sentinelSimFailureCrash();
    // 故障转移状态 改为后一个，等待选择一个 晋升的从服务器节点
    ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE;
    // 故障转移状态变更时间
    ri->failover_state_change_time = mstime();
    // 发送事件通知
    sentinelEvent(LL_WARNING,"+failover-state-select-slave",ri,"%@");
}

选择要晋升的从服务器节点函数 sentinelFailoverSelectSlave()

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
/* sentinel.c: line 4024 */
void sentinelFailoverSelectSlave(sentinelRedisInstance *ri) {
    // 往实例中挑选一个 从服务器节点，这里挑选 节点还是有一些说法的，我们后面详细解说
    sentinelRedisInstance *slave = sentinelSelectSlave(ri);
    // 没有找到合适的从服务器节点就停止故障转移
    if (slave == NULL) {
        // 发送事件通知
        sentinelEvent(LL_WARNING,"-failover-abort-no-good-slave",ri,"%@");
        // 停止故障转移
        sentinelAbortFailover(ri);
    } else {
    // 有合适的从服务器节点
        // 发送成功选择晋升节点的事件通知
        sentinelEvent(LL_WARNING,"+selected-slave",slave,"%@");
        // 设置晋升标识
        slave->flags |= SRI_PROMOTED;
        // 设置晋升从服务器节点
        ri->promoted_slave = slave;
        // 故障转移状态修改为 SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE
        ri->failover_state = SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE;
        // 故障转移状态修改时间
        ri->failover_state_change_time = mstime();
        // 发送事件通知
        sentinelEvent(LL_NOTICE,"+failover-state-send-slaveof-noone", slave, "%@");
    }
}

发送 slaveof no one 命令，从变主 处理函数为 sentinelFailoverSendSlaveOfNoOne()

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
/* sentinel.c: line 4043 */
void sentinelFailoverSendSlaveOfNoOne(sentinelRedisInstance *ri) {
    int retval;
    // 如果晋升的从服务器节点处于断线状态，那就不能发送命令了，当然现在 既然是晋升状态了，在一定时间内只要没有超时，还会重试的
    if (ri->promoted_slave->link->disconnected) {
        // 如果已经超时，就只能取消故障转移了
        if (mstime() - ri->failover_state_change_time > ri->failover_timeout) {
            // 发送事件通知
            sentinelEvent(LL_WARNING,"-failover-abort-slave-timeout",ri,"%@");
            // 取消故障转移
            sentinelAbortFailover(ri);
        }
        return;
    }
    
    // 发送 slaveof no one 命令给晋升的节点 升级为 主服务器节点
    retval = sentinelSendSlaveOf(ri->promoted_slave,NULL,0);
    if (retval != C_OK) return;
    // 发送成功的事件通知
    sentinelEvent(LL_NOTICE, "+failover-state-wait-promotion", ri->promoted_slave,"%@");
    // 故障转移状态改为 SENTINEL_FAILOVER_STATE_WAIT_PROMOTION
    ri->failover_state = SENTINEL_FAILOVER_STATE_WAIT_PROMOTION;
    // 修改故障转移状态时间
    ri->failover_state_change_time = mstime();
}

等待选择的从服务器节点变成主服务器节点，如果超时就重新选择新的从服务器节点，处理函数为 sentinelFailoverWaitPromotion()，不过逻辑比较简单，主要是判断是否超时

给所有从服务器发送 slaveof 通知，让他们同步 新的主服务器节点，处理函数为 sentinelFailoverReconfNextSlave()

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
/* sentinel.c: line 4145 */
void sentinelFailoverReconfNextSlave(sentinelRedisInstance *master) {
    dictIterator *di;
    dictEntry *de;
    int in_progress = 0;
    // 获取遍历所有从服务器节点的迭代器
    di = dictGetIterator(master->slaves);
    // 遍历所有从服务器节点
    while((de = dictNext(di)) != NULL) {
        // 节点实例
        sentinelRedisInstance *slave = dictGetVal(de);
        // 计算已经发送同步命令或者正在同步的从服务器节点
        if (slave->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG))
            in_progress++;
    }
    // 释放迭代器
    dictReleaseIterator(di);
    
    di = dictGetIterator(master->slaves);
    // 如果已经发送同步命令或者已经正在同步的从节点个数小于设置的同步个数限制，那么遍历所有的从节点
    while(in_progress < master->parallel_syncs &&
          (de = dictNext(di)) != NULL)
    {
        // 获取实例
        sentinelRedisInstance *slave = dictGetVal(de);
        int retval;
        // 跳过被晋升的节点和已经完成同步的节点
        if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue;
        // 如果从服务器节点已经发送了 salveof 命令，但是在 故障转移到下一状态的时候 超时了
        if ((slave->flags & SRI_RECONF_SENT) &&
            (mstime() - slave->slave_reconf_sent_time) >
            SENTINEL_SLAVE_RECONF_TIMEOUT)
        {
            // 发送事件通知
            sentinelEvent(LL_NOTICE,"-slave-reconf-sent-timeout",slave,"%@");
            // 清除发送 slaveof 命令的标识
            slave->flags &= ~SRI_RECONF_SENT;
            // 设置为完成同步的标识，随后重新发送 slaveof 命令，再尝试同步
            slave->flags |= SRI_RECONF_DONE;
        }
        
        // 跳过已经发送过命令和正在同步的从服务器节点
        if (slave->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG)) continue;
        // 如果连接断开了就跳过
        if (slave->link->disconnected) continue;
        // 向所有从节点发送 slaveof 命令
        retval = sentinelSendSlaveOf(slave,
                master->promoted_slave->addr->ip,
                master->promoted_slave->addr->port);
        // 发送成功
        if (retval == C_OK) {
            // 状态变更为已经发送了 slaveof 命令标识
            slave->flags |= SRI_RECONF_SENT;
            // 发送 slaveof 命令时间
            slave->slave_reconf_sent_time = mstime();
            // 发送事件通知
            sentinelEvent(LL_NOTICE,"+slave-reconf-sent",slave,"%@");
            in_progress++;
        }
    }
    dictReleaseIterator(di);
    // 判断故障转移是否结束
    sentinelFailoverDetectEnd(master);
}

我们上面提到了选举节点的逻辑还是有一些说法，接下来就进行详细逐一说明

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
/* sentinel.c: line 3946 */
sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master) {
    // 申请从节点数组空间
    sentinelRedisInstance **instance =
        zmalloc(sizeof(instance[0])*dictSize(master->slaves));
    sentinelRedisInstance *selected = NULL;
    int instances = 0;
    dictIterator *di;
    dictEntry *de;
    mstime_t max_master_down_time = 0;
    // 如果主服务器节点处于主观下线状态，则计算下线的最大时长
    if (master->flags & SRI_S_DOWN)
        max_master_down_time += mstime() - master->s_down_since_time;
    max_master_down_time += master->down_after_period * 10;
    di = dictGetIterator(master->slaves);
    // 遍历所有从服务器节点
    while((de = dictNext(di)) != NULL) {
        // 获取实例
        sentinelRedisInstance *slave = dictGetVal(de);
        mstime_t info_validity_time;
        // 如果当前从服务器节点下线了，就跳过去
        if (slave->flags & (SRI_S_DOWN|SRI_O_DOWN)) continue;
        // 如果当前从服务器节点掉线了，也跳过去
        if (slave->link->disconnected) continue;
        // PING 命令响应事件太久，也跳过去
        if (mstime() - slave->link->last_avail_time > SENTINEL_PING_PERIOD*5) continue;
        // 跳过优先级为 0 的节点
        if (slave->slave_priority == 0) continue;
        // 如果主服务器节点主观下线状态，sentinel 每秒发送 INFO 命令给 从服务器节点，否则就按照默认事件走
        if (master->flags & SRI_S_DOWN)
            info_validity_time = SENTINEL_PING_PERIOD*5;
        else
            info_validity_time = SENTINEL_INFO_PERIOD*3;
        // 如果从服务器节点接收 INFO 命令回复时间已经过期，也跳过去
        if (mstime() - slave->info_refresh > info_validity_time) continue;
        // 如果是下线时间过长的节点，也跳过去
        if (slave->master_link_down_time > max_master_down_time) continue;
        // 将剩余选中的节点放到备选数组中
        instance[instances++] = slave;
    }
    dictReleaseIterator(di);
    // 处理备选数组
    if (instances) {
        // 将备选数组中的元素排序，取第一个，作为晋升节点
        qsort(instance,instances,sizeof(sentinelRedisInstance*),
            compareSlavesForPromotion);
        selected = instance[0];
    }
    zfree(instance);
    return selected;
}

上面我们说到给备选从节点数组排序，那么按照什么逻辑排序呢，我们来看一下 排序比较函数 compareSlavesForPromotion()，总体的原则就是 优先级低的优先、复制偏移量大的优先、runid字典序小的优先

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
/* sentinel.c: line 3918 */
int compareSlavesForPromotion(const void *a, const void *b) {
    sentinelRedisInstance **sa = (sentinelRedisInstance **)a,
                          **sb = (sentinelRedisInstance **)b;
    char *sa_runid, *sb_runid;
    // 选择优先级小的
    if ((*sa)->slave_priority != (*sb)->slave_priority)
        return (*sa)->slave_priority - (*sb)->slave_priority;
    // 比较复制偏移量，量大的优先，证明数据更全
    if ((*sa)->slave_repl_offset > (*sb)->slave_repl_offset) {
        return -1;
    } else if ((*sa)->slave_repl_offset < (*sb)->slave_repl_offset) {
        return 1;
    }
    // 获取两个比较节点的 runid
    sa_runid = (*sa)->runid;
    sb_runid = (*sb)->runid;
    // NULL 的判断是为了兼容老版本，因为 NULL 的 runid 比 任何 runid 都大
    if (sa_runid == NULL && sb_runid == NULL) return 0;
    else if (sa_runid == NULL) return 1;
    else if (sb_runid == NULL) return -1;
    // 复制偏移量一样的情况下，运行 runid 字典序小的优先
    return strcasecmp(sa_runid, sb_runid);
}

至此主要的逻辑我们分析了一遍，不过有点儿绕哈，由于很多函数都是站在不同角度（主服务器节点、从服务器节点、sentinel 节点）执行的，所以看的时候需要注意，另外就是发送命令基本都是异步的，所以可能存在响应不那么及时的情况，所以一般可能需要不止一次的循环运算

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-sentinel.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 /> 喵喵、念念、健身、吉他！