public Engine.CommitId flush(FlushRequest request) throws ElasticsearchException {
    boolean waitIfOngoing = request.waitIfOngoing();
    boolean force = request.force();
    if (logger.isTraceEnabled()) {
        logger.trace("flush with {}", request);
    }
    // we allows flush while recovering, since we allow for operations to happen
    // while recovering, and we want to keep the translog at bay (up to deletes, which
    // we don't gc).
    verifyStartedOrRecovering();

    long time = System.nanoTime();
    Engine.CommitId commitId = engine().flush(force, waitIfOngoing);
    flushMetric.inc(System.nanoTime() - time);
    return commitId;

}
 

@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    FlushRequest flushRequest = new FlushRequest(Strings.splitStringByCommaToArray(request.param("index")));
    flushRequest.indicesOptions(IndicesOptions.fromRequest(request, flushRequest.indicesOptions()));
    flushRequest.force(request.paramAsBoolean("force", flushRequest.force()));
    flushRequest.waitIfOngoing(request.paramAsBoolean("wait_if_ongoing", flushRequest.waitIfOngoing()));
    client.admin().indices().flush(flushRequest, new RestBuilderListener<FlushResponse>(channel) {
        @Override
        public RestResponse buildResponse(FlushResponse response, XContentBuilder builder) throws Exception {
            builder.startObject();
            buildBroadcastShardsHeader(builder, request, response);
            builder.endObject();
            return new BytesRestResponse(OK, builder);
        }
    });
}
 

/**
 * Executes the given flush request against the engine.
 *
 * @param request the flush request
 * @return the commit ID
 */
public Engine.CommitId flush(FlushRequest request) {
    final boolean waitIfOngoing = request.waitIfOngoing();
    final boolean force = request.force();
    logger.trace("flush with {}", request);
    /*
     * We allow flushes while recovery since we allow operations to happen while recovering and we want to keep the translog under
     * control (up to deletes, which we do not GC). Yet, we do not use flush internally to clear deletes and flush the index writer
     * since we use Engine#writeIndexingBuffer for this now.
     */
    verifyNotClosed();
    final Engine engine = getEngine();
    if (engine.isRecovering()) {
        throw new IllegalIndexShardStateException(
                shardId(),
                state,
                "flush is only allowed if the engine is not recovery from translog");
    }
    final long time = System.nanoTime();
    final Engine.CommitId commitId = engine.flush(force, waitIfOngoing);
    engine.refresh("flush"); // TODO this is technically wrong we should remove this in 7.0
    flushMetric.inc(System.nanoTime() - time);
    return commitId;
}
 

private PreSyncedFlushResponse performPreSyncedFlush(PreShardSyncedFlushRequest request) {
    IndexShard indexShard = indicesService.indexServiceSafe(request.shardId().getIndex()).shardSafe(request.shardId().id());
    FlushRequest flushRequest = new FlushRequest().force(false).waitIfOngoing(true);
    logger.trace("{} performing pre sync flush", request.shardId());
    Engine.CommitId commitId = indexShard.flush(flushRequest);
    logger.trace("{} pre sync flush done. commit id {}", request.shardId(), commitId);
    return new PreSyncedFlushResponse(commitId);
}
 

public long getCount(ElasticConfiguration elastic, YamlConfiguration config) {
    IndicesAdminClient admin = elastic.getClient().admin().indices();
    IndicesExistsRequestBuilder builder = admin.prepareExists(config.getMisc().getDindex().getAs());
    assertThat(builder.execute().actionGet().isExists(), is(true));

    elastic.getClient().admin().indices().flush(new FlushRequest(config.getMisc().getDindex().getAs())).actionGet();

    SearchResponse response = elastic.getClient().prepareSearch(config.getMisc().getDindex().getAs())
            .setTypes(config.getMisc().getCtype().getAs())
            .setSearchType(SearchType.QUERY_THEN_FETCH)
            .setSize(0)
            .execute().actionGet();
    long count = response.getHits().getTotalHits();
    return count;
}
 

/**
 * 仅仅只删除索引
 * @param index
 * @param type
 * @param id
 */
