private void innerJoinCluster() {
        DiscoveryNode masterNode = null;
        final Thread currentThread = Thread.currentThread();
　　　　　//一直阻塞直到找到master節(jié)點(diǎn)，在集群剛剛啟動，或者集群master丟失的情況，這種阻塞能夠保證集群一致性
        while (masterNode == null && joinThreadControl.joinThreadActive(currentThread)) {
            masterNode = findMaster();
        }
　　　　　　//有可能自己會被選舉為master（集群啟動，或者加入時正在選舉）
　　　　　　if (clusterService.localNode().equals(masterNode)) {
　　　　　　//如果本身是master，則需要向其它所有節(jié)點(diǎn)發(fā)送集群狀態(tài)更新
            clusterService.submitStateUpdateTask("zen-disco-join (elected_as_master)", Priority.IMMEDIATE, new ProcessedClusterStateNonMasterUpdateTask() {
                @Override
                public ClusterState execute(ClusterState currentState) {
　　　　　　　　　　　　//選舉時錯誤的，之前的master狀態(tài)良好，則不更新狀態(tài)，仍舊使用之前狀態(tài)。
                    if (currentState.nodes().masterNode() != null) {
                       return currentState;
                    }
                    DiscoveryNodes.Builder builder = new DiscoveryNodes.Builder(currentState.nodes()).masterNodeId(currentState.nodes().localNode().id());
                    // update the fact that we are the master...
                    ClusterBlocks clusterBlocks = ClusterBlocks.builder().blocks(currentState.blocks()).removeGlobalBlock(discoverySettings.getNoMasterBlock()).build();
                    currentState = ClusterState.builder(currentState).nodes(builder).blocks(clusterBlocks).build();
                    // eagerly run reroute to remove dead nodes from routing table
                    RoutingAllocation.Result result = allocationService.reroute(currentState);
                    return ClusterState.builder(currentState).routingResult(result).build();
                }
                @Override
                public void onFailure(String source, Throwable t) {
                    logger.error("unexpected failure during [{}]", t, source);
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                }
                @Override
                public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
                    if (newState.nodes().localNodeMaster()) {
                        // we only starts nodesFD if we are master (it may be that we received a cluster state while pinging)
                        joinThreadControl.markThreadAsDone(currentThread);
                        nodesFD.updateNodesAndPing(newState); // start the nodes FD
                    } else {
                        // if we're not a master it means another node published a cluster state while we were pinging
                        // make sure we go through another pinging round and actively join it
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                    }
                    sendInitialStateEventIfNeeded();
                    long count = clusterJoinsCounter.incrementAndGet();
                    logger.trace("cluster joins counter set to [{}] (elected as master)", count);
                }
            });
        } else {
            // 找到的節(jié)點(diǎn)不是我，試圖連接該master
            final boolean success = joinElectedMaster(masterNode);
            // finalize join through the cluster state update thread
            final DiscoveryNode finalMasterNode = masterNode;
            clusterService.submitStateUpdateTask("finalize_join (" + masterNode + ")", new ClusterStateNonMasterUpdateTask() {
                @Override
                public ClusterState execute(ClusterState currentState) throws Exception {
                    if (!success) {
                        // failed to join. Try again...
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                        return currentState;
                    }
                    if (currentState.getNodes().masterNode() == null) {
                        // Post 1.3.0, the master should publish a new cluster state before acking our join request. we now should have
                        // a valid master.
                        logger.debug("no master node is set, despite of join request completing. retrying pings.");
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                        return currentState;
                    }
                    if (!currentState.getNodes().masterNode().equals(finalMasterNode)) {
                        return joinThreadControl.stopRunningThreadAndRejoin(currentState, "master_switched_while_finalizing_join");
                    }
                    // Note: we do not have to start master fault detection here because it's set at {@link #handleNewClusterStateFromMaster }
                    // when the first cluster state arrives.
                    joinThreadControl.markThreadAsDone(currentThread);
                    return currentState;
                }
                @Override
                public void onFailure(String source, @Nullable Throwable t) {
                    logger.error("unexpected error while trying to finalize cluster join", t);
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                }
            });
        }
    }

