Elasticsearch RestHighLevelClient API 使用总结

Elasticsearch RestHighLevelClient API 使用总结

题记

深秋初冬的一个晚上，突然间收到业务一个需求，要在老系统上使用新系统

Elasticsearch

库的数据。

目前项目情况，新、老系统并行运行，根据产品、渠道路由，但是老系统未使用

Elasticsearch

新系统基础框架

Spring Cloud Alibaba

version

->

2.2.1RELEASE

，老系统基础框架

Spring Framework

version

->

4.3.24RELEASE

。

技术响应

为了满足业务需求，经过技术分析讨论有两种实现方案：

方案一：

在新系统中暴露

HTTP

服务接口，让老系统直接调用新系统，完成数据获取；

方案二：

在老系统以最小侵入单元的形式集成

Elasticsearch

，完成数据获取；

经过系统交互分析，从系统架构设计角度考虑，为减少系统耦合，采用方案二完成数据接入。

Elasticsearch 接入

maven 依赖

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

注释：本文采用

version

->

6.8.6

客户端完成接入，原因：与生产

Elasticsearch

版本保持一致。

实现自动装配

importlombok.extern.slf4j.Slf4j;importorg.apache.http.Header;importorg.apache.http.HttpHost;importorg.apache.http.auth.AuthScope;importorg.apache.http.auth.UsernamePasswordCredentials;importorg.apache.http.client.CredentialsProvider;importorg.apache.http.impl.client.BasicCredentialsProvider;importorg.apache.http.message.BasicHeader;importorg.elasticsearch.client.RestClient;importorg.elasticsearch.client.RestClientBuilder;importorg.elasticsearch.client.RestHighLevelClient;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;/**
 * elasticsearch 配置类
 */@Slf4j@ConfigurationpublicclassElasticsearchConfig{@Value("${elasticsearch.cluster.address}")privateString clusterAddress;@Value("${elasticsearch.username}")privateString username;@Value("${elasticsearch.password}")privateString password;@Value("${elasticsearch.shards}")privateInteger numberOfShards;@Value("${elasticsearch.replicas}")privateInteger numberOfReplicas;@Value("${elasticsearch.connect_timeout}")privateLong connectTimeout;@Value("${elasticsearch.socket_timeout}")privateLong socketTimeout;publicstaticRestHighLevelClient client =null;publicIntegergetNumberOfShards(){return numberOfShards;}publicIntegergetNumberOfReplicas(){return numberOfReplicas;}/**
     * RestHighLevelClient bean创建
     */@BeanpublicRestHighLevelClientrestClient(){finalCredentialsProvider credentialsProvider =newBasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,newUsernamePasswordCredentials(username, password));Header[] defaultHeaders ={newBasicHeader("content-type","application/json")};RestClientBuilder restClientBuilder =RestClient.builder(HttpHost.create(clusterAddress));
        restClientBuilder
                .setHttpClientConfigCallback(httpAsyncClientBuilder -> httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider)).setDefaultHeaders(defaultHeaders).setRequestConfigCallback(requestConfigBuilder ->{// 连接5秒超时，套接字连接60s超时return requestConfigBuilder.setConnectTimeout(connectTimeout.intValue()).setSocketTimeout(socketTimeout.intValue());}).setHttpClientConfigCallback(httpClientBuilder ->{
                    httpClientBuilder.disableAuthCaching();return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);});

        client =newRestHighLevelClient(restClientBuilder);return client;}}

Elasticsearch

持久层接入

