springboot2+es7使用RestHighLevelClient的示例代碼

第springboot2+es7使用RestHighLevelClient的示例代碼目錄一、引入依賴jar二、perties配置三、使用其它由于spring和es的集成并不是特別友好，es的高低版本兼容問題、api更新頻率高等問題，所以我選擇是官網(wǎng)提供的原生Client(RestHighLevelClient)，但又不想去關注es的配置類以及和spring的集成配置、jar包沖突等問題，所以使用spring-boot-starter-data-elasticsearch。

一、引入依賴jar

dependency

groupIdorg.springframework.boot/groupId

artifactIdspring-boot-starter-data-elasticsearch/artifactId

/dependency

二、perties配置

spring.elasticsearch.rest.uris=:9200,:9201,:9202

spring.elasticsearch.rest.connection-timeout=5s

spring.elasticsearch.rest.read-timeout=30s

.springframework.data.convert.CustomConversions=error

spring-boot-starter-data-elasticsearch中自動裝配es的配置類：ElasticsearchRestClientAutoConfiguration、ElasticsearchRestClientProperties。

ElasticsearchRestClientAutoConfiguration：

@ConditionalOnClass({RestHighLevelClient.class})

@ConditionalOnMissingBean({RestClient.class})

@EnableConfigurationProperties({ElasticsearchRestClientProperties.class})

publicclassElasticsearchRestClientAutoConfiguration{

@Configuration(

proxyBeanMethods=false

)

@ConditionalOnMissingBean({RestHighLevelClient.class})

staticclassRestHighLevelClientConfiguration{

RestHighLevelClientConfiguration(){

}

@Bean

RestHighLevelClientelasticsearchRestHighLevelClient(RestClientBuilderrestClientBuilder){

returnnewRestHighLevelClient(restClientBuilder);

}

}

@Configuration(

proxyBeanMethods=false

)

@ConditionalOnMissingBean({RestClientBuilder.class})

staticclassRestClientBuilderConfiguration{

RestClientBuilderConfiguration(){

}

@Bean

RestClientBuilderCustomizerdefaultRestClientBuilderCustomizer(ElasticsearchRestClientPropertiesproperties){

returnnewElasticsearchRestClientAutoConfiguration.DefaultRestClientBuilderCustomizer(properties);

}

@Bean

RestClientBuilderelasticsearchRestClientBuilder(ElasticsearchRestClientPropertiesproperties,ObjectProviderRestClientBuilderCustomizerbuilderCustomizers){

HttpHost[]hosts=(HttpHost[])properties.getUris().stream().map(this::createHttpHost).toArray((x$0)-{

returnnewHttpHost[x$0];

});

RestClientBuilderbuilder=RestClient.builder(hosts);

builder.setHttpClientConfigCallback((httpClientBuilder)-{

builderCustomizers.orderedStream().forEach((customizer)-{

customizer.customize(httpClientBuilder);

});

returnhttpClientBuilder;

});

builder.setRequestConfigCallback((requestConfigBuilder)-{

builderCustomizers.orderedStream().forEach((customizer)-{

customizer.customize(requestConfigBuilder);

});

returnrequestConfigBuilder;

});

builderCustomizers.orderedStream().forEach((customizer)-{

customizer.customize(builder);

});

returnbuilder;

}

privateHttpHostcreateHttpHost(Stringuri){

try{

returnthis.createHttpHost(URI.create(uri));

}catch(IllegalArgumentExceptionvar3){

returnHttpHost.create(uri);

}

}

privateHttpHostcreateHttpHost(URIuri){

if(!StringUtils.hasLength(uri.getUserInfo())){

returnHttpHost.create(uri.toString());

}else{

try{

returnHttpHost.create((newURI(uri.getScheme(),(String)null,uri.getHost(),uri.getPort(),uri.getPath(),uri.getQuery(),uri.getFragment())).toString());

}catch(URISyntaxExceptionvar3){

thrownewIllegalStateException(var3);

}

}

}

}

}

ElasticsearchRestClientProperties：