private static void deleteIndex(String index, String type, String id){
	Client client = createTransportClient();
	DeleteResponse response = client.prepareDelete(index, type, id)
			.execute()
			.actionGet();
	boolean isFound = response.isFound();
	System.out.println("索引是否 存在:"+isFound); // 发现doc已删除则返回true
	System.out.println("****************index ***********************");
	// Index name
	String _index = response.getIndex();
	// Type name
	String _type = response.getType();
	// Document ID (generated or not)
	String _id = response.getId();
	// Version (if it's the first time you index this document, you will get: 1)
	long _version = response.getVersion();
	System.out.println(_index+","+_type+","+_id+","+_version);
	
	//优化索引
	OptimizeRequest optimizeRequest = new OptimizeRequest(index);
    OptimizeResponse optimizeResponse = client.admin().indices().optimize(optimizeRequest).actionGet();
    System.out.println(optimizeResponse.getTotalShards()+","+optimizeResponse.getSuccessfulShards()+","+optimizeResponse.getFailedShards());
    
    //刷新索引
	FlushRequest flushRequest = new FlushRequest(index);
	flushRequest.force(true);
	FlushResponse flushResponse = client.admin().indices().flush(flushRequest).actionGet();
	System.out.println(flushResponse.getTotalShards()+","+flushResponse.getSuccessfulShards()+","+flushResponse.getFailedShards());
	
}
 

public void flushIndex(String index) {
    if (client() == null) {
        return;
    }
    if (index != null) {
        client().execute(FlushAction.INSTANCE, new FlushRequest(index)).actionGet();
    }
}
 

@Override
protected XContentBuilder toXContent(FlushRequest request, FlushResponse response, XContentBuilder builder) throws IOException {
    builder.startObject();
    builder.field(Fields.OK, true);
    buildBroadcastShardsHeader(builder, response);
    builder.endObject();
    return builder;
}
 

private PreSyncedFlushResponse performPreSyncedFlush(PreShardSyncedFlushRequest request) {
    IndexShard indexShard = indicesService.indexServiceSafe(request.shardId().getIndex()).getShard(request.shardId().id());
    FlushRequest flushRequest = new FlushRequest().force(false).waitIfOngoing(true);
    LOGGER.trace("{} performing pre sync flush", request.shardId());
    indexShard.flush(flushRequest);
    final CommitStats commitStats = indexShard.commitStats();
    final Engine.CommitId commitId = commitStats.getRawCommitId();
    LOGGER.trace("{} pre sync flush done. commit id {}, num docs {}", request.shardId(), commitId, commitStats.getNumDocs());
    return new PreSyncedFlushResponse(commitId, commitStats.getNumDocs(), commitStats.syncId());
}
 

@Override
public ActionFuture<FlushResponse> flush(final FlushRequest request) {
    return execute(FlushAction.INSTANCE, request);
}
 

@Override
public void flush(final FlushRequest request, final ActionListener<FlushResponse> listener) {
    execute(FlushAction.INSTANCE, request, listener);
}
 

/**
 * 设置mapping
 * 
 * @return
 */