importcn.hutool.core.map.MapUtil;importlombok.extern.slf4j.Slf4j;importorg.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;importorg.elasticsearch.action.bulk.BulkRequest;importorg.elasticsearch.action.delete.DeleteRequest;importorg.elasticsearch.action.index.IndexRequest;importorg.elasticsearch.action.search.SearchRequest;importorg.elasticsearch.action.search.SearchResponse;importorg.elasticsearch.action.support.WriteRequest;importorg.elasticsearch.action.support.master.AcknowledgedResponse;importorg.elasticsearch.action.update.UpdateRequest;importorg.elasticsearch.client.RequestOptions;importorg.elasticsearch.client.RestHighLevelClient;importorg.elasticsearch.client.core.CountRequest;importorg.elasticsearch.client.core.CountResponse;importorg.elasticsearch.client.indices.CreateIndexRequest;importorg.elasticsearch.client.indices.GetIndexRequest;importorg.elasticsearch.index.query.BoolQueryBuilder;importorg.elasticsearch.index.query.QueryBuilders;importorg.elasticsearch.index.query.RangeQueryBuilder;importorg.elasticsearch.search.SearchHit;importorg.elasticsearch.search.builder.SearchSourceBuilder;importorg.elasticsearch.search.sort.SortOrder;importorg.elasticsearch.xcontent.XContentType;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importjava.io.IOException;importjava.util.*;/**
 * elasticsearch 持久层
 */@Slf4j@ServicepublicclassElasticsearchRepository{@AutowiredprivateRestHighLevelClient client ;privatefinalRequestOptions options =RequestOptions.DEFAULT;/**
     * 写入数据
     * @param indexName
     * @param dataMap 数据实体
     * @return
     */publicboolean insert (String indexName,Map<String,Object> dataMap){try{BulkRequest request =newBulkRequest();
            request.add(newIndexRequest(indexName,"record").id(dataMap.remove("id").toString()).opType("create").source(dataMap,XContentType.JSON));
            client.bulk(request, options);returnBoolean.TRUE;}catch(Exception e){
            log.error("ElasticsearchRepository#insert, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 批量写入数据
     * @param indexName
     * @param userIndexList
     * @return
     */publicboolean batchInsert (String indexName,List<Map<String,Object>> userIndexList){try{BulkRequest request =newBulkRequest();for(Map<String,Object> dataMap:userIndexList){
                request.add(newIndexRequest(indexName,"record").id(dataMap.remove("id").toString()).opType("create").source(dataMap,XContentType.JSON));}
            client.bulk(request, options);returnBoolean.TRUE;}catch(Exception e){
            log.error("ElasticsearchRepository#batchInsert, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 更新数据
     * 可以直接修改索引结构
     * @param indexName
     * @param dataMap
     * @return
     */publicboolean update (String indexName,Map<String,Object> dataMap){try{UpdateRequest updateRequest =newUpdateRequest(indexName,"record", dataMap.remove("id").toString());
            updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            updateRequest.doc(dataMap);
            client.update(updateRequest, options);returnBoolean.TRUE;}catch(Exception e){
            log.error("ElasticsearchRepository#update, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 根据 id 及索引删除数据
     * @param indexName
     * @param id
     * @return
     */publicboolean delete (String indexName,String id){try{DeleteRequest deleteRequest =newDeleteRequest(indexName,"record", id);
            client.delete(deleteRequest, options);returnBoolean.TRUE;}catch(Exception e){
            log.error("ElasticsearchRepository#delete, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 判断索引是否存在
     * @param indexName
     * @return
     */publicboolean checkIndex (String indexName){try{return client.indices().exists(newGetIndexRequest(indexName), options);}catch(IOException e){
            log.error("ElasticsearchRepository#checkIndex, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 创建索引
     * @param indexName
     * @param columnMap
     * @return
     */publicboolean createIndex (String indexName ,Map<String,Object> columnMap){try{if(!checkIndex(indexName)){CreateIndexRequest request =newCreateIndexRequest(indexName);if(columnMap !=null&& columnMap.size()>0){Map<String,Object> source =newHashMap<>();
                    source.put("properties", columnMap);
                    request.mapping(source);}
                client.indices().create(request, options);returnBoolean.TRUE;}}catch(IOException e){
            log.error("ElasticsearchRepository#createIndex, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 删除索引
     * @param indexName
     * @return
     */publicbooleandeleteIndex(String indexName){try{if(checkIndex(indexName)){DeleteIndexRequest request =newDeleteIndexRequest(indexName);AcknowledgedResponse response = client.indices().delete(request, options);return response.isAcknowledged();}}catch(Exception e){
            log.error("ElasticsearchRepository#deleteIndex, 索引名称:{}, 执行异常:{}", indexName, e);}returnBoolean.FALSE;}/**
     * 查询满足条件的数据条数
     * @param indexName
     * @param matchMap
     * @return
     */publicLong count (String indexName,LinkedHashMap<String,Object> matchMap){// 查询器构造BoolQueryBuilder queryBuilder =QueryBuilders.boolQuery();SearchSourceBuilder sourceBuilder =newSearchSourceBuilder();// 查询时间范围内的数据if(MapUtil.isNotEmpty(matchMap)){if(matchMap.containsKey("startTime")&& matchMap.containsKey("endTime")){RangeQueryBuilder rangequerybuilder =QueryBuilders.rangeQuery("createTime").from(Long.parseLong(String.valueOf(matchMap.get("startTime")))).to(Long.parseLong(String.valueOf(matchMap.get("endTime"))));
                queryBuilder.must(rangequerybuilder);}// 移除时间查询条件
            matchMap.remove("startTime");
            matchMap.remove("endTime");// 时间查询条件外的参数拼接if(MapUtil.isNotEmpty(matchMap)){
                matchMap.forEach((k, v)->{queryBuilder.must(QueryBuilders.termQuery(k, v));});
                sourceBuilder.query(queryBuilder);}}CountRequest countRequest =newCountRequest(indexName);
        countRequest.source(sourceBuilder);try{CountResponse countResponse = client.count(countRequest, options);return countResponse.getCount();}catch(Exception e){
            log.error("ElasticsearchRepository#count, 索引名称:{}, 执行异常:{}", indexName, e);}return0L;}/**
     * 查询满足条件的数据集合
     * 适用于满足条件的数据条数可控的全量查询 PS：单次查询条数不超过 10000条
     * @param indexName
     * @param matchMap
     * @return
     */publicList<Map<String,Object>> list (String indexName,LinkedHashMap<String,Object> matchMap){// 查询条件,指定时间并过滤指定字段值SearchSourceBuilder sourceBuilder =newSearchSourceBuilder();BoolQueryBuilder queryBuilder =QueryBuilders.boolQuery();// 查询参数拼接if(MapUtil.isNotEmpty(matchMap)){
            matchMap.forEach((k, v)->{queryBuilder.must(QueryBuilders.termQuery(k, v));});}
        sourceBuilder.query(queryBuilder);SearchRequest searchRequest =newSearchRequest(indexName);
        searchRequest.source(sourceBuilder);try{SearchResponse searchResp = client.search(searchRequest, options);List<Map<String,Object>> data =newArrayList<>();SearchHit[] searchHitArr = searchResp.getHits().getHits();for(SearchHit searchHit:searchHitArr){Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId());
                data.add(temp);}return data;}catch(Exception e){
            log.error("ElasticsearchRepository#list, 索引名称:{}, 执行异常:{}", indexName, e);}returnnull;}/**
     * 根据查询条件，分页查询
     * 适用于满足条件的数据总量较大的循环查询场景
     * @param indexName
     * @param offset 偏移量
     * @param size 条数
     * @param matchMap
     * @return
     */publicList<Map<String,Object>> page (String indexName,Integer offset,Integer size,LinkedHashMap<String,Object> matchMap){// 添加分页参数SearchSourceBuilder sourceBuilder =newSearchSourceBuilder();
        sourceBuilder.from(offset);
        sourceBuilder.size(size);
        sourceBuilder.sort("createTime",SortOrder.DESC);BoolQueryBuilder queryBuilder =QueryBuilders.boolQuery();// 查询时间范围内的数据if(MapUtil.isNotEmpty(matchMap)){if(matchMap.containsKey("startTime")&& matchMap.containsKey("endTime")){RangeQueryBuilder rangequerybuilder =QueryBuilders.rangeQuery("createTime").from(Long.parseLong(String.valueOf(matchMap.get("startTime")))).to(Long.parseLong(String.valueOf(matchMap.get("endTime"))));
                queryBuilder.must(rangequerybuilder);}// 移除时间查询条件
            matchMap.remove("startTime");
            matchMap.remove("endTime");// 时间查询条件外的参数拼接if(MapUtil.isNotEmpty(matchMap)){
                matchMap.forEach((k, v)->{queryBuilder.must(QueryBuilders.termQuery(k, v));});
                sourceBuilder.query(queryBuilder);}}// 查询请求SearchRequest searchRequest =newSearchRequest(indexName);
        searchRequest.source(sourceBuilder);try{SearchResponse searchResp = client.search(searchRequest, options);List<Map<String,Object>> data =newArrayList<>();SearchHit[] searchHitArr = searchResp.getHits().getHits();for(SearchHit searchHit:searchHitArr){Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId());
                data.add(temp);}return data;}catch(Exception e){
            log.error("ElasticsearchRepository#page, 索引名称:{}, 执行异常:{}", indexName, e);}returnnull;}/**
     * 根据条件查询，按照创建时间进行降序排列
     * 可扩展，根据更新时间、 id、证件号等
     * @param indexName
     * @param matchMap
     * @return
     */publicList<Map<String,Object>> sort (String indexName,LinkedHashMap<String,Object> matchMap){// 先升序时间，在倒序年龄SearchSourceBuilder sourceBuilder =newSearchSourceBuilder();
        sourceBuilder.sort("createTime",SortOrder.ASC);// 查询器构造BoolQueryBuilder queryBuilder =QueryBuilders.boolQuery();// 查询时间范围内的数据if(MapUtil.isNotEmpty(matchMap)){if(matchMap.containsKey("startTime")&& matchMap.containsKey("endTime")){RangeQueryBuilder rangequerybuilder =QueryBuilders.rangeQuery("createTime").from(Long.parseLong(String.valueOf(matchMap.get("startTime")))).to(Long.parseLong(String.valueOf(matchMap.get("endTime"))));
                queryBuilder.must(rangequerybuilder);}// 移除时间查询条件
            matchMap.remove("startTime");
            matchMap.remove("endTime");// 时间查询条件外的参数拼接if(MapUtil.isNotEmpty(matchMap)){
                matchMap.forEach((k, v)->{queryBuilder.must(QueryBuilders.termQuery(k, v));});
                sourceBuilder.query(queryBuilder);}}SearchRequest searchRequest =newSearchRequest(indexName);
        searchRequest.source(sourceBuilder);try{SearchResponse searchResp = client.search(searchRequest, options);List<Map<String,Object>> data =newArrayList<>();SearchHit[] searchHitArr = searchResp.getHits().getHits();for(SearchHit searchHit:searchHitArr){Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId());
                data.add(temp);}return data;}catch(Exception e){
            log.error("ElasticsearchRepository#sort, 索引名称:{}, 执行异常:{}", indexName, e);}returnnull;}}

单元测试

采用

Junit

实现

总结

1.在承接业务需求时，首先要结合功能实现的复杂度，考虑架构的合理性，在相对更合理的系统设计背景下进行功能设计、开发；
2.进行技术开发时，首先要考虑功能对模块的侵入性，在最小侵入性的前提下，采用与基础框架融合的方式，完成开发任务。