@ConfigurationProperties(

prefix="spring.elasticsearch.rest"

publicclassElasticsearchRestClientProperties{

privateListStringuris=newArrayList(Collections.singletonList("http://localhost:9200"));

privateStringusername;

privateStringpassword;

privateDurationconnectionTimeout=Duration.ofSeconds(1L);

privateDurationreadTimeout=Duration.ofSeconds(30L);

publicElasticsearchRestClientProperties(){

}

publicListStringgetUris(){

returnthis.uris;

}

publicvoidsetUris(ListStringuris){

this.uris=uris;

}

publicStringgetUsername(){

returnthis.username;

}

publicvoidsetUsername(Stringusername){

this.username=username;

}

publicStringgetPassword(){

returnthis.password;

}

publicvoidsetPassword(Stringpassword){

this.password=password;

}

publicDurationgetConnectionTimeout(){

returnthis.connectionTimeout;

}

publicvoidsetConnectionTimeout(DurationconnectionTimeout){

this.connectionTimeout=connectionTimeout;

}

publicDurationgetReadTimeout(){

returnthis.readTimeout;

}

publicvoidsetReadTimeout(DurationreadTimeout){

this.readTimeout=readTimeout;

}

}

三、使用

ES基本操作持久層

/**

*es持久層

*@authoryangzihe

*@date2025/1/24

@Repository

@Slf4j

publicclassEsRepository{

@Resource

privateRestHighLevelClienthighLevelClient;

/**

*判斷索引是否存在

*/

publicbooleanexistIndex(Stringindex){

try{

returnhighLevelClient.indices().exists(newGetIndexRequest(index),RequestOptions.DEFAULT);

}catch(IOExceptione){

log.error("es持久層異常！index={}",index,e);

}

returnBoolean.FALSE;

}

/**

*創(chuàng)建索引

*/

publicbooleancreateIndex(Stringindex,MapString,ObjectcolumnMap){

if(existIndex(index)){

returnBoolean.FALSE;

}

CreateIndexRequestrequest=newCreateIndexRequest(index);

if(columnMap!=nullcolumnMap.size()0){

MapString,Objectsource=newHashMap();

source.put("properties",columnMap);

request.mapping(source);

}

try{

highLevelClient.indices().create(request,RequestOptions.DEFAULT);

returnBoolean.TRUE;

}catch(IOExceptione){

log.error("es持久層異常！index={},columnMap={}",index,columnMap,e);

}

returnBoolean.FALSE;

}

/**

*刪除索引

*/

publicbooleandeleteIndex(Stringindex){

try{

if(existIndex(index)){

AcknowledgedResponseresponse=highLevelClient.indices().delete(newDeleteIndexRequest(index),RequestOptions.DEFAULT);

returnresponse.isAcknowledged();

}

}catch(Exceptione){

log.error("es持久層異常！index={}",index,e);

}

returnBoolean.FALSE;

}

/**

*數(shù)據(jù)新增

*/

publicbooleaninsert(Stringindex,StringjsonString){

IndexRequestindexRequest=newIndexRequest(index);

indexRequest.id(newSnowflake().nextIdStr());

indexRequest.source(jsonString,XContentType.JSON);

try{

("indexRequest={}",indexRequest);

IndexResponseindexResponse=highLevelClient.index(indexRequest,RequestOptions.DEFAULT);

("indexResponse={}",indexResponse);

returnBoolean.TRUE;

}catch(IOExceptione){

log.error("es持久層異常！index={},jsonString={}",index,jsonString,e);

}

returnBoolean.FALSE;

}

/**

*數(shù)據(jù)更新，可以直接修改索引結構

*/

publicbooleanupdate(Stringindex,MapString,ObjectdataMap){

UpdateRequestupdateRequest=newUpdateRequest(index,dataMap.remove("id").toString());

updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);

updateRequest.doc(dataMap);

try{

highLevelClient.update(updateRequest,RequestOptions.DEFAULT);

}catch(IOExceptione){

log.error("es持久層異常！index={},dataMap={}",index,dataMap,e);

returnBoolean.FALSE;

}

returnBoolean.TRUE;

}

