protected void masterOperation(final PutMappingRequest request, final ClusterState state, final ActionListener<PutMappingResponse> listener) throws ElasticsearchException {final String[] concreteIndices = clusterService.state().metaData().concreteIndices(request.indicesOptions(), request.indices());　　　　　　//构造requestPutMappingClusterStateUpdateRequest updateRequest = new PutMappingClusterStateUpdateRequest().ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout()).indices(concreteIndices).type(request.type()).source(request.source()).ignoreConflicts(request.ignoreConflicts());//调用putMapping方法，同时传入一个ListenermetaDataMappingService.putMapping(updateRequest, new ActionListener<ClusterStateUpdateResponse>() {@Overridepublic void onResponse(ClusterStateUpdateResponse response) {listener.onResponse(new PutMappingResponse(response.isAcknowledged()));}@Overridepublic void onFailure(Throwable t) {logger.debug("failed to put mappings on indices [{}], type [{}]", t, concreteIndices, request.type());listener.onFailure(t);}});}

 public void putMapping(final PutMappingClusterStateUpdateRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {//提交一个高基本的updateTaskclusterService.submitStateUpdateTask("put-mapping [" + request.type() + "]", Priority.HIGH, new AckedClusterStateUpdateTask<ClusterStateUpdateResponse>(request, listener) {@Overrideprotected ClusterStateUpdateResponse newResponse(boolean acknowledged) {return new ClusterStateUpdateResponse(acknowledged);}@Overridepublic ClusterState execute(final ClusterState currentState) throws Exception {List<String> indicesToClose = Lists.newArrayList();try {　　　　　　　　　　　　//必须针对已经在matadata中存在的index，否则抛出异常for (String index : request.indices()) {if (!currentState.metaData().hasIndex(index)) {throw new IndexMissingException(new Index(index));}}//还需要存在于indices中，否则无法进行操作。所以这里要进行预建for (String index : request.indices()) {if (indicesService.hasIndex(index)) {continue;}final IndexMetaData indexMetaData = currentState.metaData().index(index);　　　　　　　　　　　　　　//不存在就进行创建IndexService indexService = indicesService.createIndex(indexMetaData.index(), indexMetaData.settings(), clusterService.localNode().id());indicesToClose.add(indexMetaData.index());// make sure to add custom default mapping if existsif (indexMetaData.mappings().containsKey(MapperService.DEFAULT_MAPPING)) {indexService.mapperService().merge(MapperService.DEFAULT_MAPPING, indexMetaData.mappings().get(MapperService.DEFAULT_MAPPING).source(), false);}// only add the current relevant mapping (if exists)if (indexMetaData.mappings().containsKey(request.type())) {indexService.mapperService().merge(request.type(), indexMetaData.mappings().get(request.type()).source(), false);}}//合并更新MappingMap<String, DocumentMapper> newMappers = newHashMap();Map<String, DocumentMapper> existingMappers = newHashMap();　　　　　　　　　　　　//针对每个index进行Mapping合并for (String index : request.indices()) {IndexService indexService = indicesService.indexServiceSafe(index);// try and parse it (no need to add it here) so we can bail early in case of parsing exceptionDocumentMapper newMapper;DocumentMapper existingMapper = indexService.mapperService().documentMapper(request.type());if (MapperService.DEFAULT_MAPPING.equals(request.type())) {//存在defaultmapping则合并default mapping// _default_ types do not go through merging, but we do test the new settings. Also don't apply the old defaultnewMapper = indexService.mapperService().parse(request.type(), new CompressedString(request.source()), false);} else {newMapper = indexService.mapperService().parse(request.type(), new CompressedString(request.source()), existingMapper == null);if (existingMapper != null) {// first, simulateDocumentMapper.MergeResult mergeResult = existingMapper.merge(newMapper, mergeFlags().simulate(true));// if we have conflicts, and we are not supposed to ignore them, throw an exceptionif (!request.ignoreConflicts() && mergeResult.hasConflicts()) {throw new MergeMappingException(mergeResult.conflicts());}}}newMappers.put(index, newMapper);if (existingMapper != null) {existingMappers.put(index, existingMapper);}}String mappingType = request.type();if (mappingType == null) {mappingType = newMappers.values().iterator().next().type();} else if (!mappingType.equals(newMappers.values().iterator().next().type())) {throw new InvalidTypeNameException("Type name provided does not match type name within mapping definition");}if (!MapperService.DEFAULT_MAPPING.equals(mappingType) && !PercolatorService.TYPE_NAME.equals(mappingType) && mappingType.charAt(0) == '_') {throw new InvalidTypeNameException("Document mapping type name can't start with '_'");}final Map<String, MappingMetaData> mappings = newHashMap();for (Map.Entry<String, DocumentMapper> entry : newMappers.entrySet()) {String index = entry.getKey();// do the actual merge here on the master, and update the mapping sourceDocumentMapper newMapper = entry.getValue();IndexService indexService = indicesService.indexService(index);if (indexService == null) {continue;}CompressedString existingSource = null;if (existingMappers.containsKey(entry.getKey())) {existingSource = existingMappers.get(entry.getKey()).mappingSource();}DocumentMapper mergedMapper = indexService.mapperService().merge(newMapper.type(), newMapper.mappingSource(), false);CompressedString updatedSource = mergedMapper.mappingSource();if (existingSource != null) {if (existingSource.equals(updatedSource)) {// same source, no changes, ignore it} else {// use the merged mapping sourcemappings.put(index, new MappingMetaData(mergedMapper));if (logger.isDebugEnabled()) {logger.debug("[{}] update_mapping [{}] with source [{}]", index, mergedMapper.type(), updatedSource);} else if (logger.isInfoEnabled()) {logger.info("[{}] update_mapping [{}]", index, mergedMapper.type());}}} else {mappings.put(index, new MappingMetaData(mergedMapper));if (logger.isDebugEnabled()) {logger.debug("[{}] create_mapping [{}] with source [{}]", index, newMapper.type(), updatedSource);} else if (logger.isInfoEnabled()) {logger.info("[{}] create_mapping [{}]", index, newMapper.type());}}}if (mappings.isEmpty()) {// no changes, returnreturn currentState;}//根据mapping的更新情况重新生成matadataMetaData.Builder builder = MetaData.builder(currentState.metaData());for (String indexName : request.indices()) {IndexMetaData indexMetaData = currentState.metaData().index(indexName);if (indexMetaData == null) {throw new IndexMissingException(new Index(indexName));}MappingMetaData mappingMd = mappings.get(indexName);if (mappingMd != null) {builder.put(IndexMetaData.builder(indexMetaData).putMapping(mappingMd));}}return ClusterState.builder(currentState).metaData(builder).build();} finally {for (String index : indicesToClose) {indicesService.removeIndex(index, "created for mapping processing");}}}});}