import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
@Override
public void store(RelatedItemStorageLocationMapper indexLocationMapper, List<RelatedItem> relatedItems) {
    BulkRequestBuilder bulkRequest = elasticClient.prepareBulk();
    bulkRequest.setReplicationType(ReplicationType.ASYNC).setRefresh(false);

    int requestAdded = 0;
    for(RelatedItem product : relatedItems) {
        int added = addRelatedItem(indexLocationMapper, bulkRequest, product);
        if(relatedItemsDocumentIndexingEnabled && added==1) {
            added+=addRelatedItemDocument(bulkRequest, product);
        }
        requestAdded+=added;
    }

    if(requestAdded>0) {
        log.info("Sending Relating Product Index Requests to Elastic: {}",requestAdded);
        BulkResponse bulkResponse = bulkRequest.execute().actionGet();

        if(bulkResponse.hasFailures()) {
            log.warn(bulkResponse.buildFailureMessage());
        }
    }
}
 

import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
/**
 * Slower save, but high protected save.
 */
public void save() {
	IndexRequestBuilder create = Elasticsearch.getClient().prepareIndex(ES_INDEX, ES_TYPE, _id).setRefresh(true);
	create.setSource(MyDMAM.gson_kit.getGsonSimple().toJson(this));
	
	if (_id == null) {
		create.setCreate(true).setOpType(OpType.CREATE).setConsistencyLevel(WriteConsistencyLevel.ALL).setReplicationType(ReplicationType.SYNC);
		
		createAutogeneratedId();
		create.setId(_id);
		while (create.get().isCreated() == false) {
			createAutogeneratedId();
			create.setId(_id);
		}
	} else {
		create.get();
	}
}
 

import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
/**
 * Mostly just write the data you got from BigQuery into ElasticSearch.
 * 
 * @throws ElasticsearchException
 * @throws IOException
 * @throws InterruptedException
 */
private void parse(@Nonnull final List<TableRow> rows) throws ElasticsearchException, IOException, InterruptedException {
	final int size = rows.size();
	int count = 0;
	final String timestamp = String.valueOf(System.currentTimeMillis());
	int progress = 0;
	logger.info("Got {} results from BigQuery database", size);

	while (!stopThread && !rows.isEmpty()) {
		final TableRow row = rows.remove(0);
		final String[] jsonResult = getJson(row);
		final String source = jsonResult[0];
		final IndexRequestBuilder builder = esClient.prepareIndex(index, type);
		if (jsonResult[1] != null) {
			builder.setId(jsonResult[1]);
		}
		try {
			builder.setOpType(OpType.CREATE)
				.setReplicationType(ReplicationType.ASYNC)
				.setOperationThreaded(true)
				.setTimestamp(timestamp)
				.setSource(source)
				.setCreate(create)
				.execute()
				.actionGet();
			count++;
		} catch (DocumentAlreadyExistsException daee) {
			logger.trace("Document already exists, create flag is {}", daee, create);
		}
		if (++progress % 100 == 0) {
			logger.debug("Processed {} entries ({} percent done)", progress, Math.round((float) progress / (float) size * 100f));
		}
	}
	logger.info("Imported {} entries into ElasticSeach from BigQuery (some duplicates may have been skipped)", count);
	return;
}
 

import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
/**
 * Imports dataset
 *
 * @throws java.io.IOException
 */
private static void dataSetImport()
        throws IOException, ExecutionException, IOException, InterruptedException, ParseException {

    JSONParser parser = new JSONParser();
    URL url = Resources.getResource(DATA_SET_NAME);
    Object obj = parser.parse(new FileReader(url.getFile()));

    JSONObject jsonObject = (JSONObject) obj;

    IndexResponse responseBook = client.prepareIndex(ES_INDEX_BOOK, ES_TYPE_INPUT, "id")
            .setSource(jsonObject.toJSONString())
            .execute()
            .actionGet();

    String json2 = "{" +

            "\"message\":\"" + MESSAGE_TEST + "\"" +
            "}";

    IndexResponse response2 = client.prepareIndex(ES_INDEX_MESSAGE, ES_TYPE_MESSAGE).setCreate(true)
            .setSource(json2).setReplicationType(ReplicationType.ASYNC)
            .execute()
            .actionGet();

    String json = "{" +
            "\"user\":\"kimchy\"," +
            "\"postDate\":\"2013-01-30\"," +
            "\"message\":\"trying out Elasticsearch\"" +
            "}";

    IndexResponse response = client.prepareIndex(ES_INDEX, ES_TYPE).setCreate(true)
            .setSource(json)
            .execute()
            .actionGet();

    String index = response.getIndex();
    String _type = response.getType();
    String _id = response.getId();
    try {
        CountResponse countResponse = client.prepareCount(ES_INDEX).setTypes(ES_TYPE)
                .execute()
                .actionGet();

        SearchResponse searchResponse = client.prepareSearch(ES_INDEX_BOOK).setTypes(ES_TYPE_INPUT)
                .execute()
                .actionGet();

        //searchResponse.getHits().hits();
        //assertEquals(searchResponse.getCount(), 1);
    } catch (AssertionError | Exception e) {
        cleanup();
        e.printStackTrace();
    }
}
 

import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
/**
 * Set the replication type for this operation.
 */
public JestBulkRequestBuilder setReplicationType(ReplicationType replicationType) {
    this.replicationType = replicationType;
    return this;
}
 

import org.elasticsearch.action.support.replication.ReplicationType; //导入依赖的package包/类
public BulkConfiguration setReplicationType(ReplicationType replicationType) {
	bulk_request_builder.setReplicationType(replicationType);
	refresh();
	return this;
}