elasticsearch index 之 create index（二）

建立索引須要建立索引而且更新集羣index matedata，這一過程在MetaDataCreateIndexService的createIndex方法中完成。這裏會提交一個高優先級，AckedClusterStateUpdateTask類型的task。索引建立須要即時獲得反饋，異常這個task須要返回，會超時，並且這個任務的優先級也很是高。下面具體看一下它的execute方法，這個方法會在master執行任務時調用，這個方法很是長，主要完成如下三個功能：更新合併request，template中的mapping和setting，調用indiceService建立索引，對建立後的索引添加mapping。這一系列功能完成後，合併完成後生成新的matedata，並更新集羣狀態，完成了索引的建立。具體的調用方法參考上一篇。代碼以下所示：

            @Override
            public ClusterState execute(ClusterState currentState) throws Exception {
                boolean indexCreated = false;
                String removalReason = null;
                try {
　　　　　　　　　　　　//檢查request的合法性，1.5版本主要檢查index名字是否合法，如不能含有某些字符，另外就是集羣節點版本
                    validate(request, currentState);
                    for (Alias alias : request.aliases()) {//檢查別稱是否合法
                        aliasValidator.validateAlias(alias, request.index(), currentState.metaData());
                    }

                    // 查找索引模板
                    List<IndexTemplateMetaData> templates = findTemplates(request, currentState, indexTemplateFilter);
                    Map<String, Custom> customs = Maps.newHashMap();
                    // add the request mapping
                    Map<String, Map<String, Object>> mappings = Maps.newHashMap();
                    Map<String, AliasMetaData> templatesAliases = Maps.newHashMap();
                    List<String> templateNames = Lists.newArrayList();
　　　　　　　　　　　　//取出request中的mapping配置，雖然mapping能夠後面添加，多數狀況建立索引的時候仍是會附帶着mapping，在request中mapping是一個map
                    for (Map.Entry<String, String> entry : request.mappings().entrySet()) {
                        mappings.put(entry.getKey(), parseMapping(entry.getValue()));
                    }
　　　　　　　　　　　　//一些預設如warm等
                    for (Map.Entry<String, Custom> entry : request.customs().entrySet()) {
                        customs.put(entry.getKey(), entry.getValue());
                    }
                    // 將找到的template和request中的mapping合併
                    for (IndexTemplateMetaData template : templates) {
                        templateNames.add(template.getName());
                        for (ObjectObjectCursor<String, CompressedString> cursor : template.mappings()) {
                            if (mappings.containsKey(cursor.key)) {
                                XContentHelper.mergeDefaults(mappings.get(cursor.key), parseMapping(cursor.value.string()));
                            } else {
                                mappings.put(cursor.key, parseMapping(cursor.value.string()));
                            }
                        }
                        // 合併custom
                        for (ObjectObjectCursor<String, Custom> cursor : template.customs()) {
                            String type = cursor.key;
                            IndexMetaData.Custom custom = cursor.value;
                            IndexMetaData.Custom existing = customs.get(type);
                            if (existing == null) {
                                customs.put(type, custom);
                            } else {
                                IndexMetaData.Custom merged = IndexMetaData.lookupFactorySafe(type).merge(existing, custom);
                                customs.put(type, merged);
                            }
                        }
                        //處理合並別名
                        for (ObjectObjectCursor<String, AliasMetaData> cursor : template.aliases()) {
                            AliasMetaData aliasMetaData = cursor.value;
                            //if an alias with same name came with the create index request itself,
                            // ignore this one taken from the index template
                            if (request.aliases().contains(new Alias(aliasMetaData.alias()))) {
                                continue;
                            }
                            //if an alias with same name was already processed, ignore this one
                            if (templatesAliases.containsKey(cursor.key)) {
                                continue;
                            }

                            //Allow templatesAliases to be templated by replacing a token with the name of the index that we are applying it to
                            if (aliasMetaData.alias().contains("{index}")) {
                                String templatedAlias = aliasMetaData.alias().replace("{index}", request.index());
                                aliasMetaData = AliasMetaData.newAliasMetaData(aliasMetaData, templatedAlias);
                            }

                            aliasValidator.validateAliasMetaData(aliasMetaData, request.index(), currentState.metaData());
                            templatesAliases.put(aliasMetaData.alias(), aliasMetaData);
                        }
                    }

                    // 合併完template和request，如今開始處理配置基本的mapping，合併邏輯跟以前相同，只是mapping來源不一樣
                    File mappingsDir = new File(environment.configFile(), "mappings");
                    if (mappingsDir.isDirectory()) {
                        // first index level
                        File indexMappingsDir = new File(mappingsDir, request.index());
                        if (indexMappingsDir.isDirectory()) {
                            addMappings(mappings, indexMappingsDir);
                        }

                        // second is the _default mapping
                        File defaultMappingsDir = new File(mappingsDir, "_default");
                        if (defaultMappingsDir.isDirectory()) {
                            addMappings(mappings, defaultMappingsDir);
                        }
                    }
　　　　　　　　　　　　//處理index的配置（setting）
                    ImmutableSettings.Builder indexSettingsBuilder = settingsBuilder();
                    //加入模板中的setting
                    for (int i = templates.size() - 1; i >= 0; i--) {
                        indexSettingsBuilder.put(templates.get(i).settings());
                    }
                    // 加入request中的mapping，request中設置會覆蓋模板中的設置
                    indexSettingsBuilder.put(request.settings());
　　　　　　　　　　　　//處理shard，shard數量不能小於1，所以這裏須要特殊處理，若是沒有則要使用默認值
                    if (request.index().equals(ScriptService.SCRIPT_INDEX)) {
                        indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS, settings.getAsInt(SETTING_NUMBER_OF_SHARDS, 1));
                    } else {
                        if (indexSettingsBuilder.get(SETTING_NUMBER_OF_SHARDS) == null) {
                            if (request.index().equals(riverIndexName)) {
                                indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS, settings.getAsInt(SETTING_NUMBER_OF_SHARDS, 1));
                            } else {
                                indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS, settings.getAsInt(SETTING_NUMBER_OF_SHARDS, 5));
                            }
                        }
                    }
                    if (request.index().equals(ScriptService.SCRIPT_INDEX)) {
                        indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS, settings.getAsInt(SETTING_NUMBER_OF_REPLICAS, 0));
                        indexSettingsBuilder.put(SETTING_AUTO_EXPAND_REPLICAS, "0-all");
                    }
                    else {
                        if (indexSettingsBuilder.get(SETTING_NUMBER_OF_REPLICAS) == null) {
                            if (request.index().equals(riverIndexName)) {
                                indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS, settings.getAsInt(SETTING_NUMBER_OF_REPLICAS, 1));
                            } else {
                                indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS, settings.getAsInt(SETTING_NUMBER_OF_REPLICAS, 1));
                            }
                        }
                    }