private void buildMapping() {
    try {
        Fileds fileds = FieldXMLParser.getAndCache().get(indexType.getIndexNo());
        contentBuilder = XContentFactory.jsonBuilder();
        contentBuilder.startObject().startObject(indexType.getTypeName());
        // 如果有主键,用主键作为新建索引id,没有主键让es自动生成主键
        if (StringUtils.isNotBlank(fileds.getKey())) {
            String idname = fileds.getKey().toUpperCase();
            contentBuilder.startObject("_id").field("path", idname).endObject();
        }
        // 压缩
        contentBuilder.startObject("_source").field("compress", true).endObject();
        // 开启_all
        contentBuilder.startObject("_all").field("enabled", true).endObject();
        contentBuilder.startObject("properties");

        List<Filed> listfiled = fileds.getListfiled();
        for (Filed filed : listfiled) {
            // 处理类型映射
            String type_ = getFiledType(filed);
            if ("-1".equals(type_)) {
                continue;
            } else if ("date".equals(type_)) {
                String format = filed.getFormat();
                if (StringUtils.isBlank(format)) {
                    format = "yyyy-MM-dd HH:mm:ss";
                }
                contentBuilder.startObject(filed.getNameToUpperCase()).field("type", type_)
                        .field("format", format)
                        .field("store", filed.isIsstore() == true ? "yes" : "no")
                        .field("index", "not_analyzed")
                        .field("include_in_all", filed.isIsdefaultsearch()).endObject();
                continue;
            }

            // mapping基本配置
            contentBuilder.startObject(filed.getNameToUpperCase()).field("type", type_)
                    .field("store", filed.isIsstore() == true ? "yes" : "no")
                    .field("include_in_all", filed.isIsdefaultsearch()); //
            // 是否copyto
            if (filed.isIscopy()) {
                contentBuilder.field("copy_to", filed.getCopyto());
            }
            // 是否设置有权重
            if (filed.getWeight() > 0d) {
                contentBuilder.field("boost", filed.getWeight());
            }
            // 多值字段的边界分割
            if (filed.getPosition_offset_gap() > 0) {
                contentBuilder.field("position_offset_gap", filed.getPosition_offset_gap());
            }

            // 设置索引分词方式
            if (!filed.isIsindex()) {
                contentBuilder.field("index", "no");
            } else {
                if (!filed.isAnalyzer()) {
                    contentBuilder.field("index", "not_analyzed");
                } else {
                    String indexAnalyzer = filed.getIndexAnalyzer();
                    String searchAnalyzer = filed.getSearchAnalyzer();
                    if (StringUtils.isBlank(searchAnalyzer)) {
                        searchAnalyzer = indexAnalyzer;
                    }
                    contentBuilder.field("indexAnalyzer", indexAnalyzer).field(
                            "searchAnalyzer", searchAnalyzer);
                }
            }
            contentBuilder.endObject();
        }

        // 建议器
        if (fileds.getSuggest() != null) {
            contentBuilder.startObject(fileds.getSuggest().key)
                    .field("type", fileds.getSuggest().type)
                    .field("index_analyzer", fileds.getSuggest().getIndexAnalyzer())
                    .field("search_analyzer", fileds.getSuggest().getSearchAnalyzer())
                    .field("payloads", fileds.getSuggest().isPayloads()).endObject();
        }

        // 构造mapping请求
        PutMappingRequest mappingRequest = Requests.putMappingRequest(currentIndexName)
                .type(indexType.getTypeName()).source(contentBuilder.endObject().endObject());
        ESClient.getClient().admin().indices().putMapping(mappingRequest).actionGet();
        ESClient.getClient().admin().indices().flush(new FlushRequest(currentIndexName))
                .actionGet();
        log.debug("create mappings. index:{},type:{},mapping:{}", currentIndexName,
                indexType.getTypeName(), contentBuilder.string());
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}
 

/**
 * flushを行う.
 * @param index flush対象のindex名
 * @return 非同期応答
 */
public ActionFuture<FlushResponse> flushTransLog(String index) {
    ActionFuture<FlushResponse> ret = esTransportClient.admin().indices().flush(new FlushRequest(index));
    this.fireEvent(Event.afterRequest, index, null, null, null, "Flush");
    return ret;
}
 

/**
 * flushを行う.
 * @param index flush対象のindex名
 * @return 非同期応答
 */
public ActionFuture<FlushResponse> flushTransLog(String index) {
    ActionFuture<FlushResponse> ret = esTransportClient.admin().indices().flush(new FlushRequest(index));
    this.fireEvent(Event.afterRequest, index, null, null, null, "Flush");
    return ret;
}
 

public FlushRequestBuilder(Client client, JsonToString<JsonInput> jsonToString, StringToJson<JsonOutput> stringToJson) {
    super(client, new FlushRequest(), jsonToString, stringToJson);
}
 

@Override
protected ActionFuture<FlushResponse> doExecute(FlushRequest request) {
    return client.admin().indices().flush(request);
}
 

protected void flushShard(IndexShard shard, boolean force) {
    shard.flush(new FlushRequest(shard.shardId().getIndexName()).force(force));
}
 

/**
 * Explicitly flush one or more indices (releasing memory from the node).
 *
 * @param request The flush request
 * @return A result future
 * @see org.elasticsearch.client.Requests#flushRequest(String...)
 */
ActionFuture<FlushResponse> flush(FlushRequest request);
 

/**
 * Explicitly flush one or more indices (releasing memory from the node).
 *
 * @param request  The flush request
 * @param listener A listener to be notified with a result
 * @see org.elasticsearch.client.Requests#flushRequest(String...)
 */
void flush(FlushRequest request, ActionListener <FlushResponse> listener);
 

/**
 * Creates a flush indices request.
 *
 * @param indices The indices to flush. Use <tt>null</tt> or <tt>_all</tt> to execute against all indices
 * @return The flush request
 * @see org.elasticsearch.client.IndicesAdminClient#flush(org.elasticsearch.action.admin.indices.flush.FlushRequest)
 */
public static FlushRequest flushRequest(String... indices) {
    return new FlushRequest(indices);
}
 

/**
 * Creates a flush indices request.
 *
 * @param indices The indices to flush. Use {@code null} or {@code _all} to execute against all indices
 * @return The flush request
 * @see org.elasticsearch.client.IndicesAdminClient#flush(org.elasticsearch.action.admin.indices.flush.FlushRequest)
 */
public static FlushRequest flushRequest(String... indices) {
    return new FlushRequest(indices);
}