《Elasticsearch 源码解析与优化实战》第11章：gateway 模块分析

translog：/Data/apps/search/datas/nodes/0/indices/2A1sfea4RQmGiN3duyF5xw/0/translog/translog-1.tlog

集群层面：/Data/apps/search/datas/nodes/0/_state/global-2231.st  node-2.st
索引层面：/Data/apps/search/datas/nodes/0/indices/2A1sfea4RQmGiN3duyF5xw/_state/state-1.st
分片层面：/Data/apps/search/datas/nodes/0/indices/2A1sfea4RQmGiN3duyF5xw/0/_state/state-1.st

private void callClusterStateAppliers (ClusterChangedEvent clusterChangedEvent) {
    //遍历全部的applier,依次调用各模块对集群状态的处理
    clusterStateAppliers.forEach (applier -> {
        try {
            //调用各模块实现的applyClusterState
            applier. applyClusterState (clusterChangedEvent);
        } catch (Exception ex) {
            //某个模块应用集群状态出现异常时打印日志，但应用过程不会终止
            logger .warn("failed to notify ClusterStateApplier", ex);
        });
}

public void appl applyClusterState(ClusterChangedEvent event) {
    //只有具备Master资格的节点和数据节点才会持久化元数据
    if (state.nodes().getLocalNode().isMasterNode() || state.nodes().getLocalNode().isDataNode()) {
        //检查全局元信息是否发生变化，如果有变化，则将其写入磁盘
        if (previousMetaData == nulll || !MetaData.isGlobalStateEquals(previousMetaData, newMetaData)) {
            metaStateService.writeGlobalState ("changed", newMetaData);
        }
        //将发生变化的元信息写磁盘
        for (IndexMetaWriteinfo indexMetaWrite : writeinfo) { 
            metaStateServ.writeindex(indexMetaWrite.reason, indexMetaWrite.newMetaData);
        }
    }
}

//临时文件名
final Path tmpStatePath = stateLocation. resolve (fileName + ". tmp");

//目标文件名
final Path finalStatePath = stateLocation. resolve (fileName);

//写临时文件
OutputStreamIndexOutput out = new OutputStreamIndexOutput (Files.newOutputStream(tmpStatePath), ...);
out.writeInt(format.index());

//从系统cache刷到磁盘中，保证持久化
IOUtils.fsync(tmpStatePath, false); // fsync the state file

//move为目标文件，move操作为系统原子操作
Files.move (tmpStatePath, finalStatePath, StandardCopyOption.ATOMIC_MOVE);

//具有Master资格的节点列表
String[] nodesIds = clusterService.state().nodes().getMasterNodes().keys().toArray(String.class);
//发送获取请求并等待结果
TransportNodesListGatewayMetaState.NodesGatewayMetaState nodesState = listGatewayMetaState.list(nodesIds, null).actionGet();

int requiredAllocation = Math.max(1, minimumMasterNodesProvider.get());

minimumMasterNodesProvider的值由下面的配置项决定：discovery.zen.minimum_master_nodes

选举集群级元数据: 
public void performStateRecovery(...) {
    //遍历请求的所有节点
    for (TransportNodesListGatewayMetaState.NodeGatewayMetaState nodeState : nodesState.getNode ()) {
        //根据元信息中记录的版本号选举元信息
        if (electedGlobalState == null) {
            electedGlobalState = nodeState . metaData() ;
        } else if (nodeState.metaData().version() > electedGlobalState.version()) {
            electedGlobalState = nodeState .metaData() ;
        }
    }
}

选举索引级元数据:
public void performStateRecovery(. ..) {
    final Object[] keys = indices.keys;
    //遍历集群的全部索引
    for (int i = 0; i < keys. length; i++) {
        if (keys[i] != null) {
            Index index = (Index) keys[i];
            IndexMetaData electedIndexMetaData = null;
            //遍历请求的全部节点，对特定索引选择版本号最高的作为该索引元数据
            for (TransportNodesListGatewayMetaState.NodeGa tewayMetaStatenodeState : nodesState.getNodes()) {
                IndexMetaData indexMetaData = nodeState.metaData ().index(index);
                if (electedIndexMetaData == null) {
                    electedIndexMetaData = indexMetaData;
                } else if (indexMetaData.getVersion() > electedIndexMetaData.getVersion()) {
                    electedIndexMetaData = indexMetaData;
                }
            }
        }
    }
}