　　　　　　　　　　　　//處理副本
                    if (settings.get(SETTING_AUTO_EXPAND_REPLICAS) != null && indexSettingsBuilder.get(SETTING_AUTO_EXPAND_REPLICAS) == null) {
                        indexSettingsBuilder.put(SETTING_AUTO_EXPAND_REPLICAS, settings.get(SETTING_AUTO_EXPAND_REPLICAS));
                    }

                    if (indexSettingsBuilder.get(SETTING_VERSION_CREATED) == null) {
                        DiscoveryNodes nodes = currentState.nodes();
                        final Version createdVersion = Version.smallest(version, nodes.smallestNonClientNodeVersion());
                        indexSettingsBuilder.put(SETTING_VERSION_CREATED, createdVersion);
                    }

                    if (indexSettingsBuilder.get(SETTING_CREATION_DATE) == null) {
                        indexSettingsBuilder.put(SETTING_CREATION_DATE, System.currentTimeMillis());
                    }

                    indexSettingsBuilder.put(SETTING_UUID, Strings.randomBase64UUID());
　　　　　　　　　　　　//建立setting
                    Settings actualIndexSettings = indexSettingsBuilder.build();
　　　　　　　　　　　　// 經過indiceservice建立索引
                    indicesService.createIndex(request.index(), actualIndexSettings, clusterService.localNode().id());
                    indexCreated = true;
                    //若是建立成功這裏就能夠獲取到對應的indexservice，不然會拋出異常
                    IndexService indexService = indicesService.indexServiceSafe(request.index());
