ElasticsearchJavaClient工具类分析

server/2025/1/11 0:57:41/

最近升级了Elasticsearch版本，从7.X升级到8.X的变化还是比较大的，原来7版本用的是RestHighLevelClient，8.X弃用RestHighLevelClient转而支持ElasticsearchClient，并且api调用方式经过建造者模式的改造，变成了链式调用。

因此为了更好地使用ElasticsearchClient的api操作Elasticsearch，封装了一个工具类，包含了常用的一些数据操作的方法。废话不多说直接上代码。。。

1、pom依赖

        <dependency><groupId>co.elastic.clients</groupId><artifactId>elasticsearch-java</artifactId><version>8.15.2</version></dependency><dependency><artifactId>elasticsearch-rest-client</artifactId><groupId>org.elasticsearch.client</groupId><version>8.15.2</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.30</version></dependency>

2、工具类代码

java">import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Result;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.AnalyzeRequest;
import co.elastic.clients.elasticsearch.indices.AnalyzeResponse;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.analyze.AnalyzeToken;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import java.io.IOException;
import java.io.StringReader;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;/*** Elasticsearch工具类* Elasticsearch版本：8.15.3*/
public class ElasticsearchJavaClient {private ElasticsearchClient client;/*** 构造方法，获取客户端（未开启认证）* @param httpUrls*/public ElasticsearchJavaClient(String[] httpUrls){HttpHost[] httpHosts = Arrays.stream(httpUrls).map(HttpHost::create).toArray(HttpHost[]::new);this.client = new ElasticsearchClient(new RestClientTransport(RestClient.builder(httpHosts).build(),new JacksonJsonpMapper()));}/*** 构造方法，获取客户端（开启认证，通过用户名密码进行认证并获取客户端）* @param httpUrls* @param username* @param password*/public ElasticsearchJavaClient(String[] httpUrls, String username, String password){HttpHost[] httpHosts = Arrays.stream(httpUrls).map(HttpHost::create).toArray(HttpHost[]::new);final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));RestClientBuilder builder = RestClient.builder(httpHosts);builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);}});this.client = new ElasticsearchClient(new RestClientTransport(builder.build(), new JacksonJsonpMapper()));}/*** 创建索引* @param indexName  索引名* @param numberOfShards  分片数* @param numberOfReplicas  副本数* @param mapping  mapping设计json字符串* @return*/public boolean createIndex(String indexName, Integer numberOfShards,Integer numberOfReplicas, String mapping) {CreateIndexResponse response = null;try {response = client.indices().create(builder -> builder.index(indexName).settings(b -> b.numberOfReplicas(numberOfReplicas.toString()).numberOfShards(numberOfShards.toString())).mappings(a -> a.withJson(new StringReader(mapping))));} catch (IOException e) {e.printStackTrace();} finally {client.shutdown();}return response.acknowledged();}/*** 删除索引* @param indexName  索引名* @return*/public boolean deleteIndex(String indexName) {try {return client.indices().delete(a -> a.index(indexName)).acknowledged();} catch (IOException e) {e.printStackTrace();} finally {client.shutdown();}return false;}/*** 判断索引是否已存在* @param indexName 索引名* @return*/public boolean indexExisit(String indexName) {try {return client.indices().exists(req -> req.index(indexName)).value();} catch (IOException e) {e.printStackTrace();} finally {client.shutdown();}return false;}/*** 由于数据落盘有默认的1秒延迟，刷新后使数据能被检索到* @param indexString*/public void refresh(String indexString){try {client.indices().refresh(req -> req.index(indexString));} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}}/*** 插入数据* @param indexName* @param data* @return*/public String insertData(String indexName, JSONObject data){try {IndexResponse response = client.index(a -> a.index(indexName).document(data));return response.id();} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return null;}/*** 根据索引和_id查询数据* @param indexName* @param id* @return*/public Map<String, Object> getDocById(String indexName, String id) {GetRequest request = GetRequest.of(g -> g.index(indexName).id(id));try {GetResponse<Map> response = client.get(request, Map.class);if(response.found()){return response.source();}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return null;}/*** 根据索引和_id查询数据* @param indexName* @param id* @return*/public JSONObject getDocInfoById(String indexName, String id) {GetRequest request = GetRequest.of(g -> g.index(indexName).id(id));try {GetResponse<JSONObject> response = client.get(request, JSONObject.class);if(response.found()){return response.source();}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return null;}/*** 根据索引和_id查询数据，并过滤掉无需返回的字段* @param indexName* @param id* @param excludes* @return*/public JSONObject getDocInfoById(String indexName, String id, String [] excludes) {GetRequest request = GetRequest.of(g -> g.index(indexName).id(id).sourceExcludes(Arrays.asList(excludes)));try {GetResponse<JSONObject> response = client.get(request, JSONObject.class);if(response.found()){return response.source();}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return null;}/*** 根据索引和_id查询数据，并过指定要返回的字段* @param indexName* @param id* @param includes* @return*/public JSONObject getDocInfoByIdWithIncludes(String indexName, String id, String [] includes) {GetRequest request = GetRequest.of(g -> g.index(indexName).id(id).sourceIncludes(Arrays.asList(includes)));try {GetResponse<JSONObject> response = client.get(request, JSONObject.class);if(response.found()){return response.source();}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return null;}/*** 判断数据是否存在* @param indexName* @param id* @return*/public boolean exists(String indexName, String id) {GetRequest request = GetRequest.of(g -> g.index(indexName).id(id));try {GetResponse<JSONObject> response = client.get(request, JSONObject.class);return response.found();} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 根据索引和_id删除数据* @param indexName* @param id* @return*/public boolean deleteDocById(String indexName, String id) {DeleteRequest request = DeleteRequest.of(a -> a.index(indexName).id(id));try {DeleteResponse response = client.delete(request);if(response != null && response.result() != null){return Result.Deleted.jsonValue().equals(response.result().jsonValue());}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 更新数据* @param indexName* @param id* @param newDoc* @return*/public boolean updateDocById(String indexName, String id, JSONObject newDoc) {UpdateRequest request = UpdateRequest.of(r -> r.id(id).index(indexName).doc(newDoc));request.refresh();try {UpdateResponse response = client.update(request, JSONObject.class);if(response != null && response.result() != null){return Result.Updated.jsonValue().equals(response.result().jsonValue());}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 对输入的text使用analyzerName进行分词，返回分词后的词项* @param analyzerName* @param text* @return*/public List<AnalyzeToken> analyze(String analyzerName, String text){AnalyzeRequest analyzeRequest = new AnalyzeRequest.Builder().analyzer(analyzerName).text(text).build();AnalyzeResponse response = null;try {response = client.indices().analyze(analyzeRequest);} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return response.tokens();}/*** 批量删除* @param requestList* @return*/public boolean bulkDelete(List<DeleteRequest> requestList){List<BulkOperation> ops = requestList.stream().map(req -> BulkOperation.of(op -> op.delete(d -> d.id(req.id()).index(req.index())))).collect(Collectors.toList());try {BulkResponse response = client.bulk(r -> r.operations(ops));if(response != null ){return true;}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 批量更新* @param requestList* @return*/public boolean bulkUpdate(List<UpdateRequest> requestList){List<BulkOperation> ops = requestList.stream().map(req -> BulkOperation.of(op -> op.update(d -> d.id(req.id()).index(req.index()).action(a -> a.doc(req.doc()))))).collect(Collectors.toList());try {BulkResponse response = client.bulk(r -> r.operations(ops));if(response != null ){return true;}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 批量插入数据* @param requestList* @return*/public boolean bulkInsert(List<IndexRequest> requestList){List<BulkOperation> ops = requestList.stream().map(req -> BulkOperation.of(op -> op.index(i -> i.document(req.document()).index(req.index())))).collect(Collectors.toList());try {BulkResponse response = client.bulk(r -> r.operations(ops));if(response != null ){return true;}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 通过脚本批量更新* @param index* @param query* @param script* @return*/public boolean updateByquery(String index, BoolQuery query, String script){try {UpdateByQueryResponse response = client.updateByQuery(q -> q.index(index).query(query._toQuery()).script(s -> s.source(script)));if(response != null ){return true;}} catch (IOException e) {e.printStackTrace();}finally {client.shutdown();}return false;}/*** 检索* @param indexName* @param pageNo* @param pageSize* @param sortField* @param sortOrder* @param boolQuery* @return*/public SearchResponse search(String indexName, Integer pageNo, Integer pageSize,String sortField, SortOrder sortOrder, BoolQuery boolQuery) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).from((pageNo - 1) * pageSize).size(pageSize).sort(s -> s.field(f -> f.field(sortField).order(sortOrder))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, Integer pageNo, Integer pageSize,String sortField, SortOrder sortOrder, BoolQuery boolQuery, String[] excludes) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).source(a -> a.filter(f -> f.excludes(Arrays.asList(excludes)))).from((pageNo - 1) * pageSize).size(pageSize).sort(s -> s.field(f -> f.field(sortField).order(sortOrder))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, Integer pageNo, Integer pageSize, BoolQuery boolQuery,String sortField, SortOrder sortOrder, String sortField2, SortOrder sortOrder2) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).from((pageNo - 1) * pageSize).size(pageSize).sort(s -> s.field(f -> f.field(sortField).order(sortOrder).field(sortField2).order(sortOrder2))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, Integer pageNo, Integer pageSize, BoolQuery boolQuery,String sortField, SortOrder sortOrder, String sortField2, SortOrder sortOrder2,String[] excludes) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).source(a -> a.filter(f -> f.excludes(Arrays.asList(excludes)))).from((pageNo - 1) * pageSize).size(pageSize).sort(s -> s.field(f -> f.field(sortField).order(sortOrder).field(sortField2).order(sortOrder2))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, Integer pageNo, Integer pageSize, BoolQuery boolQuery,String sortField, SortOrder sortOrder, String[] includes, String[] excludes) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).source(a -> a.filter(f -> f.excludes(Arrays.asList(excludes)).includes(Arrays.asList(includes)))).from((pageNo - 1) * pageSize).size(pageSize).sort(s -> s.field(f -> f.field(sortField).order(sortOrder))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, Integer pageNo, Integer pageSize,  BoolQuery boolQuery,String sortField, SortOrder sortOrder, String time) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).from((pageNo - 1) * pageSize).size(pageSize).scroll(new Time.Builder().time(time).build()).sort(s -> s.field(f -> f.field(sortField).order(sortOrder))).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}/*** 查询符合条件的数据条数* @param indexName* @param boolQuery* @return*/public CountResponse count(String indexName, BoolQuery boolQuery) {try {return client.count(c -> c.index(indexName).query(q -> q.bool(boolQuery)));} catch (IOException e) {e.printStackTrace();}return null;}public SearchResponse search(String indexName, BoolQuery boolQuery) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}public SearchResponse search(String indexName, BoolQuery boolQuery, int size) {SearchRequest request = new SearchRequest.Builder().index(indexName).query(q -> q.bool(boolQuery)).size(size).build();SearchResponse response = null;try {response = client.search(request, JSONObject.class);} catch (IOException e) {e.printStackTrace();}return response;}
}

如果本文对你有帮助，请点赞、收藏 + 关注，谢谢！！（本文将持续更新）


http://www.ppmy.cn/server/157333.html

相关文章

如何升级node.js版本

如何升级node.js版本

升级Node.js可以通过多种方式来完成&#xff0c;以下是四种常见的方法&#xff1a; 方法一&#xff1a;使用Node.js官方安装程序 访问Node.js的官方网站&#xff0c;下载对应你操作系统的最新版本安装程序。通常&#xff0c;你可以 https://nodejs.org/en/download 找到你需…
阅读更多...
第 2 章 HTML 网页编程基础教案

第 2 章 HTML 网页编程基础教案

谢从华&#xff0c;高蕴梅 著.Web前端设计基础入门——HTML5、CSS3、JavaScript&#xff08;微课视频版&#xff09;,2023, 清华大学出版社. ISBN&#xff1a;9787302641261. 一、教学目标&#xff08;Objectives&#xff09; 1. 知识目标 - 学生能够理解 HTML 的基本概念&…
阅读更多...
React 如何进行路由变化监听

React 如何进行路由变化监听

一、使用react-router库&#xff08;以react-router-dom为例&#xff09; 1. 历史&#xff08;history&#xff09;对象监听 1.1 原理 react-router内部使用history对象来管理路由历史记录。可以通过访问history对象来监听路由变化。在基于类的组件中&#xff0c;可以通过组…
阅读更多...
通义灵码在跨领域应用拓展之物联网篇

通义灵码在跨领域应用拓展之物联网篇

目录 一.引言 二.通义灵码简介 三.通义灵码在物联网领域的设备端应用 1.传感器数据采集 (1).不同类型传感器的数据读取 (2).数据转换与预处理 2.设备控制指令接收和执行 (1).指令解析与处理 (2).设备动作执行 四.通义灵码在物联网领域的云端平台应用 1.数据存储和管…
阅读更多...
计算机网络的定义与发展历程

计算机网络的定义与发展历程

计算机网络的定义 计算机网络是指通过通信设备和传输介质将分布在不同地点的计算机及其相关设备&#xff08;如打印机、服务器等&#xff09;连接起来&#xff0c;按照一定的通信协议进行数据交换与资源共享的系统。计算机网络的基本功能包括&#xff1a;信息的传输、资源共享…
阅读更多...
欧拉公式和傅里叶变换

欧拉公式和傅里叶变换

注&#xff1a;英文引文机翻&#xff0c;未校。 中文引文未整理去重&#xff0c;如有异常&#xff0c;请看原文。 Euler’s Formula and Fourier Transform Posted byczxttkl October 7, 2018 Euler’s formula states that e i x cos ⁡ x i sin ⁡ x e^{ix} \cos{x} i …
阅读更多...
深入 Solana 共识 - 从分叉到最终确定性

深入 Solana 共识 - 从分叉到最终确定性

介绍 ​ 共识是每个区块链构建的基本要素。它确保交易&#xff0c;无论是花费代币还是执行智能合约&#xff0c;都能在没有中央权威的情况下得到正确的验证和执行。设计和构建共识协议有很多方法。在这篇博客文章中&#xff0c;我们将详细介绍 Solana 共识协议。本文是为那些对…
阅读更多...
VSCode 插件

VSCode 插件

VSCode 插件 1. GitHub Copilot - AI 代码助手 功能&#xff1a;根据上下文提供实时代码补全&#xff0c;支持自然语言转代码&#xff0c;提供符合现代编程规范的建议。进阶技巧&#xff1a; 使用快捷键 Alt ] 切换多个建议。写注释时&#xff0c;描述业务逻辑而不是具体实现…
阅读更多...
最新文章

Copyright @ 2022~2023