/**

*刪除數(shù)據(jù)

*/

publicbooleandelete(Stringindex,Stringid){

DeleteRequestdeleteRequest=newDeleteRequest(index,id);

try{

highLevelClient.delete(deleteRequest,RequestOptions.DEFAULT);

}catch(IOExceptione){

log.error("es持久層異常！index={},id={}",index,id,e);

returnBoolean.FALSE;

}

returnBoolean.TRUE;

}

}

ES查詢持久層

/**

*es查詢持久層

*@authoryangzihe

*@date2025/1/25

@Repository

@Slf4j

publicclassEsSearchRepository{

@Resource

privateRestHighLevelClienthighLevelClient;

/**

*分頁查詢

*

*@paramqueryPO分頁查詢對象

*

*@return分頁查詢結果

*/

publicEsQueryRespPOMapString,ObjectsearchPage(EsQueryReqPOqueryPO){

//默認分頁參數(shù)設置

if(queryPO.getPageNum()==null){

queryPO.setPageNum(1);

}

if(queryPO.getPageSize()==null){

queryPO.setPageSize(10);

}

//設置索引

SearchRequestsearchRequest=newSearchRequest(queryPO.getIndex());

//封裝查詢源對象

SearchSourceBuildersourceBuilder=newSearchSourceBuilder();

searchRequest.source(sourceBuilder);

//查詢條件

sourceBuilder.query(queryPO.getQuery());

//排序字段

if(StringUtils.isNotBlank(queryPO.getSortField())queryPO.getSort()!=null){

FieldSortBuilderorder=newFieldSortBuilder(queryPO.getSortField()).order(queryPO.getSort());

sourceBuilder.sort(order);

}

//開始行數(shù)，默認0

sourceBuilder.from((queryPO.getPageNum()-1)*queryPO.getPageSize());

//頁大小，默認10

sourceBuilder.size(queryPO.getPageSize());

//查詢結果

SearchResponsesearchResponse=null;

try{

//("es查詢請求對象：searchRequest={}",searchRequest);

("es查詢請求對象source：sourceBuilder={}",searchRequest.source());

//執(zhí)行搜索

searchResponse=highLevelClient.search(searchRequest,RequestOptions.DEFAULT);

("es查詢響應結果：searchResponse={}",searchResponse);

}catch(IOExceptione){

log.error("es查詢，IO異常！searchRequest={}",searchRequest,e);

//異常處理

returnEsQueryRespPO.error("es查詢，IO異常！");

}

if(RestStatus.OK.equals(searchResponse.status())){

//解析對象

SearchHit[]hits=searchResponse.getHits().getHits();

//獲取source

ListMapString,ObjectsourceList=Arrays.stream(hits).map(SearchHit::getSourceAsMap).collect(Collectors.toList());

longtotalHits=searchResponse.getHits().getTotalHits().value;

returnEsQueryRespPO.success(sourceList,queryPO.getPageNum(),queryPO.getPageSize(),totalHits);

}else{

log.error("es查詢返回的狀態(tài)碼異常！searchResponse.status={},searchRequest={}",searchResponse.status(),searchRequest);

returnEsQueryRespPO.error("es查詢返回的狀態(tài)碼異常！");

}

}

/**

*聚合的分頁查詢

*

*@paramqueryPO查詢請求對象

*

*@return聚合分頁查詢結果

*/