　　　　　　　　　　　　//獲取mappingService試圖放置mapping
                    MapperService mapperService = indexService.mapperService();
                    // 爲索引添加mapping，首先是默認mapping
                    if (mappings.containsKey(MapperService.DEFAULT_MAPPING)) {
                        try {
                            mapperService.merge(MapperService.DEFAULT_MAPPING, new CompressedString(XContentFactory.jsonBuilder().map(mappings.get(MapperService.DEFAULT_MAPPING)).string()), false);
                        } catch (Exception e) {
                            removalReason = "failed on parsing default mapping on index creation";
                            throw new MapperParsingException("mapping [" + MapperService.DEFAULT_MAPPING + "]", e);
                        }
                    }
                    for (Map.Entry<String, Map<String, Object>> entry : mappings.entrySet()) {
                        if (entry.getKey().equals(MapperService.DEFAULT_MAPPING)) {
                            continue;
                        }
                        try {
                            // apply the default here, its the first time we parse it
                            mapperService.merge(entry.getKey(), new CompressedString(XContentFactory.jsonBuilder().map(entry.getValue()).string()), true);
                        } catch (Exception e) {
                            removalReason = "failed on parsing mappings on index creation";
                            throw new MapperParsingException("mapping [" + entry.getKey() + "]", e);
                        }
                    }
　　　　　　　　　　　　//添加request中的別稱
                    IndexQueryParserService indexQueryParserService = indexService.queryParserService();
                    for (Alias alias : request.aliases()) {
                        if (Strings.hasLength(alias.filter())) {
                            aliasValidator.validateAliasFilter(alias.name(), alias.filter(), indexQueryParserService);
                        }
                    }
                    for (AliasMetaData aliasMetaData : templatesAliases.values()) {
                        if (aliasMetaData.filter() != null) {
                            aliasValidator.validateAliasFilter(aliasMetaData.alias(), aliasMetaData.filter().uncompressed(), indexQueryParserService);
                        }
                    }

                    // 如下更新Index的matedata，
                    Map<String, MappingMetaData> mappingsMetaData = Maps.newHashMap();
                    for (DocumentMapper mapper : mapperService.docMappers(true)) {
                        MappingMetaData mappingMd = new MappingMetaData(mapper);
                        mappingsMetaData.put(mapper.type(), mappingMd);
                    }

                    final IndexMetaData.Builder indexMetaDataBuilder = IndexMetaData.builder(request.index()).settings(actualIndexSettings);
                    for (MappingMetaData mappingMd : mappingsMetaData.values()) {
                        indexMetaDataBuilder.putMapping(mappingMd);
                    }

                    for (AliasMetaData aliasMetaData : templatesAliases.values()) {
                        indexMetaDataBuilder.putAlias(aliasMetaData);
                    }
                    for (Alias alias : request.aliases()) {
                        AliasMetaData aliasMetaData = AliasMetaData.builder(alias.name()).filter(alias.filter())
                                .indexRouting(alias.indexRouting()).searchRouting(alias.searchRouting()).build();
                        indexMetaDataBuilder.putAlias(aliasMetaData);
                    }

                    for (Map.Entry<String, Custom> customEntry : customs.entrySet()) {
                        indexMetaDataBuilder.putCustom(customEntry.getKey(), customEntry.getValue());
                    }

                    indexMetaDataBuilder.state(request.state());
　　　　　　　　　　　　//matedata更新完畢，build新的matedata
                    final IndexMetaData indexMetaData;
                    try {
                        indexMetaData = indexMetaDataBuilder.build();
                    } catch (Exception e) {
                        removalReason = "failed to build index metadata";
                        throw e;
                    }

                    indexService.indicesLifecycle().beforeIndexAddedToCluster(new Index(request.index()),
                            indexMetaData.settings());
　　　　　　　　　　　　//更新集羣的matedata，將新build的indexmatadata加入到metadata中
                    MetaData newMetaData = MetaData.builder(currentState.metaData())
                            .put(indexMetaData, false)
                            .build();

                    logger.info("[{}] creating index, cause [{}], templates {}, shards [{}]/[{}], mappings {}", request.index(), request.cause(), templateNames, indexMetaData.numberOfShards(), indexMetaData.numberOfReplicas(), mappings.keySet());
