使用ElasticsearchClient写个操作ES的工具类

好的，以下是一个简单的操作 Elasticsearch 的工具类，使用的是 Elasticsearch 官方提供的 Java 客户端——Elasticsearch Rest High Level Client。

首先，需要添加 Elasticsearch Rest High Level Client 的依赖：

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.10.2</version>
</dependency>

然后，就可以编写 Elasticsearch 操作的工具类了：

import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.List;

public class ElasticsearchUtil {

    private RestHighLevelClient client;

    public ElasticsearchUtil(String host, int port) {
        client = new RestHighLevelClient(RestClient.builder(new HttpHost(host, port)));
    }

    /**
     * 插入数据
     *
     * @param index    索引名称
     * @param id       文档ID
     * @param document 文档内容，可以是 JSON 字符串或者 Map、XContentBuilder 对象
     * @return 插入结果
     */
    public IndexResponse insert(String index, String id, Object document) throws IOException {
        IndexRequest request = new IndexRequest(index).id(id).source(document, XContentType.JSON);
        return client.index(request);
    }

    /**
     * 更新数据
     *
     * @param index    索引名称
     * @param id       文档ID
     * @param document 文档内容，可以是 JSON 字符串或者 Map、XContentBuilder 对象
     * @return 更新结果
     */
    public UpdateResponse update(String index, String id, Object document) throws IOException {
        UpdateRequest request = new UpdateRequest(index, id).doc(document, XContentType.JSON);
        return client.update(request);
    }

    /**
     * 删除数据
     *
     * @param index 索引名称
     * @param id    文档ID
     * @return 删除结果
     */
    public DeleteResponse delete(String index, String id) throws IOException {
        DeleteRequest request = new DeleteRequest(index, id);
        return client.delete(request);
    }

    /**
     * 根据ID查询数据
     *
     * @param index 索引名称
     * @param id    文档ID
     * @return 查询结果
     */
    public GetResponse getById(String index, String id) throws IOException {
        GetRequest request = new GetRequest(index, id);
        return client.get(request);
    }

    /**
     * 根据条件查询数据
     *
     * @param index 索引名称
     * @param query 查询条件，可以使用 QueryBuilders 类构建
     * @return 查询结果
     */
    public SearchResponse search(String index, QueryBuilders query) throws IOException {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().query(query);
        SearchRequest request = new SearchRequest(index).source(searchSourceBuilder);
        return client.search(request);
    }

    /**
     * 批量插入数据
     *
     * @param index      索引名称
     * @param documents  文档列表，每个元素可以是 JSON 字符串或者 Map、XContentBuilder 对象
     * @param batchSize  批量大小
     * @param retryTimes 失败重试次数
     * @return 批量插入结果
     */
    public BulkResponse bulkInsert(String index, List<Object> documents, int batchSize, int retryTimes) throws IOException {
        BulkRequest request = new BulkRequest();
        for (int i = 0; i < documents.size(); i++) {
            request.add(new IndexRequest(index).source(documents.get(i), XContentType.JSON));
            if ((i + 1) % batchSize == 0 || i == documents.size() - 1) {
                boolean success = false;
                int retry = 0;
                while (!success && retry < retryTimes) {
                    BulkResponse response = client.bulk(request);
                    if (response.hasFailures()) {
                        retry++;
                    } else {
                        success = true;
                        return response;
                    }
                }
                request.requests().clear();
            }
        }
        return null;
    }

    /**
     * 关闭 Elasticsearch 客户端
     */
    public void close() throws IOException {
        client.close();
    }
}

使用时只需要创建一个 ElasticsearchUtil 对象，然后调用相应的方法即可：

public static void main(String[] args) throws IOException {
    ElasticsearchUtil util = new ElasticsearchUtil("localhost", 9200);

    // 插入数据
    String document = "{\"name\":\"张三\",\"age\":20}";
    IndexResponse response = util.insert("test", "1", document);
    System.out.println(response.getResult().name());

    // 更新数据
    document = "{\"name\":\"李四\",\"age\":25}";
    UpdateResponse updateResponse = util.update("test", "1", document);
    System.out.println(updateResponse.getResult().name());

    // 查询数据
    GetResponse getResponse = util.getById("test", "1");
    System.out.println(getResponse.getSourceAsString());

    // 删除数据
    DeleteResponse deleteResponse = util.delete("test", "1");
    System.out.println(deleteResponse.getResult().name());

    // 条件查询数据
    QueryBuilders query = QueryBuilders.matchQuery("name", "张三");
    SearchResponse searchResponse = util.search("test", query);
    System.out.println(searchResponse.getHits().getTotalHits().value);

    // 批量插入数据
    List<Object> documents = new ArrayList<>();
    for (int i = 0; i < 10000; i++) {
        documents.add("{\"name\":\"张三\",\"age\":20}");
    }
    BulkResponse bulkResponse = util.bulkInsert("test", documents, 100, 3);
    System.out.println(bulkResponse.hasFailures());

    util.close();
}