publicEsQueryRespPOAggregationBucketPOsearchAggregation(EsQueryReqPOqueryPO){

//設置索引

SearchRequestsearchRequest=newSearchRequest(queryPO.getIndex());

//封裝查詢源對象

SearchSourceBuildersourceBuilder=newSearchSourceBuilder();

searchRequest.source(sourceBuilder);

//查詢條件

sourceBuilder.query(queryPO.getQuery());

//排序字段

if(StringUtils.isNotBlank(queryPO.getSortField())queryPO.getSort()!=null){

FieldSortBuilderorder=newFieldSortBuilder(queryPO.getSortField()).order(queryPO.getSort());

sourceBuilder.sort(order);

}

//頁大小0，只返回聚合結果

sourceBuilder.size(0);

//設置聚合查詢,可以設置多個聚合查詢條件，只要聚合查詢命名不同就行

//聚合分組條件,groupby

sourceBuilder.aggregation(queryPO.getTermsAggregation());

//聚合統(tǒng)計條件,count分組后的數(shù)據(jù)，計算分組后的總大小

sourceBuilder.aggregation(queryPO.getCardinalityAggregation());

//查詢結果

SearchResponsesearchResponse=null;

try{

//("es查詢請求對象：searchRequest={}",searchRequest);

("es查詢請求對象source：sourceBuilder={}",searchRequest.source());

//執(zhí)行搜索

searchResponse=highLevelClient.search(searchRequest,RequestOptions.DEFAULT);

("es查詢響應結果：searchResponse={}",searchResponse);

}catch(IOExceptione){

log.error("es查詢，IO異常！searchRequest={}",searchRequest,e);

returnEsQueryRespPO.error("es查詢，IO異常！");

}

if(RestStatus.OK.equals(searchResponse.status())){

//解析對象

Aggregationsaggregations=searchResponse.getAggregations();

longdocTotal=searchResponse.getHits().getTotalHits().value;

//遍歷terms聚合結果

Termsterms=aggregations.get(queryPO.getTermsAggregation().getName());

ListAggregationBucketPObucketList=terms.getBuckets().stream().map(bucket-{

Stringkey=bucket.getKeyAsString();

longdocCount=bucket.getDocCount();

returnnewAggregationBucketPO(key,docCount,docTotal);

}).collect(Collectors.toList());

//總數(shù)量

Cardinalitycardinality=aggregations.get(queryPO.getCardinalityAggregation().getName());

longtotalHits=cardinality.getValue();

returnEsQueryRespPO.success(bucketList,queryPO.getPageNum(),queryPO.getPageSize(),totalHits);

}else{

log.error("es查詢返回的狀態(tài)碼異常！searchResponse.status={},searchRequest={}",searchResponse.status(),searchRequest);

returnEsQueryRespPO.error("es查詢返回的狀態(tài)碼異常！");

}

}

}

其中，EsQueryReqPO、EsQueryRespPO、AggregationBucketPO等類如下：

/**

*es查詢請求對象

@Data

publicclassEsQueryReqPO{

/**

*索引名

*/

String[]index;

/**

*查詢條件

*/

QueryBuilderquery;

/**

*排序字段

*/

StringsortField;

/**

*排序方式SortOrder.ASC、SortOrder.DESC

*/

SortOrdersort;

/**

*頁數(shù)

*/

privateIntegerpageNum;

/**

*頁大小

*/

privateIntegerpageSize;

/**

*聚合分組條件,groupby

*/

privateTermsAggregationBuildertermsAggregation;

/**

*聚合統(tǒng)計條件,count分組后的數(shù)據(jù)

*/

privateCardinalityAggregationBuildercardinalityAggregation;

*es分頁響應對象

*@authoryangzihe

*@date2025/1/25

@Data

@NoArgsConstructor

@AllArgsConstructor

publicclassEsQueryRespPOT{

/**

*是否成功

*/

privateBooleansuccess;

/**

*信息

*/

privateStringmessage;

/**

*頁數(shù)

*/

privateIntegerpageNum;

/**

*頁大小

*/

privateIntegerpageSize;

/**

*總大小

*/

privateLongtotalSize;

/**

*數(shù)據(jù)

*/

privateListTsourceList;

publicstaticTEsQueryRespPOTsuccess(ListTsourceList,IntegerpageNum,IntegerpageSize,

LongtotalSize){

EsQueryRespPOTesQueryRespPO=newEsQueryRespPO();

esQueryRespPO.setSuccess(true);

esQueryRespPO.setSourceList(sourceList);

esQueryRespPO.setPageNum(pageNum);

esQueryRespPO.setPageSize(pageSize);

esQueryRespPO.setTotalSize(totalSize);

returnesQueryRespPO;

}

publicstaticEsQueryRespPOerror(){

EsQueryRespPOesQueryRespPO=newEsQueryRespPO();

esQueryRespPO.setSuccess(false);

esQueryRespPO.setMessage("es查詢異常");

returnesQueryRespPO;

}

publicstaticEsQueryRespPOerror(Stringmessage){

EsQueryRespPOesQueryRespPO=newEsQueryRespPO();

esQueryRespPO.setSuccess(false);

esQueryRespPO.setMessage(message);

returnesQueryRespPO;

}

}