　　　　　　　　　　　　//阻塞集羣，更新matadata
                    ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
                    if (!request.blocks().isEmpty()) {
                        for (ClusterBlock block : request.blocks()) {
                            blocks.addIndexBlock(request.index(), block);
                        }
                    }
                    if (request.state() == State.CLOSE) {
                        blocks.addIndexBlock(request.index(), MetaDataIndexStateService.INDEX_CLOSED_BLOCK);
                    }

                    ClusterState updatedState = ClusterState.builder(currentState).blocks(blocks).metaData(newMetaData).build();

                    if (request.state() == State.OPEN) {
                        RoutingTable.Builder routingTableBuilder = RoutingTable.builder(updatedState.routingTable())
                                .addAsNew(updatedState.metaData().index(request.index()));
                        RoutingAllocation.Result routingResult = allocationService.reroute(ClusterState.builder(updatedState).routingTable(routingTableBuilder).build());
                        updatedState = ClusterState.builder(updatedState).routingResult(routingResult).build();
                    }
                    removalReason = "cleaning up after validating index on master";
                    return updatedState;
                } finally {
                    if (indexCreated) {
                        // Index was already partially created - need to clean up
                        indicesService.removeIndex(request.index(), removalReason != null ? removalReason : "failed to create index");
                    }
                }
            }
        });
    }

 以上就是建立index的create方法，方法中主要進行了兩個動做：合併更新index的matadata和建立index。更新合併matadata的過程都在上面的代碼中體現了。建立索引是調用indiceSerivice構建一個guice的injector，這個injector包含了Index的全部功能（如分詞，類似度等）。同時會將其存儲到indiceService中，以一個map的格式存儲Map<String, Tuple<IndexService, Injector>> indices。運行中的集羣每次對某個索引的操做都首先從indice中獲取對應的IndexService。這一部分代碼以下所示：

   public synchronized IndexService createIndex(String sIndexName, @IndexSettings Settings settings, String localNodeId) throws ElasticsearchException {
        if (!lifecycle.started()) {
            throw new ElasticsearchIllegalStateException("Can't create an index [" + sIndexName + "], node is closed");
        }
        Index index = new Index(sIndexName);
　　　　//檢測index是否已經存在
        if (indices.containsKey(index.name())) {
            throw new IndexAlreadyExistsException(index);
        }

        indicesLifecycle.beforeIndexCreated(index, settings);
        logger.debug("creating Index [{}], shards [{}]/[{}]", sIndexName, settings.get(SETTING_NUMBER_OF_SHARDS), settings.get(SETTING_NUMBER_OF_REPLICAS));
        Settings indexSettings = settingsBuilder()
                .put(this.settings)
                .put(settings)
                .classLoader(settings.getClassLoader())
                .build();
　　　　//構建index對應的injector
        ModulesBuilder modules = new ModulesBuilder();
        modules.add(new IndexNameModule(index));
        modules.add(new LocalNodeIdModule(localNodeId));
        modules.add(new IndexSettingsModule(index, indexSettings));
        modules.add(new IndexPluginsModule(indexSettings, pluginsService));
        modules.add(new IndexStoreModule(indexSettings));
        modules.add(new IndexEngineModule(indexSettings));
        modules.add(new AnalysisModule(indexSettings, indicesAnalysisService));
        modules.add(new SimilarityModule(indexSettings));
        modules.add(new IndexCacheModule(indexSettings));
        modules.add(new IndexFieldDataModule(indexSettings));
        modules.add(new CodecModule(indexSettings));
        modules.add(new MapperServiceModule());
        modules.add(new IndexQueryParserModule(indexSettings));
        modules.add(new IndexAliasesServiceModule());
        modules.add(new IndexGatewayModule(indexSettings, injector.getInstance(Gateway.class)));
        modules.add(new IndexModule(indexSettings));
        Injector indexInjector;
        try {
            indexInjector = modules.createChildInjector(injector);
        } catch (CreationException e) {
            throw new IndexCreationException(index, Injectors.getFirstErrorFailure(e));
        } catch (Throwable e) {
            throw new IndexCreationException(index, e);
        }
        IndexService indexService = indexInjector.getInstance(IndexService.class);
        indicesLifecycle.afterIndexCreated(indexService);
　　　　　//將Indexservice和IndexInjector加入到indice map中
        indices = newMapBuilder(indices).put(index.name(), new Tuple<>(indexService, indexInjector)).immutableMap();

        return indexService;
    }

以上方法就是具體建立索引的過程，它是在master上操做的，同時它是同步方法。這樣才能保證集羣的Index建立一致性，所以這也會致使以前所說的大量建立建立索引時候的速度瓶頸。可是建立大量索引的動做是不常見的，須要儘可能避免。建立一個索引對於一個集羣來講就是開啓對於該索引的各類操做，所以這裏經過guice將索引的各個功能模塊注入，並得到index操做的接口類Indexservice。若是這個方法執行成功，則能夠合併template及request中的mapping，而且向剛建立的索引添加合併後的mapping，最後構建新的matadata，並將集羣新的matadata發送給各個節點完成索引建立。

總結：索引建立的過程包括三步：更新集羣matadata，調用indiceService中建立索引，向新建立的索引中放置（合併到Index對應的mappingService中）mapping。這三步都在以上的兩個方法中。完成這三步，集羣中就保存了新索引的信息，同時索引配置和mapping放置也完成。索引就能夠正常使用。