在Java中实现Elasticsearch多索引批量操作：优化单次请求

本教程旨在指导如何在Java应用程序中通过一次请求，高效地向Elasticsearch的多个不同索引执行批量数据操作。我们将探讨Elasticsearch原生批量API的原理，并详细展示如何使用新的Java API Client和旧的Java Rest High-Level Client来构建包含多索引操作的批量请求，从而优化性能并简化代码。

1. 引言

在处理elasticsearch数据时，我们经常需要对不同类型的数据进行批量索引、更新或删除。一个常见的场景是，我们有多个实体列表，例如 personupdatelist、addressupdatelist 和 positionupdatelist，它们分别对应不同的elasticsearch索引。如果按照传统做法，为每个列表调用一次批量更新操作（例如 this.operations.bulkupdate(list, class)），则会产生多次网络往返，这在数据量大时会显著降低应用程序的性能。

Elasticsearch的批量（Bulk）API设计之初就考虑到了这种需求，它允许在单个请求中对多个索引执行多种操作。本文将深入探讨如何在Java应用程序中，利用Elasticsearch官方客户端实现这一高效的多索引批量操作，从而优化性能并简化代码结构。

2. Elasticsearch批量API基础

Elasticsearch的 _bulk API 允许在一个请求体中包含多个操作（如索引、创建、更新、删除）和文档。其核心优势在于，这些操作可以针对不同的索引和文档类型，而无需为每个操作或每个索引发送单独的HTTP请求。

一个典型的 _bulk 请求体结构如下：

POST _bulk{"index":{"_index":"index_1", "_id":"1"}}{"field1":"value1", "field2":"value2"}{"index":{"_index":"index_2", "_id":"1"}}{"data":"some_data"}{"delete":{"_index":"index_1", "_id":"2"}}

从上述示例可以看出，_bulk 请求体由一系列操作元数据行和可选的文档源数据行组成。每个操作元数据行指定了要执行的操作类型（index、create、update、delete）以及目标索引和文档ID。紧随其后的行（如果操作需要）则是文档的JSON源数据。这种结构使得在一个HTTP请求中处理异构数据成为可能。

立即学习“Java免费学习笔记（深入）”；

3. Java客户端实现：构建多索引批量请求

Elasticsearch提供了多种Java客户端来与集群进行交互。我们将重点介绍两种主要的客户端：新的Java API Client和旧的Java Rest High-Level Client。

3.1 使用Elasticsearch Java API Client (推荐)

Elasticsearch Java API Client 是Elasticsearch官方推荐的、面向未来的Java客户端，它提供了类型安全的API和更好的性能。

要使用此客户端实现多索引批量操作，您需要构建一个 BulkRequest 对象，并向其添加多个操作。每个操作可以指定不同的索引和文档。