/**

*聚合桶對象

*@authoryangzihe

*@date2025/1/26

@Data

@NoArgsConstructor

@AllArgsConstructor

publicclassAggregationBucketPO{

/**

*聚合Bucket的key名

*/

privateStringkey;

/**

*聚合Bucket的文檔數(shù)量

*/

privateLongdocCount;

/**

*文檔總數(shù)量

*/

privateLongdocTotal;

}

ES多級(二級)聚合分桶查詢

importmon.exception.EsException;

importcom.yy.armor.manager.persist.es.po.AggregationBucketPO;

importcom.yy.armor.manager.persist.es.po.EsMultiAggregationReqPO;

importjava.io.IOException;

importjava.util.List;

importjava.util.stream.Collectors;

importjavax.annotation.Resource;

importlombok.extern.slf4j.Slf4j;

importmons.collections4.CollectionUtils;

importmons.lang3.StringUtils;

importpress.utils.Lists;

importorg.elasticsearch.action.search.SearchRequest;

importorg.elasticsearch.action.search.SearchResponse;

importorg.elasticsearch.client.RequestOptions;

importorg.elasticsearch.client.RestHighLevelClient;

importorg.elasticsearch.rest.RestStatus;

importorg.elasticsearch.search.aggregations.AggregationBuilders;

importorg.elasticsearch.search.aggregations.bucket.terms.Terms;

importorg.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;

importorg.elasticsearch.search.builder.SearchSourceBuilder;

importorg.elasticsearch.search.sort.FieldSortBuilder;

importorg.springframework.stereotype.Repository;

@Repository

@Slf4j