private DiscoveryNode findMaster() {
　　　　　　//ping集群中的節(jié)點(diǎn)
        ZenPing.PingResponse[] fullPingResponses = pingService.pingAndWait(pingTimeout);
        if (fullPingResponses == null) {return null;
        }// 過濾所得到的ping響應(yīng)，慮除client節(jié)點(diǎn)，單純的data節(jié)點(diǎn)
        List<ZenPing.PingResponse> pingResponses = Lists.newArrayList();
        for (ZenPing.PingResponse pingResponse : fullPingResponses) {
            DiscoveryNode node = pingResponse.node();
            if (masterElectionFilterClientNodes && (node.clientNode() || (!node.masterNode() && !node.dataNode()))) {
                // filter out the client node, which is a client node, or also one that is not data and not master (effectively, client)
            } else if (masterElectionFilterDataNodes && (!node.masterNode() && node.dataNode())) {
                // filter out data node that is not also master
            } else {
                pingResponses.add(pingResponse);
            }
        }
       final DiscoveryNode localNode = clusterService.localNode();
        List<DiscoveryNode> pingMasters = newArrayList();
　　　　　//獲取所有ping響應(yīng)中的master節(jié)點(diǎn)，如果master節(jié)點(diǎn)是節(jié)點(diǎn)本身則過濾掉。pingMasters列表結(jié)果要么為空（本節(jié)點(diǎn)是master）要么是同一個節(jié)點(diǎn)（出現(xiàn)不同節(jié)點(diǎn)則集群出現(xiàn)了問題
不過沒關(guān)系，后面會進(jìn)行選舉）
        for (ZenPing.PingResponse pingResponse : pingResponses) {
            if (pingResponse.master() != null) {
                if (!localNode.equals(pingResponse.master())) {
                    pingMasters.add(pingResponse.master());
                }
            }
        }
        // nodes discovered during pinging
        Set<DiscoveryNode> activeNodes = Sets.newHashSet();
        // nodes discovered who has previously been part of the cluster and do not ping for the very first time
        Set<DiscoveryNode> joinedOnceActiveNodes = Sets.newHashSet();
        Version minimumPingVersion = localNode.version();
　　　　for (ZenPing.PingResponse pingResponse : pingResponses) {
   　　　　 activeNodes.add(pingResponse.node());
    　　　　minimumPingVersion = Version.smallest(pingResponse.node().version(), minimumPingVersion);
    　　　　if (pingResponse.hasJoinedOnce() != null && pingResponse.hasJoinedOnce()) {
        　　joinedOnceActiveNodes.add(pingResponse.node());
    　　　　}
　　　　}

//本節(jié)點(diǎn)暫時是master也要加入候選節(jié)點(diǎn)進(jìn)行選舉
        if (localNode.masterNode()) {
            activeNodes.add(localNode);
            long joinsCounter = clusterJoinsCounter.get();
            if (joinsCounter > 0) {
                logger.trace("adding local node to the list of active nodes who has previously joined the cluster (joins counter is [{}})", joinsCounter);
                joinedOnceActiveNodes.add(localNode);
            }
        }
　　　　　　//pingMasters為空，則本節(jié)點(diǎn)是master節(jié)點(diǎn)，
　　　　if (pingMasters.isEmpty()) {
            if (electMaster.hasEnoughMasterNodes(activeNodes)) {//保證選舉數(shù)量,說明有足夠多的節(jié)點(diǎn)選舉本節(jié)點(diǎn)為master，但是這還不夠，本節(jié)點(diǎn)還需要再選舉一次，如果
　　　　　　　　　　本次選舉節(jié)點(diǎn)仍舊是自己，那么本節(jié)點(diǎn)才能成為master。這里就體現(xiàn)了master選舉的第二條原則。
                DiscoveryNode master = electMaster.electMaster(joinedOnceActiveNodes);
                if (master != null) {
                    return master;
                }
                return electMaster.electMaster(activeNodes);
            } else {
                // if we don't have enough master nodes, we bail, because there are not enough master to elect from
                logger.trace("not enough master nodes [{}]", activeNodes);
                return null;
            }
        } else {
　　　　　　　　//pingMasters不為空（pingMasters列表中應(yīng)該都是同一個節(jié)點(diǎn)），本節(jié)點(diǎn)沒有被選舉為master，那就接受之前的選舉。
            return electMaster.electMaster(pingMasters);
        }
    }

clusterService.submitStateUpdateTask("zen-disco-master_failed (" + masterNode + ")", Priority.IMMEDIATE,?new?ProcessedClusterStateNonMasterUpdateTask() {           
 　　　　　　@Override
            public ClusterState execute(ClusterState currentState) {
                //獲取到當(dāng)前集群狀態(tài)下的所有節(jié)點(diǎn)
                DiscoveryNodes discoveryNodes = DiscoveryNodes.builder(currentState.nodes())
                        // make sure the old master node, which has failed, is not part of the nodes we publish
                        .remove(masterNode.id())
                        .masterNodeId(null).build();
　　　　　　　　　　//rejoin過程仍然是重復(fù)findMaster過程
　　　　　　　　　　if (rejoin) {
                    return rejoin(ClusterState.builder(currentState).nodes(discoveryNodes).build(), "master left (reason = " + reason + ")");
                }
　　　　　　　　　　//無法達(dá)到選舉數(shù)量，進(jìn)行findMaster過程
                if (!electMaster.hasEnoughMasterNodes(discoveryNodes)) {
                    return rejoin(ClusterState.builder(currentState).nodes(discoveryNodes).build(), "not enough master nodes after master left (reason = " + reason + ")");
                }
　　　　　　　　　　//在當(dāng)前集群狀態(tài)下，如果候選節(jié)點(diǎn)數(shù)量達(dá)到預(yù)期數(shù)量，那么選舉出來的節(jié)點(diǎn)一定是同一個節(jié)點(diǎn)，因?yàn)樗械墓?jié)點(diǎn)看到的集群states是一致的
                final DiscoveryNode electedMaster = electMaster.electMaster(discoveryNodes); // elect master
                final DiscoveryNode localNode = currentState.nodes().localNode();
              ....
            }