Redis Cluster 的实现 - 初始化(1)

56gt

void clusterInit(void) {　　
    int saveconf = 0;
　　

　　
    /* 对 redisServer 结构中的 clusterState 进行初始化 */
　　
    server.cluster = zmalloc(sizeof(clusterState));
　　
    server.cluster->myself = NULL;
　　
    server.cluster->currentEpoch = 0;
　　
    // 初始为 FAIL 状态
　　
    server.cluster->state = REDIS_CLUSTER_FAIL;
　　
    // master 节点数
　　
    server.cluster->size = 1;
　　
    server.cluster->todo_before_sleep = 0;
　　
    server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL);
　　
    server.cluster->nodes_black_list =
　　
        dictCreate(&clusterNodesBlackListDictType,NULL);
　　
    server.cluster->failover_auth_time = 0;
　　
    server.cluster->failover_auth_count = 0;
　　
    server.cluster->failover_auth_rank = 0;
　　
    server.cluster->failover_auth_epoch = 0;
　　
    server.cluster->lastVoteEpoch = 0;
　　
    server.cluster->stats_bus_messages_sent = 0;
　　
    server.cluster->stats_bus_messages_received = 0;
　　
    memset(server.cluster->slots,0, sizeof(server.cluster->slots));
　　
    clusterCloseAllSlots();
　　
    /* Lock the cluster config file to make sure every node uses
　　
     * its own nodes.conf. */
　　
    if (clusterLockConfig(server.cluster_configfile) == REDIS_ERR)
　　
        exit(1);
　　
    /* Load or create a new nodes configuration. */
　　

　　
    // 加载或创建一个新的 节点配置 文件
　　
    // 如果加载失败，则通过 createClusterNode 来创建一个 flags = MYSELF|MASTER 的新节点
　　
    if (clusterLoadConfig(server.cluster_configfile) == REDIS_ERR) {
　　
        /* No configuration found. We will just use the random name provided
　　
         * by the createClusterNode() function. */
　　
        myself = server.cluster->myself =
　　
            createClusterNode(NULL,REDIS_NODE_MYSELF|REDIS_NODE_MASTER);
　　
        redisLog(REDIS_NOTICE,"No cluster configuration found, I'm %.40s",
　　
            myself->name);
　　

　　
        // 将此节点加入 cluster 的 nodes hash table 中
　　
        // 该 nodes 维护的是一张 nodeName -> node 的 hash 表
　　
        clusterAddNode(myself);
　　
        saveconf = 1;
　　
    }
　　

　　
    // 具体的配置文件名由参数 cluster-config-file 来指定
　　
    if (saveconf) clusterSaveConfigOrDie(1);
　　
    /* We need a listening TCP port for our cluster messaging needs. */
　　
    server.cfd_count = 0;
　　
    /* Port sanity check II
　　
     * The other handshake port check is triggered too late to stop
　　
     * us from trying to use a too-high cluster port number. */
　　
    // 这里是做端口校验，要求 redis 面向客户端的监听端口必须小于 55535，这样才能避免出现
　　
    // cluster 通道的监听端口 > 65535 的情形
　　
    if (server.port > (65535-REDIS_CLUSTER_PORT_INCR)) {
　　
        redisLog(REDIS_WARNING, "Redis port number too high. "
　　
                   "Cluster communication port is 10,000 port "
　　
                   "numbers higher than your Redis port. "
　　
                   "Your Redis port number must be "
　　
                   "lower than 55535.");
　　
        exit(1);
　　
    }
　　

　　
    // 打开 cluster 通道的 非阻塞监听端口
　　
    if (listenToPort(server.port+REDIS_CLUSTER_PORT_INCR,
　　
        server.cfd,&server.cfd_count) == REDIS_ERR)
　　
    {
　　
        exit(1);
　　
    } else {
　　
        int j;
　　

　　
        // 在一个或多个非阻塞监听套接字上创建 ACCEPT 事件处理器
　　
        // 可以根据系统平台选择合适的 事件模型(如：Linux 上的 epoll，具体查看 aeApiAddEvent 源码)
　　
        // 这里指定了 clusterAcceptHandler 函数作为 ACCEPT 事件处理器
　　
        for (j = 0; j < server.cfd_count; j++) {
　　
            if (aeCreateFileEvent(server.el, server.cfd[j], AE_READABLE,
　　
                clusterAcceptHandler, NULL) == AE_ERR)
　　
                    redisPanic("Unrecoverable error creating Redis Cluster "
　　
                                "file event.");
　　
        }
　　
    }
　　
    /* The slots -> keys map is a sorted set. Init it. */
　　
    server.cluster->slots_to_keys = zslCreate();
　　
    /* Set myself->port to my listening port, we'll just need to discover
　　
     * the IP address via MEET messages. */
　　
    myself->port = server.port;
　　
    resetManualFailover();
　　
}