springboot2+es7에서 RestHighLevelClient를 사용하는 방법

 org.springframework.boot spring-boot-starter-data-elasticsearch 

spring.elasticsearch.rest.uris=http://127.0.0.1:9200,http://127.0.0.1:9201,http://127.0.0.1:9202 spring.elasticsearch.rest.connection-timeout=5s spring.elasticsearch.rest.read-timeout=30s logging.level.org.springframework.data.convert.CustomConversions=error

@ConditionalOnClass({RestHighLevelClient.class}) @ConditionalOnMissingBean({RestClient.class}) @EnableConfigurationProperties({ElasticsearchRestClientProperties.class}) public class ElasticsearchRestClientAutoConfiguration { @Configuration( proxyBeanMethods = false ) @ConditionalOnMissingBean({RestHighLevelClient.class}) static class RestHighLevelClientConfiguration { RestHighLevelClientConfiguration() { } @Bean RestHighLevelClient elasticsearchRestHighLevelClient(RestClientBuilder restClientBuilder) { return new RestHighLevelClient(restClientBuilder); } } @Configuration( proxyBeanMethods = false ) @ConditionalOnMissingBean({RestClientBuilder.class}) static class RestClientBuilderConfiguration { RestClientBuilderConfiguration() { } @Bean RestClientBuilderCustomizer defaultRestClientBuilderCustomizer(ElasticsearchRestClientProperties properties) { return new ElasticsearchRestClientAutoConfiguration.DefaultRestClientBuilderCustomizer(properties); } @Bean RestClientBuilder elasticsearchRestClientBuilder(ElasticsearchRestClientProperties properties, ObjectProvider builderCustomizers) { HttpHost[] hosts = (HttpHost[])properties.getUris().stream().map(this::createHttpHost).toArray((x$0) -> { return new HttpHost[x$0]; }); RestClientBuilder builder = RestClient.builder(hosts); builder.setHttpClientConfigCallback((httpClientBuilder) -> { builderCustomizers.orderedStream().forEach((customizer) -> { customizer.customize(httpClientBuilder); }); return httpClientBuilder; }); builder.setRequestConfigCallback((requestConfigBuilder) -> { builderCustomizers.orderedStream().forEach((customizer) -> { customizer.customize(requestConfigBuilder); }); return requestConfigBuilder; }); builderCustomizers.orderedStream().forEach((customizer) -> { customizer.customize(builder); }); return builder; } private HttpHost createHttpHost(String uri) { try { return this.createHttpHost(URI.create(uri)); } catch (IllegalArgumentException var3) { return HttpHost.create(uri); } } private HttpHost createHttpHost(URI uri) { if (!StringUtils.hasLength(uri.getUserInfo())) { return HttpHost.create(uri.toString()); } else { try { return HttpHost.create((new URI(uri.getScheme(), (String)null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment())).toString()); } catch (URISyntaxException var3) { throw new IllegalStateException(var3); } } } } }

@ConfigurationProperties( prefix = "spring.elasticsearch.rest" ) public class ElasticsearchRestClientProperties { private List uris = new ArrayList(Collections.singletonList("http://localhost:9200")); private String username; private String password; private Duration connectionTimeout = Duration.ofSeconds(1L); private Duration readTimeout = Duration.ofSeconds(30L); public ElasticsearchRestClientProperties() { } public List getUris() { return this.uris; } public void setUris(List uris) { this.uris = uris; } public String getUsername() { return this.username; } public void setUsername(String username) { this.username = username; } public String getPassword() { return this.password; } public void setPassword(String password) { this.password = password; } public Duration getConnectionTimeout() { return this.connectionTimeout; } public void setConnectionTimeout(Duration connectionTimeout) { this.connectionTimeout = connectionTimeout; } public Duration getReadTimeout() { return this.readTimeout; } public void setReadTimeout(Duration readTimeout) { this.readTimeout = readTimeout; } }