publicclassEsSearchRepository{

@Resource

privateRestHighLevelClienthighLevelClient;

/**

*多級聚合查詢(二級聚合)

*

*@paramreqPO查詢請求對象

*

*@return聚合查詢結果

*/

publicListAggregationBucketPOsearchMultiAggregation(EsMultiAggregationReqPOreqPO){

//設置索引

SearchRequestsearchRequest=newSearchRequest(reqPO.getIndex());

//封裝查詢源對象

SearchSourceBuildersourceBuilder=newSearchSourceBuilder();

searchRequest.source(sourceBuilder);

//查詢條件

sourceBuilder.query(reqPO.getQuery());

//排序字段

if(StringUtils.isNotBlank(reqPO.getSortField())reqPO.getSort()!=null){

FieldSortBuilderorder=newFieldSortBuilder(reqPO.getSortField()).order(reqPO.getSort());

sourceBuilder.sort(order);

}

//頁大小0，只返回聚合結果

sourceBuilder.size(0);

//聚合分桶。創(chuàng)建terms桶聚合，聚合名字=terms_by_XXX,字段=XXX

TermsAggregationBuildertermsAggregationBuilder=AggregationBuilders.terms("terms_by_"+reqPO.getField()).field(reqPO.getField());

if(reqPO.getFieldSize()!=null){

termsAggregationBuilder.size(reqPO.getFieldSize());

}

//二級聚合分桶

TermsAggregationBuildersubTermsAggregationBuilder=AggregationBuilders.terms("terms_by_"+reqPO.getSubField()).field(reqPO.getSubField());

if(reqPO.getSubFieldSize()!=null){

subTermsAggregationBuilder.size(reqPO.getSubFieldSize());

}

termsAggregationBuilder.subAggregation(subTermsAggregationBuilder);

//聚合分組條件

sourceBuilder.aggregation(termsAggregationBuilder);

//查詢結果

SearchResponsesearchResponse=null;

try{

("es查詢請求對象source：sourceBuilder={}",searchRequest.source());

//執(zhí)行搜索

searchResponse=highLevelClient.search(searchRequest,RequestOptions.DEFAULT);

("es查詢響應結果：searchResponse={}",searchResponse);

}catch(IOExceptione){

log.error("es查詢，IO異常！searchRequest={}",searchRequest,e);

thrownewEsException("es查詢，IO異常！");

}

if(RestStatus.OK.equals(searchResponse.status())){

//遍歷terms聚合結果

Termsterms=searchResponse.getAggregations().get(termsAggregationBuilder.getName());

ListAggregationBucketPObucketList=terms.getBuckets().stream().map(bucket-{

//一級聚合分桶的數(shù)據(jù)

Stringkey=bucket.getKeyAsString();

longdocCount=bucket.getDocCount();

//二級聚合分桶的數(shù)據(jù)

TermssubTerms=bucket.getAggregations().get(subTermsAggregationBuilder.getName());

ListAggregationBucketPOsubBucketList=convertTerms(subTerms);

returnnewAggregationBucketPO(key,docCount,subBucketList);

}).collect(Collectors.toList());

returnbucketList;

}else{

log.error("es查詢返回的狀態(tài)碼異常！searchResponse.status={},searchRequest={}",searchResponse.status(),searchRequest);

thrownewEsException("es查詢返回的狀態(tài)碼異常！");

}

}

privateListAggregationBucketPOconvertTerms(Termsterms){

if(CollectionUtils.isEmpty(terms.getBuckets())){

returnLists.newArrayList();

}

returnterms.getBuckets().stream().map(bucket-{

Stringkey=bucket.getKeyAsString();

longdocCount=bucket.getDocCount();

returnnewAggregationBucketPO(key,docCount);

}).collect(Collectors.toList());

}

}

其中，EsMultiAggregationReqPO、AggregationBucketPO類如下：

@Data

publicclassEsMultiAggregationReqPO{

/**

*索引名

*/

String[]index;

/**

*查詢條件

*/

QueryBuilderquery;

/**

*聚合分桶字段

*/

privateStringfield;

/**

*二級聚合分桶字段

*/

privateStringsubField;

/**

*聚合分桶大小，非必傳

*/

privateIntegerfieldSize;

/**

*二級聚合分桶大小，非必傳

*/

privateIntegersubFieldSize;

/**

*排序字段，非必傳

*/

StringsortField;

/**

*排序方式SortOrder.ASC、SortOrder.DESC，非必傳

*/

SortOrdersort;

}

@Data

@NoArgsConstructor

@AllArgsConstructor

publicclassAggregationBucketPO{

*聚合Bucket的key名

privateStringkey;

*聚合Bucket的文檔數(shù)量

privateLongdocCount;

*子桶集合

privateListAggregationBucketPOsubBucketList;

publicAggregationBucketPO(Stringkey,LongdocCount){

this.key=key;

this.docCount=docCount;

二級聚合分桶測試代碼

@PostConstruct

privatevoidinit(){

//查詢對象的封裝

EsMultiAggregationReqPOreqPO=newEsMultiAggregationReqPO();

reqPO.setIndex(newString[]{"test_log"});

ListLongids=Lists.newArrayList();

ids.add(140L);

ids.add(141L);

ids.add(142L);

QueryBuilderqueryBuilder4=QueryBuilders.termsQuery("eventId",ids);

BoolQueryBuilderqueryBuilder=QueryBuilders.boolQuery().must(queryBuilder4);

reqPO.setQuery(queryBuilder);

reqPO.setField("eventId");

reqPO.setFieldSize(9999);

reqPO.setSubField("riskFlag");

//執(zhí)行查詢

ListAggregationBucketPOesQueryRespPO=searchMultiAggregation(reqPO);

System.out.println("esQueryRespPO="+esQueryRespPO);

}

其它

如果沒有用spring-boot-starter-data-elasticsearch來自動