import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
/**
 * 获取多个对象（根据ID）
 *
 * @param transportClient
 * @throws IOException
 */
private static void queryByMultiGet(TransportClient transportClient) throws IOException {

	MultiGetResponse multiGetItemResponses = transportClient.prepareMultiGet()
			.add("product_index", "product", "1")
			.add("product_index", "product", "2")
			.add("product_index", "product", "3")
			.add("product_index", "product", "4")
			.add("product_index", "product", "5")
			.get();

	String resultJSON = null;
	for (MultiGetItemResponse multiGetItemResponse : multiGetItemResponses) {
		GetResponse getResponse = multiGetItemResponse.getResponse();
		if (getResponse.isExists()) {
			resultJSON = getResponse.getSourceAsString();
		}
	}
	logger.info("--------------------------------：" + resultJSON);
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public MultiGetResponse toMultiGetResponse() {
    List<Map> docs = getAs(entityWrapper, "docs", List.class);
    MultiGetItemResponse[] itemResponses = new MultiGetItemResponse[docs.size()];
    for (int i = 0; i < docs.size(); i++) {
        Map<String, Object> doc = docs.get(i);
        MultiGetResponse.Failure failure = null;
        GetResponse getResponse = null;
        if (doc.containsKey("error")) {
            String index = getAsString(doc, "_index");
            String type = getAsString(doc, "_type");
            String id = getAsString(doc, "_id");
            String error = getAsString(doc, "error");
            failure = new MultiGetResponse.Failure(index, type, id, error);
        } else {
            getResponse = GetResponseAccessor.build(toGetResult(doc, doc));
        }
        MultiGetItemResponse itemResponse = new MultiGetItemResponse(getResponse, failure);
        itemResponses[i] = itemResponse;
    }
    return new MultiGetResponse(itemResponses);
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
/**
 * This method get the multiple Documents
 */
@Override
public void getMultipleDocument() {
    try {
        client = ESclient.getInstant();
        MultiGetResponse multipleItems = client.prepareMultiGet()
                .add("school", "tenth", "1")
                .add("school", "nineth", "1", "2", "3", "4")
                .add("college", "be", "1")
                .get();

        multipleItems.forEach(multipleItem -> {
            GetResponse response = multipleItem.getResponse();
            if (response.isExists()) {
                String json = response.getSourceAsString();
                log.info("Respense Data : " + json);
            }
        });
    } catch (Exception ex) {
        log.error("Exception occurred while get Multiple Document : " + ex, ex);
    }
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) throws Exception {
    MultiGetRequest multiGetRequest = new MultiGetRequest();
    multiGetRequest.refresh(request.paramAsBoolean("refresh", multiGetRequest.refresh()));
    multiGetRequest.preference(request.param("preference"));
    multiGetRequest.realtime(request.paramAsBoolean("realtime", null));
    multiGetRequest.ignoreErrorsOnGeneratedFields(request.paramAsBoolean("ignore_errors_on_generated_fields", false));

    String[] sFields = null;
    String sField = request.param("fields");
    if (sField != null) {
        sFields = Strings.splitStringByCommaToArray(sField);
    }

    FetchSourceContext defaultFetchSource = FetchSourceContext.parseFromRestRequest(request);
    multiGetRequest.add(request.param("index"), request.param("type"), sFields, defaultFetchSource, request.param("routing"), RestActions.getRestContent(request), allowExplicitIndex);

    client.multiGet(multiGetRequest, new RestToXContentListener<MultiGetResponse>(channel));
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
/**
 * Find a set of documents.
 *
 * @param ids          The unique IDs of the documents you wish to find.
 * @param index        The index in which the document exists in Elasticsearch.
 * @param documentType The document type in Elasticsearch for this document.
 * @param javaType     The concrete java type the document represents.
 */
final protected <T extends IDocument> IPage<T> findDocuments(Set<String> ids, String index, String documentType, Class<T> javaType) {
    MultiGetResponse response = this.getClient().prepareMultiGet().add(index, documentType, ids).execute().actionGet();

    List<T> results = Lists.newArrayListWithCapacity(ids.size());
    for (MultiGetItemResponse item : response.getResponses()) {
        if (item.isFailed()) {
            throw new SearchServiceException("Unexpected failure while searching for documents.");
        }
        results.add(this.getDocumentFromSearchResponse(item.getResponse(), javaType));
    }

    IPage page = Page.of(results);
    page.setPageSize(results.size());
    page.setPageNumber(1);
    page.setTotalElements(results.size());

    return page;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override public JsonArray getTpsMetric(int instanceId, long startTimeBucket, long endTimeBucket) {
    MultiGetRequestBuilder prepareMultiGet = getClient().prepareMultiGet();

    long timeBucket = startTimeBucket;
    do {
        String id = timeBucket + Const.ID_SPLIT + instanceId;
        prepareMultiGet.add(InstanceMetricTable.TABLE, InstanceMetricTable.TABLE_TYPE, id);
        timeBucket = TimeBucketUtils.INSTANCE.addSecondForSecondTimeBucket(TimeBucketUtils.TimeBucketType.SECOND.name(), timeBucket, 1);
    }
    while (timeBucket <= endTimeBucket);

    JsonArray metrics = new JsonArray();
    MultiGetResponse multiGetResponse = prepareMultiGet.get();
    for (MultiGetItemResponse response : multiGetResponse.getResponses()) {
        if (response.getResponse().isExists()) {
            metrics.add(((Number)response.getResponse().getSource().get(InstanceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue());
        } else {
            metrics.add(0);
        }
    }
    return metrics;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override public JsonArray getRespTimeMetric(int instanceId, long startTimeBucket, long endTimeBucket) {
    MultiGetRequestBuilder prepareMultiGet = getClient().prepareMultiGet();

    int i = 0;
    long timeBucket;
    do {
        timeBucket = TimeBucketUtils.INSTANCE.addSecondForSecondTimeBucket(TimeBucketUtils.TimeBucketType.SECOND.name(), startTimeBucket, i);
        String id = timeBucket + Const.ID_SPLIT + instanceId;
        prepareMultiGet.add(InstanceMetricTable.TABLE, InstanceMetricTable.TABLE_TYPE, id);
        i++;
    }
    while (timeBucket <= endTimeBucket);

    JsonArray metrics = new JsonArray();
    MultiGetResponse multiGetResponse = prepareMultiGet.get();
    for (MultiGetItemResponse response : multiGetResponse.getResponses()) {
        if (response.getResponse().isExists()) {
            long callTimes = ((Number)response.getResponse().getSource().get(InstanceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue();
            long costTotal = ((Number)response.getResponse().getSource().get(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue();
            metrics.add(costTotal / callTimes);
        } else {
            metrics.add(0);
        }
    }
    return metrics;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override public JsonArray getMetric(int instanceId, long startTimeBucket, long endTimeBucket) {
    MultiGetRequestBuilder prepareMultiGet = getClient().prepareMultiGet();

    long timeBucket = startTimeBucket;
    do {
        timeBucket = TimeBucketUtils.INSTANCE.addSecondForSecondTimeBucket(TimeBucketUtils.TimeBucketType.SECOND.name(), timeBucket, 1);
        String id = timeBucket + Const.ID_SPLIT + instanceId;
        prepareMultiGet.add(CpuMetricTable.TABLE, CpuMetricTable.TABLE_TYPE, id);
    }
    while (timeBucket <= endTimeBucket);

    JsonArray metrics = new JsonArray();
    MultiGetResponse multiGetResponse = prepareMultiGet.get();
    for (MultiGetItemResponse response : multiGetResponse.getResponses()) {
        if (response.getResponse().isExists()) {
            double cpuUsed = ((Number)response.getResponse().getSource().get(CpuMetricTable.COLUMN_USAGE_PERCENT)).doubleValue();
            metrics.add((int)(cpuUsed * 100));
        } else {
            metrics.add(0);
        }
    }
    return metrics;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
private void bulkFetchList(Set<Key> stillToFetch, final GDSCallback<Map<Key, Entity>> callback) {
	MultiGetRequestBuilder requestBuilder = gds.getClient().prepareMultiGet();

	for (Key key : stillToFetch) {
		requestBuilder.add(gds.indexFor(key.kind), key.kind, key.id);
	}
	
	requestBuilder.execute(new ActionListener<MultiGetResponse>() {
		
		@Override
		public void onResponse(MultiGetResponse response) {
			Map<Key, Entity> resultMap = Collections.synchronizedMap(new HashMap<Key, Entity>());
			for (MultiGetItemResponse itemResponse : response.getResponses()) {
				Entity entity = new Entity(itemResponse.getType(), itemResponse.getResponse().getId(), itemResponse.getResponse().getSourceAsMap());
				resultMap.put(new Key(itemResponse.getType(), itemResponse.getId()), entity);
			}
			callback.onSuccess(resultMap, null);
		}
		
		@Override
		public void onFailure(Throwable e) {
			callback.onSuccess(null, e);
		}
	});
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override
public <T> List<T> getByIds(Class<T> clazz, List<String> ids) {
    String typeName = MappingProcessor.getIndexTypeName(clazz);
    if (logger.isDebugEnabled()) {
        logger.debug("Get objects by ids, class: {}, type: {}, ids: {}", clazz.getSimpleName(), typeName, ids);
    }
    MultiGetResponse responses = client.prepareMultiGet().add(getIndexName(), typeName, ids).get();
    List<T> results = new ArrayList<T>();
    if (responses != null) {
        for (MultiGetItemResponse response : responses) {
            if (response.getResponse() != null) {
                results.add(objectProcessor.fromJsonString(response.getResponse().getSourceAsString(), clazz));
            }
        }
    }
    return results;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@Override
public Map<String, Table> loadAll(Collection<String> keys) {
    logger.info("Load all called for multiple keys");
    MultiGetResponse response = elasticsearchConnection.getClient()
            .prepareMultiGet()
            .add(TABLE_META_INDEX, TABLE_META_TYPE, keys)
            .execute()
            .actionGet();
    Map<String, Table> tables = Maps.newHashMap();
    for (MultiGetItemResponse multiGetItemResponse : response) {
        try {
            Table table = objectMapper.readValue(multiGetItemResponse.getResponse().getSourceAsString(),
                    Table.class);
            tables.put(table.getName(), table);
        } catch (Exception e) {
            throw new RuntimeException("Error getting data for table: " + multiGetItemResponse.getId());
        }
    }
    logger.info("Loaded value count: " + tables.size());
    return tables;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
/**
 * http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/docs-multi-get.html
 * @param clzz
 * @param ids
 * @return
 */
public <T> List<T> queryForObjects(Class<T> clzz, String... ids) throws SearchException {
       String typeName = this.getTypeNameFromClass(clzz);
       Timer.Context timer = searchMetrics.indexTimer(typeName, "queryForObjects");
	try {
		MultiGetRequestBuilder request = client.prepareMultiGet();
		request.add(indexName, typeName, ids);
		MultiGetResponse response = request.execute().actionGet();
		List<T> results = new ArrayList<T>();
		for(MultiGetItemResponse resp : response.getResponses()){
			if(!resp.isFailed()){
				T o = this.docMapper.asObject(clzz, resp.getResponse().getSourceAsString());
				results.add(o);
			}
		}
           searchMetrics.indexIncr(typeName, "queryForObjects");
		return results;
	} catch (Exception e) {
           searchMetrics.failedIncr(typeName, "queryForObjects", e);
           logger.error("queryForObjects Error. ", e);
           throw new SearchException("queryForObjects Error. typeName="+typeName+", ids="+ids, e);
	}finally {
           timer.stop();
       }
   }
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public Set<String> existBulk(String indexName, String typeName, final Collection<String> ids) {
    if (ids == null || ids.size() == 0) return new HashSet<>();
    MultiGetResponse multiGetItemResponses = elasticsearchClient.prepareMultiGet()
            .add(indexName, typeName, ids)
            .get();
    Set<String> er = new HashSet<>();
    for (MultiGetItemResponse itemResponse : multiGetItemResponses) { 
        GetResponse response = itemResponse.getResponse();
        if (response.isExists()) {                      
            er.add(response.getId());
        }
    }
    return er;
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
private void assertGetFieldException(String index, String type, String docId, String field) {
    try {
        client().prepareGet().setIndex(index).setType(type).setId(docId).setStoredFields(field);
        fail();
    } catch (ElasticsearchException e) {
        assertTrue(e.getMessage().contains("You can only get this field after refresh() has been called."));
    }
    MultiGetResponse multiGetResponse = client().prepareMultiGet().add(new MultiGetRequest.Item(index, type, docId).storedFields(field)).get();
    assertNull(multiGetResponse.getResponses()[0].getResponse());
    assertTrue(multiGetResponse.getResponses()[0].getFailure().getMessage().contains("You can only get this field after refresh() has been called."));
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
private GetResponse multiGetDocument(String index, String type, String docId, String field, @Nullable String routing) {
  MultiGetRequest.Item getItem = new MultiGetRequest.Item(index, type, docId).storedFields(field);
    if (routing != null) {
        getItem.routing(routing);
    }
    MultiGetRequestBuilder multiGetRequestBuilder = client().prepareMultiGet().add(getItem);
    MultiGetResponse multiGetResponse = multiGetRequestBuilder.get();
    assertThat(multiGetResponse.getResponses().length, equalTo(1));
    return multiGetResponse.getResponses()[0].getResponse();
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public void testThatMgetShouldWorkWithOneIndexMissing() throws IOException {
    createIndex("test");

    client().prepareIndex("test", "test", "1").setSource(jsonBuilder().startObject().field("foo", "bar").endObject())
            .setRefreshPolicy(IMMEDIATE).get();

    MultiGetResponse mgetResponse = client().prepareMultiGet()
            .add(new MultiGetRequest.Item("test", "test", "1"))
            .add(new MultiGetRequest.Item("nonExistingIndex", "test", "1"))
            .get();
    assertThat(mgetResponse.getResponses().length, is(2));

    assertThat(mgetResponse.getResponses()[0].getIndex(), is("test"));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(false));

    assertThat(mgetResponse.getResponses()[1].getIndex(), is("nonExistingIndex"));
    assertThat(mgetResponse.getResponses()[1].isFailed(), is(true));
    assertThat(mgetResponse.getResponses()[1].getFailure().getMessage(), is("no such index"));
    assertThat(((ElasticsearchException) mgetResponse.getResponses()[1].getFailure().getFailure()).getIndex().getName(),
        is("nonExistingIndex"));

    mgetResponse = client().prepareMultiGet()
            .add(new MultiGetRequest.Item("nonExistingIndex", "test", "1"))
            .get();
    assertThat(mgetResponse.getResponses().length, is(1));
    assertThat(mgetResponse.getResponses()[0].getIndex(), is("nonExistingIndex"));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(true));
    assertThat(mgetResponse.getResponses()[0].getFailure().getMessage(), is("no such index"));
    assertThat(((ElasticsearchException) mgetResponse.getResponses()[0].getFailure().getFailure()).getIndex().getName(),
        is("nonExistingIndex"));
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public void testThatMgetShouldWorkWithMultiIndexAlias() throws IOException {
    assertAcked(prepareCreate("test").addAlias(new Alias("multiIndexAlias")));
    assertAcked(prepareCreate("test2").addAlias(new Alias("multiIndexAlias")));

    client().prepareIndex("test", "test", "1").setSource(jsonBuilder().startObject().field("foo", "bar").endObject())
        .setRefreshPolicy(IMMEDIATE).get();

    MultiGetResponse mgetResponse = client().prepareMultiGet()
        .add(new MultiGetRequest.Item("test", "test", "1"))
        .add(new MultiGetRequest.Item("multiIndexAlias", "test", "1"))
        .get();
    assertThat(mgetResponse.getResponses().length, is(2));

    assertThat(mgetResponse.getResponses()[0].getIndex(), is("test"));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(false));

    assertThat(mgetResponse.getResponses()[1].getIndex(), is("multiIndexAlias"));
    assertThat(mgetResponse.getResponses()[1].isFailed(), is(true));
    assertThat(mgetResponse.getResponses()[1].getFailure().getMessage(), containsString("more than one indices"));

    mgetResponse = client().prepareMultiGet()
        .add(new MultiGetRequest.Item("multiIndexAlias", "test", "1"))
        .get();
    assertThat(mgetResponse.getResponses().length, is(1));
    assertThat(mgetResponse.getResponses()[0].getIndex(), is("multiIndexAlias"));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(true));
    assertThat(mgetResponse.getResponses()[0].getFailure().getMessage(), containsString("more than one indices"));
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public void testThatParentPerDocumentIsSupported() throws Exception {
    assertAcked(prepareCreate("test").addAlias(new Alias("alias"))
            .addMapping("test", jsonBuilder()
                    .startObject()
                    .startObject("test")
                    .startObject("_parent")
                    .field("type", "foo")
                    .endObject()
                    .endObject()
                    .endObject()));

    client().prepareIndex("test", "test", "1").setParent("4").setRefreshPolicy(IMMEDIATE)
            .setSource(jsonBuilder().startObject().field("foo", "bar").endObject())
            .get();

    MultiGetResponse mgetResponse = client().prepareMultiGet()
            .add(new MultiGetRequest.Item(indexOrAlias(), "test", "1").parent("4"))
            .add(new MultiGetRequest.Item(indexOrAlias(), "test", "1"))
            .get();

    assertThat(mgetResponse.getResponses().length, is(2));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(false));
    assertThat(mgetResponse.getResponses()[0].getResponse().isExists(), is(true));

    assertThat(mgetResponse.getResponses()[1].isFailed(), is(true));
    assertThat(mgetResponse.getResponses()[1].getResponse(), nullValue());
    assertThat(mgetResponse.getResponses()[1].getFailure().getMessage(), equalTo("routing is required for [test]/[test]/[1]"));
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public void testThatSourceFilteringIsSupported() throws Exception {
    assertAcked(prepareCreate("test").addAlias(new Alias("alias")));
    BytesReference sourceBytesRef = jsonBuilder().startObject()
            .array("field", "1", "2")
            .startObject("included").field("field", "should be seen").field("hidden_field", "should not be seen").endObject()
            .field("excluded", "should not be seen")
            .endObject().bytes();
    for (int i = 0; i < 100; i++) {
        client().prepareIndex("test", "type", Integer.toString(i)).setSource(sourceBytesRef, XContentType.JSON).get();
    }

    MultiGetRequestBuilder request = client().prepareMultiGet();
    for (int i = 0; i < 100; i++) {
        if (i % 2 == 0) {
            request.add(new MultiGetRequest.Item(indexOrAlias(), "type", Integer.toString(i))
                .fetchSourceContext(new FetchSourceContext(true, new String[] {"included"}, new String[] {"*.hidden_field"})));
        } else {
            request.add(new MultiGetRequest.Item(indexOrAlias(), "type", Integer.toString(i))
                .fetchSourceContext(new FetchSourceContext(false)));
        }
    }

    MultiGetResponse response = request.get();

    assertThat(response.getResponses().length, equalTo(100));
    for (int i = 0; i < 100; i++) {
        MultiGetItemResponse responseItem = response.getResponses()[i];
        assertThat(responseItem.getIndex(), equalTo("test"));
        if (i % 2 == 0) {
            Map<String, Object> source = responseItem.getResponse().getSourceAsMap();
            assertThat(source.size(), equalTo(1));
            assertThat(source, hasKey("included"));
            assertThat(((Map<String, Object>) source.get("included")).size(), equalTo(1));
            assertThat(((Map<String, Object>) source.get("included")), hasKey("field"));
        } else {
            assertThat(responseItem.getResponse().getSourceAsBytes(), nullValue());
        }
    }
}
 

import org.elasticsearch.action.get.MultiGetResponse; //导入依赖的package包/类
public void testThatRoutingPerDocumentIsSupported() throws Exception {
    assertAcked(prepareCreate("test").addAlias(new Alias("alias"))
            .setSettings(Settings.builder()
                    .put(indexSettings())
                    .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, between(2, DEFAULT_MAX_NUM_SHARDS))));

    final String id = routingKeyForShard("test", 0);
    final String routingOtherShard = routingKeyForShard("test", 1);

    client().prepareIndex("test", "test", id).setRefreshPolicy(IMMEDIATE).setRouting(routingOtherShard)
            .setSource(jsonBuilder().startObject().field("foo", "bar").endObject())
            .get();

    MultiGetResponse mgetResponse = client().prepareMultiGet()
            .add(new MultiGetRequest.Item(indexOrAlias(), "test", id).routing(routingOtherShard))
            .add(new MultiGetRequest.Item(indexOrAlias(), "test", id))
            .get();

    assertThat(mgetResponse.getResponses().length, is(2));
    assertThat(mgetResponse.getResponses()[0].isFailed(), is(false));
    assertThat(mgetResponse.getResponses()[0].getResponse().isExists(), is(true));
    assertThat(mgetResponse.getResponses()[0].getResponse().getIndex(), is("test"));

    assertThat(mgetResponse.getResponses()[1].isFailed(), is(false));
    assertThat(mgetResponse.getResponses()[1].getResponse().isExists(), is(false));
    assertThat(mgetResponse.getResponses()[1].getResponse().getIndex(), is("test"));
}