/** * es持久层 * * @author yangzihe * @date 2022/1/24 */ @Repository @Slf4j public class EsRepository { @Resource private RestHighLevelClient highLevelClient; /** * 判断索引是否存在 */ public boolean existIndex(String index) { try { return highLevelClient.indices().exists(new GetIndexRequest(index), RequestOptions.DEFAULT); } catch (IOException e) { log.error("es持久层异常！index={}", index, e); } return Boolean.FALSE; } /** * 创建索引 */ public boolean createIndex(String index, Map columnMap) { if (existIndex(index)) { return Boolean.FALSE; } CreateIndexRequest request = new CreateIndexRequest(index); if (columnMap != null && columnMap.size() > 0) { Map source = new HashMap<>(); source.put("properties", columnMap); request.mapping(source); } try { highLevelClient.indices().create(request, RequestOptions.DEFAULT); return Boolean.TRUE; } catch (IOException e) { log.error("es持久层异常！index={}, columnMap={}", index, columnMap, e); } return Boolean.FALSE; } /** * 删除索引 */ public boolean deleteIndex(String index) { try { if (existIndex(index)) { AcknowledgedResponse response = highLevelClient.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT); return response.isAcknowledged(); } } catch (Exception e) { log.error("es持久层异常！index={}", index, e); } return Boolean.FALSE; } /** * 数据新增 */ public boolean insert(String index, String jsonString) { IndexRequest indexRequest = new IndexRequest(index); indexRequest.id(new Snowflake().nextIdStr()); indexRequest.source(jsonString, XContentType.JSON); try { log.info("indexRequest={}", indexRequest); IndexResponse indexResponse = highLevelClient.index(indexRequest, RequestOptions.DEFAULT); log.info("indexResponse={}", indexResponse); return Boolean.TRUE; } catch (IOException e) { log.error("es持久层异常！index={}, jsonString={}", index, jsonString, e); } return Boolean.FALSE; } /** * 数据更新，可以直接修改索引结构 */ public boolean update(String index, Map dataMap) { UpdateRequest updateRequest = new UpdateRequest(index, dataMap.remove("id").toString()); updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); updateRequest.doc(dataMap); try { highLevelClient.update(updateRequest, RequestOptions.DEFAULT); } catch (IOException e) { log.error("es持久层异常！index={}, dataMap={}", index, dataMap, e); return Boolean.FALSE; } return Boolean.TRUE; } /** * 删除数据 */ public boolean delete(String index, String id) { DeleteRequest deleteRequest = new DeleteRequest(index, id); try { highLevelClient.delete(deleteRequest, RequestOptions.DEFAULT); } catch (IOException e) { log.error("es持久层异常！index={}, id={}", index, id, e); return Boolean.FALSE; } return Boolean.TRUE; } }