import co.elastic.clients.elasticsearch.ElasticsearchClient;import co.elastic.clients.elasticsearch.core.BulkRequest;import co.elastic.clients.elasticsearch.core.BulkResponse;import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;import co.elastic.clients.json.JsonData;import java.io.IOException;import java.util.List;import java.util.Map;public class MultiIndexBulkExample {    private final ElasticsearchClient esClient;    public MultiIndexBulkExample(ElasticsearchClient esClient) {        this.esClient = esClient;    }    /**     * 执行多索引批量操作     * @param personDataList 个人数据列表，将索引到 "person_index"     * @param addressDataList 地址数据列表，将索引到 "address_index"     * @param positionDataList 职位数据列表，将索引到 "position_index"     * @throws IOException 如果Elasticsearch操作失败     */    public void bulkSaveMultipleIndices(            List<Map> personDataList,            List<Map> addressDataList,            List<Map> positionDataList) throws IOException {        BulkRequest.Builder br = new BulkRequest.Builder();        // 添加person数据到 "person_index"        for (Map person : personDataList) {            br.operations(op -> op                .index(idx -> idx                    .index("person_index")                    .id(person.get("id").toString()) // 假设每个文档都有一个id字段                    .document(JsonData.of(person))                )            );        }        // 添加address数据到 "address_index"        for (Map address : addressDataList) {            br.operations(op -> op                .index(idx -> idx                    .index("address_index")                    .id(address.get("id").toString())                    .document(JsonData.of(address))                )            );        }        // 添加position数据到 "position_index"        for (Map position : positionDataList) {            br.operations(op -> op                .index(idx -> idx                    .index("position_index")                    .id(position.get("id").toString())                    .document(JsonData.of(position))                )            );        }        // 执行批量请求        BulkResponse result = esClient.bulk(br.build());        // 处理批量操作结果        if (result.errors()) {            System.err.println("Bulk operation had errors:");            for (BulkResponseItem item : result.items()) {                if (item.error() != null) {                    System.err.println("  " + item.error().reason());                }            }        } else {            System.out.println("Bulk operation completed successfully.");        }    }    // 示例用法 (需要初始化ElasticsearchClient)    public static void main(String[] args) throws IOException {        // 实际应用中，esClient应该通过依赖注入或配置来获取        // 例如:        // RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();        // ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());        // ElasticsearchClient esClient = new ElasticsearchClient(transport);        // 假设我们已经有了esClient实例        ElasticsearchClient esClient = null; // 替换为实际的客户端实例        MultiIndexBulkExample service = new MultiIndexBulkExample(esClient);        // 准备示例数据        List<Map> persons = List.of(            Map.of("id", "p1", "name", "Alice", "age", 30),            Map.of("id", "p2", "name", "Bob", "age", 25)        );        List<Map> addresses = List.of(            Map.of("id", "a1", "street", "Main St", "city", "New York"),            Map.of("id", "a2", "street", "Park Ave", "city", "Los Angeles")        );        List<Map> positions = List.of(            Map.of("id", "pos1", "title", "Engineer", "department", "IT"),            Map.of("id", "pos2", "title", "Manager", "department", "HR")        );        // 调用批量保存方法        // service.bulkSaveMultipleIndices(persons, addresses, positions);        // 注意: 在实际运行前，请确保esClient已正确初始化    }}

代码说明：

BulkRequest.Builder 是构建批量请求的入口。br.operations(op -> op.index(…)) 方法用于向批量请求中添加一个索引操作。在 index 操作中，您需要指定目标索引名称（index(“index_name”)）、文档ID（id(“document_id”)）以及文档内容（document(JsonData.of(yourObject))）。JsonData.of() 可以将您的Java对象转换为Elasticsearch可接受的JSON格式。最后，通过 esClient.bulk(br.build()) 执行请求。

3.2 使用Elasticsearch Java Rest High-Level Client (已弃用，但仍广泛使用)

Elasticsearch Java Rest High-Level Client 是早期广泛使用的客户端，但自Elasticsearch 7.15.0起已被标记为弃用，并将在Elasticsearch 8.0.0中移除。尽管如此，许多现有项目仍在使用它。

使用此客户端，您需要构建一个 BulkRequest 对象，并向其添加多个 IndexRequest、UpdateRequest 或 DeleteRequest。

畅图

AI可视化工具

147 查看详情

import org.elasticsearch.action.bulk.BulkRequest;import org.elasticsearch.action.bulk.BulkResponse;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.RequestOptions;import org.elasticsearch.client.RestHighLevelClient;import org.elasticsearch.common.xcontent.XContentType;import java.io.IOException;import java.util.List;import java.util.Map;public class MultiIndexBulkHighLevelClientExample {    private final RestHighLevelClient restHighLevelClient;    public MultiIndexBulkHighLevelClientExample(RestHighLevelClient restHighLevelClient) {        this.restHighLevelClient = restHighLevelClient;    }    /**     * 执行多索引批量操作     * @param personDataList 个人数据列表，将索引到 "person_index"     * @param addressDataList 地址数据列表，将索引到 "address_index"     * @param positionDataList 职位数据列表，将索引到 "position_index"     * @throws IOException 如果Elasticsearch操作失败     */    public void bulkSaveMultipleIndices(            List<Map> personDataList,            List<Map> addressDataList,            List<Map> positionDataList) throws IOException {        BulkRequest request = new BulkRequest();        // 添加person数据到 "person_index"        for (Map person : personDataList) {            request.add(new IndexRequest("person_index")                    .id(person.get("id").toString())                    .source(person, XContentType.JSON)); // 将Map转换为JSON        }        // 添加address数据到 "address_index"        for (Map address : addressDataList) {            request.add(new IndexRequest("address_index")                    .id(address.get("id").toString())                    .source(address, XContentType.JSON));        }        // 添加position数据到 "position_index"        for (Map position : positionDataList) {            request.add(new IndexRequest("position_index")                    .id(position.get("id").toString())                    .source(position, XContentType.JSON));        }        // 执行批量请求        BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);        // 处理批量操作结果        if (bulkResponse.hasFailures()) {            System.err.println("Bulk operation had failures:");            // 遍历失败项            bulkResponse.forEach(bulkItemResponse -> {                if (bulkItemResponse.isFailed()) {                    System.err.println("  " + bulkItemResponse.getFailureMessage());                }            });        } else {            System.out.println("Bulk operation completed successfully.");        }    }    // 示例用法 (需要初始化RestHighLevelClient)    public static void main(String[] args) throws IOException {        // 实际应用中，restHighLevelClient应该通过依赖注入或配置来获取        // 例如:        // RestHighLevelClient restHighLevelClient = new RestHighLevelClient(        //     RestClient.builder(new HttpHost("localhost", 9200)));        // 假设我们已经有了restHighLevelClient实例        RestHighLevelClient restHighLevelClient = null; // 替换为实际的客户端实例        MultiIndexBulkHighLevelClientExample service = new MultiIndexBulkHighLevelClientExample(restHighLevelClient);        // 准备示例数据 (同上)        List<Map> persons = List.of(            Map.of("id", "p1", "name", "Alice", "age", 30),            Map.of("id", "p2", "name", "Bob", "age", 25)        );        List<Map> addresses = List.of(            Map.of("id", "a1", "street", "Main St", "city", "New York"),            Map.of("id", "a2", "street", "Park Ave", "city", "Los Angeles")        );        List<Map> positions = List.of(            Map.of("id", "pos1", "title", "Engineer", "department", "IT"),            Map.of("id", "pos2", "title", "Manager", "department", "HR")        );        // 调用批量保存方法        // service.bulkSaveMultipleIndices(persons, addresses, positions);        // 注意: 在实际运行前，请确保restHighLevelClient已正确初始化    }}

代码说明：

BulkRequest 对象用于聚合所有批量操作。request.add(new IndexRequest(“index_name”).id(“document_id”).source(yourObject, XContentType.JSON)) 方法用于添加一个索引操作。source() 方法可以接受各种类型的输入，例如Map、字符串或XContentBuilder。这里我们使用Map并指定 XContentType.JSON。最后，通过 restHighLevelClient.bulk(request, RequestOptions.DEFAULT) 执行请求。

4. 与Spring Data Elasticsearch的集成考量

Spring Data Elasticsearch 提供了一个高级抽象 ElasticsearchOperations，它简化了与Elasticsearch的交互。然而，ElasticsearchOperations 中的 bulkUpdate(List entities, Class entityClass) 等方法通常是针对单一实体类型和单一索引设计的。这意味着它们期望 entities 列表中的所有对象都属于 entityClass 类型，并且会索引到由 entityClass 定义的默认索引中。

当您需要执行真正意义上的“多索引、多类型”批量操作时，即一次性向多个不同索引写入不同类型的数据，ElasticsearchOperations 可能没有直接提供一个一站式的方法。在这种情况下，您需要：

获取底层Elasticsearch客户端： ElasticsearchOperations 通常会封装底层的Elasticsearch客户端（ElasticsearchClient 或 RestHighLevelClient）。您可以通过注入 ElasticsearchClient 或 RestHighLevelClient bean，或者如果 ElasticsearchOperations 提供了访问底层客户端的方法（例如，通过 ElasticsearchRestTemplate 获取），来直接使用它们。

// 假设您已经注入了ElasticsearchClient (对于新客户端)// @Autowired// private ElasticsearchClient esClient;// 或者 RestHighLevelClient (对于旧客户端)// @Autowired// private RestHighLevelClient restHighLevelClient;

手动构建BulkRequest： 按照上述第3节的示例，手动构建 BulkRequest，将不同类型、不同索引的操作添加到同一个请求中。

执行请求： 使用获取到的底层客户端执行构建好的 BulkRequest。

通过这种方式，即使在使用Spring Data Elasticsearch的项目中，您也可以灵活地利用Elasticsearch原生客户端的强大功能，实现复杂的多索引批量操作。

5. 批量操作的注意事项

在执行批量操作时，需要考虑以下几点以确保系统的稳定性、性能和数据完整性：

错误处理： 批量操作并非事务性的。这意味着即使请求中的某些操作失败，其他操作仍然可能成功。因此，在收到 BulkResponse 后，务必检查 result.errors() (新客户端) 或 bulkResponse.hasFailures() (旧客户端)，并遍历 items() 或 forEach() 来识别并处理每个失败的子操作。批量大小： 合理设置批量请求的大小至关重要。过大： 可能导致内存溢出（客户端或Elasticsearch节点）、网络传输延迟增加、请求超时。过小： 导致频繁的网络往返，降低整体吞吐量。建议： 批量大小通常在几百到几千个文档之间，具体取决于文档大小、网络带宽和Elasticsearch集群的资源。最佳实践是通过测试和监控来确定最适合您场景的值。性能优化：异步执行： 对于对响应时间不敏感的场景，可以考虑异步执行批量请求，以避免阻塞主线程。线程池： 如果有大量数据需要批量处理，可以利用线程池并行发送多个批量请求。数据一致性： 批量操作是非事务性的。如果一个批量请求中的部分操作失败，您需要自行处理数据回滚或重试逻辑，以确保应用程序层面的数据一致性。Elasticsearch本身不提供跨文档或跨索引的事务保证。ID管理： 确保为每个文档提供唯一的ID。如果未提供，Elasticsearch会自动生成一个。但在更新或删除操作中，ID是必需的。

6. 总结

通过本教程，我们了解了Elasticsearch批量API的强大功能，它允许在单个请求中对多个不同索引执行数据操作，从而显著提高数据处理效率。我们详细展示了如何使用Elasticsearch Java API Client和Java Rest High-Level Client来构建和执行此类多索引批量请求，并讨论了与Spring Data Elasticsearch集成时的策略。

在实际应用中，通过将多个独立的数据列表聚合到一个 BulkRequest 中，您可以有效地减少网络往返次数，降低服务器负载，并优化应用程序的整体性能。同时，合理处理批量操作的响应和错误，并注意批量大小等性能考量，是构建健壮且高效的Elasticsearch集成方案的关键。

以上就是在Java中实现Elasticsearch多索引批量操作：优化单次请求的详细内容，更多请关注创想鸟其它相关文章！