jjzjj

Elasticsearch RestHighLevelClient API 使用总结

shang_xs 2024-01-01 原文

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版本保持一致。

实现自动装配


import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
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.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * elasticsearch 配置类
 */
@Slf4j
@Configuration
public class ElasticsearchConfig {

    @Value("${elasticsearch.cluster.address}")
    private String clusterAddress;

    @Value("${elasticsearch.username}")
    private String username;

    @Value("${elasticsearch.password}")
    private String password;

    @Value("${elasticsearch.shards}")
    private Integer numberOfShards;

    @Value("${elasticsearch.replicas}")
    private Integer numberOfReplicas;

    @Value("${elasticsearch.connect_timeout}")
    private Long connectTimeout;

    @Value("${elasticsearch.socket_timeout}")
    private Long socketTimeout;

    public static RestHighLevelClient client = null;

    public Integer getNumberOfShards() {
        return numberOfShards;
    }

    public Integer getNumberOfReplicas() {
        return numberOfReplicas;
    }

    /**
     * RestHighLevelClient bean创建
     */
    @Bean
    public RestHighLevelClient restClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(username, password));
        Header[] defaultHeaders = {new BasicHeader("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 = new RestHighLevelClient(restClientBuilder);
        return client;
    }


}

Elasticsearch 持久层接入


import cn.hutool.core.map.MapUtil;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.*;

/**
 * elasticsearch 持久层
 */
@Slf4j
@Service
public class ElasticsearchRepository {

    @Autowired
    private RestHighLevelClient client ;

    private final RequestOptions options = RequestOptions.DEFAULT;