/** * es查询持久层 * * @author yangzihe * @date 2022/1/25 */ @Repository @Slf4j public class EsSearchRepository { @Resource private RestHighLevelClient highLevelClient; /** * 分页查询 * * @param queryPO 分页查询对象 * * @return 分页查询结果 */ public EsQueryRespPO> searchPage(EsQueryReqPO queryPO) { // 默认分页参数设置 if (queryPO.getPageNum() == null) { queryPO.setPageNum(1); } if (queryPO.getPageSize() == null) { queryPO.setPageSize(10); } // 设置索引 SearchRequest searchRequest = new SearchRequest(queryPO.getIndex()); // 封装查询源对象 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); searchRequest.source(sourceBuilder); // 查询条件 sourceBuilder.query(queryPO.getQuery()); // 排序字段 if (StringUtils.isNotBlank(queryPO.getSortField()) && queryPO.getSort() != null) { FieldSortBuilder order = new FieldSortBuilder(queryPO.getSortField()).order(queryPO.getSort()); sourceBuilder.sort(order); } // 开始行数，默认0 sourceBuilder.from((queryPO.getPageNum() - 1) * queryPO.getPageSize()); // 页大小，默认10 sourceBuilder.size(queryPO.getPageSize()); // 查询结果 SearchResponse searchResponse = null; try { // log.info("es查询请求对象：searchRequest={}", searchRequest); log.info("es查询请求对象source：sourceBuilder={}", searchRequest.source()); // 执行搜索 searchResponse = highLevelClient.search(searchRequest, RequestOptions.DEFAULT); log.info("es查询响应结果：searchResponse={}", searchResponse); } catch (IOException e) { log.error("es查询，IO异常！searchRequest={}", searchRequest, e); // 异常处理 return EsQueryRespPO.error("es查询，IO异常！"); } if (RestStatus.OK.equals(searchResponse.status())) { // 解析对象 SearchHit[] hits = searchResponse.getHits().getHits(); // 获取source List> sourceList = Arrays.stream(hits).map(SearchHit::getSourceAsMap).collect(Collectors.toList()); long totalHits = searchResponse.getHits().getTotalHits().value; return EsQueryRespPO.success(sourceList, queryPO.getPageNum(), queryPO.getPageSize(), totalHits); } else { log.error("es查询返回的状态码异常！searchResponse.status={}, searchRequest={}", searchResponse.status(), searchRequest); return EsQueryRespPO.error("es查询返回的状态码异常！"); } } /** * 聚合的分页查询 * * @param queryPO 查询请求对象 * * @return 聚合分页查询结果 */ public EsQueryRespPO searchAggregation(EsQueryReqPO queryPO) { // 设置索引 SearchRequest searchRequest = new SearchRequest(queryPO.getIndex()); // 封装查询源对象 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); searchRequest.source(sourceBuilder); // 查询条件 sourceBuilder.query(queryPO.getQuery()); // 排序字段 if (StringUtils.isNotBlank(queryPO.getSortField()) && queryPO.getSort() != null) { FieldSortBuilder order = new FieldSortBuilder(queryPO.getSortField()).order(queryPO.getSort()); sourceBuilder.sort(order); } // 页大小0，只返回聚合结果 sourceBuilder.size(0); // 设置聚合查询,可以设置多个聚合查询条件，只要聚合查询命名不同就行 // 聚合分组条件, group by sourceBuilder.aggregation(queryPO.getTermsAggregation()); // 聚合统计条件, count分组后的数据，计算分组后的总大小 sourceBuilder.aggregation(queryPO.getCardinalityAggregation()); // 查询结果 SearchResponse searchResponse = null; try { // log.info("es查询请求对象：searchRequest={}", searchRequest); log.info("es查询请求对象source：sourceBuilder={}", searchRequest.source()); // 执行搜索 searchResponse = highLevelClient.search(searchRequest, RequestOptions.DEFAULT); log.info("es查询响应结果：searchResponse={}", searchResponse); } catch (IOException e) { log.error("es查询，IO异常！searchRequest={}", searchRequest, e); return EsQueryRespPO.error("es查询，IO异常！"); } if (RestStatus.OK.equals(searchResponse.status())) { // 解析对象 Aggregations aggregations = searchResponse.getAggregations(); long docTotal = searchResponse.getHits().getTotalHits().value; // 遍历terms聚合结果 Terms terms = aggregations.get(queryPO.getTermsAggregation().getName()); List bucketList = terms.getBuckets().stream().map(bucket -> { String key = bucket.getKeyAsString(); long docCount = bucket.getDocCount(); return new AggregationBucketPO(key, docCount, docTotal); }).collect(Collectors.toList()); // 总数量 Cardinality cardinality = aggregations.get(queryPO.getCardinalityAggregation().getName()); long totalHits = cardinality.getValue(); return EsQueryRespPO.success(bucketList, queryPO.getPageNum(), queryPO.getPageSize(), totalHits); } else { log.error("es查询返回的状态码异常！searchResponse.status={}, searchRequest={}", searchResponse.status(), searchRequest); return EsQueryRespPO.error("es查询返回的状态码异常！"); } } }

/** * es查询请求对象 */ @Data public class EsQueryReqPO { /** * 索引名 */ String[] index; /** * 查询条件 */ QueryBuilder query; /** * 排序字段 */ String sortField; /** * 排序方式 SortOrder.ASC、SortOrder.DESC */ SortOrder sort; /** * 页数 */ private Integer pageNum; /** * 页大小 */ private Integer pageSize; /** * 聚合分组条件, group by */ private TermsAggregationBuilder termsAggregation; /** * 聚合统计条件, count分组后的数据 */ private CardinalityAggregationBuilder cardinalityAggregation; } /** * es分页响应对象 * * @author yangzihe * @date 2022/1/25 */ @Data @NoArgsConstructor @AllArgsConstructor public class EsQueryRespPO { /** * 是否成功 */ private Boolean success; /** * 信息 */ private String message; /** * 页数 */ private Integer pageNum; /** * 页大小 */ private Integer pageSize; /** * 总大小 */ private Long totalSize; /** * 数据 */ private List sourceList; public static  EsQueryRespPO success(List sourceList, Integer pageNum, Integer pageSize, Long totalSize) { EsQueryRespPO esQueryRespPO = new EsQueryRespPO<>(); esQueryRespPO.setSuccess(true); esQueryRespPO.setSourceList(sourceList); esQueryRespPO.setPageNum(pageNum); esQueryRespPO.setPageSize(pageSize); esQueryRespPO.setTotalSize(totalSize); return esQueryRespPO; } public static EsQueryRespPO error() { EsQueryRespPO esQueryRespPO = new EsQueryRespPO(); esQueryRespPO.setSuccess(false); esQueryRespPO.setMessage("es查询异常"); return esQueryRespPO; } public static EsQueryRespPO error(String message) { EsQueryRespPO esQueryRespPO = new EsQueryRespPO(); esQueryRespPO.setSuccess(false); esQueryRespPO.setMessage(message); return esQueryRespPO; } }

/** * 聚合桶对象 * * @author yangzihe * @date 2022/1/26 */ @Data @NoArgsConstructor @AllArgsConstructor public class AggregationBucketPO { /** * 聚合Bucket的key名 */ private String key; /** * 聚合Bucket的文档数量 */ private Long docCount; /** * 文档总数量 */ private Long docTotal; }

import com.yy.armor.manager.common.exception.EsException; import com.yy.armor.manager.persist.es.po.AggregationBucketPO; import com.yy.armor.manager.persist.es.po.EsMultiAggregationReqPO; import java.io.IOException; import java.util.List; import java.util.stream.Collectors; import javax.annotation.Resource; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.shade.org.apache.commons.compress.utils.Lists; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.sort.FieldSortBuilder; import org.springframework.stereotype.Repository; @Repository @Slf4j public class EsSearchRepository { @Resource private RestHighLevelClient highLevelClient; /** * 多级聚合查询(二级聚合) * * @param reqPO 查询请求对象 * * @return 聚合查询结果 */ public List searchMultiAggregation(EsMultiAggregationReqPO reqPO) { // 设置索引 SearchRequest searchRequest = new SearchRequest(reqPO.getIndex()); // 封装查询源对象 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); searchRequest.source(sourceBuilder); // 查询条件 sourceBuilder.query(reqPO.getQuery()); // 排序字段 if (StringUtils.isNotBlank(reqPO.getSortField()) && reqPO.getSort() != null) { FieldSortBuilder order = new FieldSortBuilder(reqPO.getSortField()).order(reqPO.getSort()); sourceBuilder.sort(order); } // 页大小0，只返回聚合结果 sourceBuilder.size(0); // 聚合分桶。创建terms桶聚合，聚合名字=terms_by_XXX, 字段=XXX TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("terms_by_" + reqPO.getField()).field(reqPO.getField()); if (reqPO.getFieldSize() != null) { termsAggregationBuilder.size(reqPO.getFieldSize()); } // 二级聚合分桶 TermsAggregationBuilder subTermsAggregationBuilder = AggregationBuilders.terms("terms_by_" + reqPO.getSubField()).field(reqPO.getSubField()); if (reqPO.getSubFieldSize() != null) { subTermsAggregationBuilder.size(reqPO.getSubFieldSize()); } termsAggregationBuilder.subAggregation(subTermsAggregationBuilder); // 聚合分组条件 sourceBuilder.aggregation(termsAggregationBuilder); // 查询结果 SearchResponse searchResponse = null; try { log.info("es查询请求对象source：sourceBuilder={}", searchRequest.source()); // 执行搜索 searchResponse = highLevelClient.search(searchRequest, RequestOptions.DEFAULT); log.info("es查询响应结果：searchResponse={}", searchResponse); } catch (IOException e) { log.error("es查询，IO异常！searchRequest={}", searchRequest, e); throw new EsException("es查询，IO异常！"); } if (RestStatus.OK.equals(searchResponse.status())) { // 遍历terms聚合结果 Terms terms = searchResponse.getAggregations().get(termsAggregationBuilder.getName()); List bucketList = terms.getBuckets().stream().map(bucket -> { // 一级聚合分桶的数据 String key = bucket.getKeyAsString(); long docCount = bucket.getDocCount(); // 二级聚合分桶的数据 Terms subTerms = bucket.getAggregations().get(subTermsAggregationBuilder.getName()); List subBucketList = convertTerms(subTerms); return new AggregationBucketPO(key, docCount, subBucketList); }).collect(Collectors.toList()); return bucketList; } else { log.error("es查询返回的状态码异常！searchResponse.status={}, searchRequest={}", searchResponse.status(), searchRequest); throw new EsException("es查询返回的状态码异常！"); } } private List convertTerms(Terms terms) { if (CollectionUtils.isEmpty(terms.getBuckets())) { return Lists.newArrayList(); } return terms.getBuckets().stream().map(bucket -> { String key = bucket.getKeyAsString(); long docCount = bucket.getDocCount(); return new AggregationBucketPO(key, docCount); }).collect(Collectors.toList()); } }

@Data public class EsMultiAggregationReqPO { /** * 索引名 */ String[] index; /** * 查询条件 */ QueryBuilder query; /** * 聚合分桶字段 */ private String field; /** * 二级聚合分桶字段 */ private String subField; /** * 聚合分桶大小，非必传 */ private Integer fieldSize; /** * 二级聚合分桶大小，非必传 */ private Integer subFieldSize; /** * 排序字段，非必传 */ String sortField; /** * 排序方式 SortOrder.ASC、SortOrder.DESC，非必传 */ SortOrder sort; }

@Data @NoArgsConstructor @AllArgsConstructor public class AggregationBucketPO { /** * 聚合Bucket的key名 */ private String key; /** * 聚合Bucket的文档数量 */ private Long docCount; /** * 子桶集合 */ private List subBucketList; public AggregationBucketPO(String key, Long docCount) { this.key = key; this.docCount = docCount; } }

@PostConstruct private void init() { // 查询对象的封装 EsMultiAggregationReqPO reqPO = new EsMultiAggregationReqPO(); reqPO.setIndex(new String[]{"test_log"}); List ids = Lists.newArrayList(); ids.add(140L); ids.add(141L); ids.add(142L); QueryBuilder queryBuilder4 = QueryBuilders.termsQuery("eventId", ids); BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(queryBuilder4); reqPO.setQuery(queryBuilder); reqPO.setField("eventId"); reqPO.setFieldSize(9999); reqPO.setSubField("riskFlag"); // 执行查询 List esQueryRespPO = searchMultiAggregation(reqPO); System.out.println("esQueryRespPO=" + esQueryRespPO); }

/** * @author yangzihe * @date 2022/1/25 */ @Configuration public class EsClientConfig { @Value("${spring.elasticsearch.rest.uris}") private List uris; @Bean public RestHighLevelClient restHighLevelClient() { List httpHostList = uris.stream().map(HttpHost::create).collect(Collectors.toList()); HttpHost[] httpHost = new HttpHost[uris.size()]; httpHostList.toArray(httpHost); RestClientBuilder clientBuilder = RestClient.builder(httpHost); return new RestHighLevelClient(clientBuilder); } }

  cn.hutool hutool-all 5.7.14 

@RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = StartApplication.class) public class EsTest { @Resource private EsSearchRepository esSearchRepository; @Test public void testSearchAggregation() { // 查询对象的封装 EsQueryReqPO queryPO = new EsQueryReqPO(); queryPO.setIndex(new String[]{"yzh2", "yzh3"}); queryPO.setPageNum(1); queryPO.setPageSize(10); // 时间戳范围 QueryBuilder queryBuilder1 = QueryBuilders.rangeQuery("timestamp") .from(System.currentTimeMillis() - 1000) .to(System.currentTimeMillis()); // 登录标识 QueryBuilder queryBuilder2 = QueryBuilders.termQuery("name", "yang"); BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(queryBuilder1).must(queryBuilder2); queryPO.setQuery(queryBuilder); // 根据userName分组。创建terms桶聚合，聚合名字=terms_by_userName, 字段=payload.userName.keyword TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders .terms("terms_by_userName").field("payload.userName.keyword"); termsAggregationBuilder.size(queryPO.getPageSize() * queryPO.getPageNum()); termsAggregationBuilder.subAggregation(new BucketSortPipelineAggregationBuilder("bucket_field", null) .from((queryPO.getPageNum() - 1) * queryPO.getPageSize()).size(queryPO.getPageSize())); queryPO.setTermsAggregation(termsAggregationBuilder); // 根据userName聚合count统计. cardinality名=count_userName, 字段=payload.userName.keyword CardinalityAggregationBuilder cardinalityAggregationBuilder = AggregationBuilders .cardinality("count_userName").field("payload.userName.keyword"); queryPO.setCardinalityAggregation(cardinalityAggregationBuilder); // 执行查询 EsQueryRespPO esQueryRespPO = esSearchRepository.searchAggregation(queryPO); } }