    /**
     * 写入数据
     * @param indexName
     * @param dataMap 数据实体
     * @return
     */
    public boolean insert (String indexName, Map<String,Object> dataMap){

        try {
            BulkRequest request = new BulkRequest();
            request.add(new IndexRequest(indexName,"record").id(dataMap.remove("id").toString())
                    .opType("create").source(dataMap, XContentType.JSON));
            client.bulk(request, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            log.error("ElasticsearchRepository#insert, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }

    /**
     * 批量写入数据
     * @param indexName
     * @param userIndexList
     * @return
     */
    public boolean batchInsert (String indexName, List<Map<String,Object>> userIndexList){

        try {
            BulkRequest request = new BulkRequest();
            for (Map<String,Object> dataMap:userIndexList){
                request.add(new IndexRequest(indexName,"record").id(dataMap.remove("id").toString())
                        .opType("create").source(dataMap,XContentType.JSON));
            }
            client.bulk(request, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            log.error("ElasticsearchRepository#batchInsert, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }


    /**
     * 更新数据
     * 可以直接修改索引结构
     * @param indexName
     * @param dataMap
     * @return
     */
    public boolean update (String indexName, Map<String,Object> dataMap){

        try {
            UpdateRequest updateRequest = new UpdateRequest(indexName,"record", dataMap.remove("id").toString());
            updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            updateRequest.doc(dataMap) ;
            client.update(updateRequest, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            log.error("ElasticsearchRepository#update, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }


    /**
     * 根据 id 及索引删除数据
     * @param indexName
     * @param id
     * @return
     */
    public boolean delete (String indexName, String id){

        try {
            DeleteRequest deleteRequest = new DeleteRequest(indexName,"record", id);
            client.delete(deleteRequest, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            log.error("ElasticsearchRepository#delete, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }


    /**
     * 判断索引是否存在
     * @param indexName
     * @return
     */
    public boolean checkIndex (String indexName) {

        try {
            return client.indices().exists(new GetIndexRequest(indexName), options);
        } catch (IOException e) {
            log.error("ElasticsearchRepository#checkIndex, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE ;
    }


    /**
     * 创建索引
     * @param indexName
     * @param columnMap
     * @return
     */
    public boolean createIndex (String indexName ,Map<String, Object> columnMap){

        try {
            if(!checkIndex(indexName)){
                CreateIndexRequest request = new CreateIndexRequest(indexName);
                if (columnMap != null && columnMap.size()>0) {
                    Map<String, Object> source = new HashMap<>();
                    source.put("properties", columnMap);
                    request.mapping(source);
                }
                client.indices().create(request, options);
                return Boolean.TRUE ;
            }
        } catch (IOException e) {
            log.error("ElasticsearchRepository#createIndex, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }


    /**
     * 删除索引
     * @param indexName
     * @return
     */
    public boolean deleteIndex(String indexName) {

        try {
            if(checkIndex(indexName)){
                DeleteIndexRequest request = new DeleteIndexRequest(indexName);
                AcknowledgedResponse response = client.indices().delete(request, options);
                return response.isAcknowledged();
            }
        } catch (Exception e) {
            log.error("ElasticsearchRepository#deleteIndex, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return Boolean.FALSE;
    }


    /**
     * 查询满足条件的数据条数
     * @param indexName
     * @param matchMap
     * @return
     */
    public Long count (String indexName, LinkedHashMap<String, Object> matchMap){

        // 查询器构造
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        // 查询时间范围内的数据
        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 = new CountRequest(indexName);
        countRequest.source(sourceBuilder);
        try {
            CountResponse countResponse = client.count(countRequest, options);
            return countResponse.getCount();
        } catch (Exception e) {
            log.error("ElasticsearchRepository#count, 索引名称:{}, 执行异常:{}", indexName, e);
        }
        return 0L;
    }


    /**
     * 查询满足条件的数据集合
     * 适用于满足条件的数据条数可控的全量查询 PS：单次查询条数不超过 10000条
     * @param indexName
     * @param matchMap
     * @return
     */
    public List<Map<String,Object>> list (String indexName, LinkedHashMap<String, Object> matchMap) {

        // 查询条件,指定时间并过滤指定字段值
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();

        // 查询参数拼接
        if (MapUtil.isNotEmpty(matchMap)){
            matchMap.forEach((k, v) -> {queryBuilder.must(QueryBuilders.termQuery(k, v));});
        }
        sourceBuilder.query(queryBuilder);
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            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);
        }
        return null ;
    }


    /**
     * 根据查询条件，分页查询
     * 适用于满足条件的数据总量较大的循环查询场景
     * @param indexName
     * @param offset 偏移量
     * @param size 条数
     * @param matchMap
     * @return
     */
    public List<Map<String,Object>> page (String indexName, Integer offset, Integer size, LinkedHashMap<String, Object> matchMap) {

        // 添加分页参数
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        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 = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);

        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            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);
        }
        return null ;
    }


    /**
     * 根据条件查询，按照创建时间进行降序排列
     * 可扩展，根据更新时间、 id、证件号等
     * @param indexName
     * @param matchMap
     * @return
     */
    public List<Map<String,Object>> sort (String indexName, LinkedHashMap<String, Object> matchMap) {

        // 先升序时间，在倒序年龄
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        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 = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            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);
        }
        return null ;
    }


}

单元测试

采用 Junit实现

总结

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

RestHighLevelClientElasticsearchspanclasstoken大数据搜索引擎

有关Elasticsearch RestHighLevelClient API 使用总结的更多相关文章

  1. ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2

    我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div

  2. ruby - 使用 RubyZip 生成 ZIP 文件时设置压缩级别 - 2

    我有一个Ruby程序，它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重，我想提高压缩级别，因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗？是否有另一个允许指定压缩级别的Ruby库？ 最佳答案 这是我通过查看ruby​​zip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d

  3. ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2

    类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc

  4. ruby-on-rails - 使用 Ruby on Rails 进行自动化测试 - 最佳实践 - 2

    很好奇，就使用ruby​​onrails自动化单元测试而言，你们正在做什么？您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您？git中的预提交Hook？只是手动调用？我完全理解测试，但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的，并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您？ 最佳答案 不确定您到底想听什么，但是有几个级别的自动代码库控制:在处理某项功能时，您可以使用类似autotest的内容获得关于哪些有效，哪些无效的即时反馈。要确保您的提

  5. ruby - 在 Ruby 中使用匿名模块 - 2

    假设我做了一个模块如下:m=Module.newdoclassCendend三个问题:除了对m的引用之外，还有什么方法可以访问C和m中的其他内容？我可以在创建匿名模块后为其命名吗(就像我输入“module...”一样)？如何在使用完匿名模块后将其删除，使其定义的常量不再存在？ 最佳答案 三个答案:是的，使用ObjectSpace.此代码使c引用你的类(class)C不引用m:c=nilObjectSpace.each_object{|obj|c=objif(Class===objandobj.name=~/::C$/)}当然这取决于

  6. ruby - 如何在 Ruby 中顺序创建 PI - 2

    出于纯粹的兴趣，我很好奇如何按顺序创建PI，而不是在过程结果之后生成数字，而是让数字在过程本身生成时显示。如果是这种情况，那么数字可以自行产生，我可以对以前看到的数字实现垃圾收集，从而创建一个无限系列。结果只是在Pi系列之后每秒生成一个数字。这是我通过互联网筛选的结果:这是流行的计算机友好算法，类机器算法:defarccot(x,unity)xpow=unity/xn=1sign=1sum=0loopdoterm=xpow/nbreakifterm==0sum+=sign*(xpow/n)xpow/=x*xn+=2sign=-signendsumenddefcalc_pi(digits

  7. ruby - 使用 ruby​​ 和 savon 的 SOAP 服务 - 2

    我正在尝试使用ruby​​和Savon来使用网络服务。测试服务为http://www.webservicex.net/WS/WSDetails.aspx?WSID=9&CATID=2require'rubygems'require'savon'client=Savon::Client.new"http://www.webservicex.net/stockquote.asmx?WSDL"client.get_quotedo|soap|soap.body={:symbol=>"AAPL"}end返回SOAP异常。检查soap信封，在我看来soap请求没有正确的命名空间。任何人都可以建议我

  8. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2

    关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。

  9. ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2

    我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t

  10. ruby - 使用 ruby​​ 将 HTML 转换为纯文本并维护结构/格式 - 2

    我想将html转换为纯文本。不过，我不想只删除标签，我想智能地保留尽可能多的格式。为插入换行符标签，检测段落并格式化它们等。输入非常简单，通常是格式良好的html(不是整个文档，只是一堆内容，通常没有anchor或图像)。我可以将几个正则表达式放在一起，让我达到80%，但我认为可能有一些现有的解决方案更智能。 最佳答案 首先，不要尝试为此使用正则表达式。很有可能你会想出一个脆弱/脆弱的解决方案，它会随着HTML的变化而崩溃，或者很难管理和维护。您可以使用Nokogiri快速解析HTML并提取文本:require'nokogiri'h

随机推荐