ELK高级搜索七Spring boot 接入Elasticsearch

女爷i 2024-03-17 21:41 92阅读 0赞

##  ##

**目录**

Java api 实现文档管理

一、maven依赖

二、使用步骤

创建索引

ES API的操作步骤

查询文档测试

异步查询文档测试

分页查询文档信息

创建文档测试

异步创建文档

编辑文档

删除文档

创建索引

异步创建索引

索引是否存在

删除索引

--------------------

## Java api 实现文档管理 ##

> es技术比较特殊，不像其他分布式、大数据课程，haddop、spark、hbase。es代码层面很好写，难的是概念的理解。最重要的是他的rest api。跨语言的。在真实生产中，探查数据、分析数据，使用rest更方便。

--------------------

## 一、maven依赖 ##

使用spring boot 开发elasticsearch，需要先引入elasticsearch 的相关依赖。

<dependency>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-high-level-client</artifactId>
                    <version>7.14.2</version>
                    <exclusions>
                        <exclusion>
                            <groupId>org.elasticsearch</groupId>
                            <artifactId>elasticsearch</artifactId>
                        </exclusion>
                    </exclusions>
                </dependency>
                <dependency>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                    <version>7.14.2</version>
                </dependency>

## 二、使用步骤 ##

###  ###

#### 创建索引 ####

使用kibana 创建索引，并手动添加几条数据，方便后面测试。索引的名称：article ，别名：news。

PUT  /article
    {
        "settings":{
            "number_of_shards":3,
            "number_of_replicas":2
        },
        "mappings":{
            "dynamic":false,
            "properties":{
                "title":{
                    "type":"text",
                    "analyzer":"ik_smart",
                    "search_analyzer":"ik_max_word"
                },
                "content":{
                    "type":"text",
                    "analyzer":"ik_smart",
                    "search_analyzer":"ik_max_word"
                },
                "categoryName":{
                    "type":"keyword"
                },
                "view_count":{
                  "type": "integer"
                },
                "publishTime":{
                    "type":"date",
                    "format":"yyyy-MM-dd HH:mm:ss||yyyy-MM-dd"
                }
            }
        },
        "aliases":{
            "news":{
     
            }
        }
    }

#### ES API的操作步骤 ####

获取连接客户端  
构建请求  
执行请求  
获取结果  
获取连接

/**
     * ElasticSearch 配置类定义
     *
     * @author yangyanping
     * @date 2022-10-06
     */
    @Data
    @Configuration
    @ConfigurationProperties(prefix = "es.config")
    public class ElasticSearchConfig {
     
        private String user;
     
        private String passwd;
     
        private String ip;
     
        private String httpUrl;
     
        private int port;
     
        private String bookIndex;
     
        @Bean(destroyMethod = "close", name = "client")
        public RestHighLevelClient restHighLevelClient() {
            // 设置验证信息，填写账号及密码
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(user, passwd));
            // 初始化 RestClient, hostName 和 port 填写集群的内网 VIP 地址与端口
     
            RestClientBuilder builder = StringUtils.isNotBlank(ip) ?
                    RestClient.builder(new HttpHost(ip, port, "http")) :
                    RestClient.builder(new HttpHost(httpUrl, port, "https")) ;
            // 设置认证信息
            builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            });
     
            return new RestHighLevelClient(builder);
        }
    }

#### 查询文档测试 ####

/**
         * 根据文档ID查询
         */
        @Test
        public void tetGetDoc() throws Exception {
            // 构建请求
            GetRequest getRequest = new GetRequest("news", "1");
            // 执行请求
            GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
            if (getResponse != null && getResponse.isExists()) {
                String sourceAsString = getResponse.getSourceAsString();
                // 获取结果
                System.out.println(sourceAsString);
            } else {
                System.out.println("查询异常");
            }
        }

输出结果

{
        "title":"芯片人才争夺",
        "content":"芯片人才争夺“生猛”,需要大量人才",
        "categoryName":"科技",
        "view_count":60,
        "publishTime":"2022-04-19 12:00:00"
    }

#### 异步查询文档测试 ####

/**
         * 异步根据文档ID查询
         */
        @Test
        public void tetAsyncGetDoc() throws Exception {
            // 构建请求
            GetRequest getRequest = new GetRequest("news", "1");
     
            // 执行请求
            client.getAsync(getRequest, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse getResponse) {
                    if(getResponse != null && getResponse.isExists()) {
                        // 获取结果
                        System.out.println(JSON.toJSONString(getResponse));
                    } else {
                        System.out.println("查询异常");
                    }
                }
     
                @Override
                public void onFailure(Exception e) {
                    System.out.println(e);
                }
            });
     
            try {
                Thread.sleep(4000);
            } catch (Exception ex) {
     
            }
        }

输出结果

{
        "exists":true,
        "fields":{
     
        },
        "fragment":false,
        "id":"1",
        "index":"article",
        "primaryTerm":1,
        "seqNo":0,
        "source":{
            "publishTime":"2022-04-19 12:00:00",
            "title":"芯片人才争夺",
            "categoryName":"科技",
            "content":"芯片人才争夺“生猛”,需要大量人才",
            "view_count":60
        },
        "sourceAsBytes":"eyJ0aXRsZSI6IuiKr+eJh+S6uuaJjeS6ieWkuiIsImNvbnRlbnQiOiLoiq/niYfkurrmiY3kuonlpLrigJznlJ/njJvigJ0s6ZyA6KaB5aSn6YeP5Lq65omNIiwiY2F0ZWdvcnlOYW1lIjoi56eR5oqAIiwidmlld19jb3VudCI6NjAsInB1Ymxpc2hUaW1lIjoiMjAyMi0wNC0xOSAxMjowMDowMCJ9",
        "sourceAsBytesRef":{
            "fragment":true
        },
        "sourceAsMap":{
            "$ref":"$.source"
        },
        "sourceAsString":"{\"title\":\"芯片人才争夺\",\"content\":\"芯片人才争夺“生猛”,需要大量人才\",\"categoryName\":\"科技\",\"view_count\":60,\"publishTime\":\"2022-04-19 12:00:00\"}",
        "sourceEmpty":false,
        "sourceInternal":{
            "$ref":"$.sourceAsBytesRef"
        },
        "type":"_doc",
        "version":1
    }

#### 分页查询文档信息 ####

/**
         * 分页查询总结
         * from+ size 分页，如果数据量不大或者from、size不大的情况下，效率还是蛮高的。但是在深度分页的情况下，这种使用方式效率是非常低的，
         * 并发一旦过大，还有可能直接拖垮整个ElasticSearch的集群。
         * scroll 分页通常不会用在客户端，因为每一个 scroll_id 都会占用大量的资源，一般是后台用于全量读取数据使用
         * search_after通过维护一个实时游标来避免scroll的缺点，它可以用于实时请求和高并发场景，一般用于客户端的分页查询
         * 大体而言就是在这三种分页方式中，from + size不适合数据量很大的场景，scroll不适合实时场景，
         * 而search after在es5.x版本之后应运而生，较好的解决了这个问题。
         */
        @Override
        public PageResult<BookEs> queryPage(BookEsQuery query) {
            try {
                BoolQueryBuilder qb = QueryBuilders.boolQuery();
                //用于执行全文查询的标准查询，包括模糊匹配和短语或者接近查询
                if (StringUtils.isNotBlank(query.getKeyword())) {
                    String keyword = query.getKeyword();
                    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                    /**
                     * 基于全文本的查找：Match Query、Match Phrase Query、Query String Query
                     *
                     * 特点：
                     *
                     * 索引和搜索时都会进行分词，查询字符串先传递到一个合适的分词器，然后生成一个供查询的词项列表
                     * 查询的时候。会先对输入的文本进行分词，然后每个此项逐个进行查询，最终将结果进行合并。并为每个文档生成一个算分。
                     */
                    boolQueryBuilder.should(QueryBuilders.termsQuery("bookName.keyword", new String[]{keyword}));
                    boolQueryBuilder.should(QueryBuilders.termsQuery("author.keyword", new String[]{keyword}));
    
                    //如果不是精确搜索，则模糊匹配
                    if (!Objects.equals(query.getSearchType(), Integer.valueOf(2))) {
                        boolQueryBuilder.should(QueryBuilders.matchQuery("bookName", keyword));
                        boolQueryBuilder.should(QueryBuilders.matchQuery("author", keyword));
                    }
    
                    qb.must(boolQueryBuilder);
                }
    
                if (StringUtils.isNotBlank(query.getBookName())) {
                    String bookName = query.getBookName();
                    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                    boolQueryBuilder.should(QueryBuilders.termsQuery("bookName.keyword", new String[]{bookName}));
    
                    //如果不是精确搜索，则模糊匹配
                    if (!Objects.equals(query.getSearchType(), Integer.valueOf(2))) {
                        boolQueryBuilder.should(QueryBuilders.matchQuery("bookName", bookName));
                    }
    
                    qb.must(boolQueryBuilder);
                }
    
                if (ArrayUtils.isNotEmpty(query.getBookTags())) {
                    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                    boolQueryBuilder.should(QueryBuilders.termsQuery("bookTags.keyword", query.getBookTags()));
                    qb.must(boolQueryBuilder);
                }
    
                if (StringUtils.isNotBlank(query.getAuthor())) {
                    String author = query.getAuthor();
                    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                    boolQueryBuilder.should(QueryBuilders.termsQuery("author.keyword", new String[]{author}));
    
                    //如果不是精确搜索，则模糊匹配
                    if (!Objects.equals(query.getSearchType(), Integer.valueOf(2))) {
                        boolQueryBuilder.should(QueryBuilders.matchQuery("author", author));
                    }
    
                    qb.must(boolQueryBuilder);
                }
    
                if (ArrayUtils.isNotEmpty(query.getBookSources())) {
                    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
                    List<String> cpNames =  Arrays.stream(query.getBookSources()).map(BookSourceEnum::getCpName).collect(Collectors.toList());
                    String[] bookSources = new String[cpNames.size()];
                    boolQueryBuilder.should(QueryBuilders.termsQuery("bookSource.keyword", cpNames.toArray(bookSources)));
                    qb.must(boolQueryBuilder);
                }
    
                if (Objects.equals(ChannelTypeEnum.MALE_CHANNEL.getCode(), query.getChannelType()) || Objects.equals(ChannelTypeEnum.FEMALE_CHANNEL.getCode(), query.getChannelType())) {
                    qb.must(QueryBuilders.termsQuery("channelType", new Integer[]{query.getChannelType()}));
                }
    
                if (query.getTwoCategoryId() != null) {
                    qb.must(QueryBuilders.termsQuery("twoCategoryId", new Integer[]{query.getTwoCategoryId()}));
                }
    
                if (BookSerialStatusEnum.SERIALIZING.equals(query.getBookSerialStatusEnum()) || BookSerialStatusEnum.FINISH.equals(query.getBookSerialStatusEnum())) {
                    qb.must(QueryBuilders.termsQuery("serialStatus", new Integer[]{query.getBookSerialStatusEnum().getCode()}));
                }
    
                if (query.getBookWordTypeEnum() != null) {
                    qb.must(QueryBuilders.termsQuery("bookWordType", new Integer[]{query.getBookWordTypeEnum().getCode()}));
                }
    
                if (query.getUpdateTimeTypeEnum() != null) {
                    UpdateTimeTypeEnum updateTimeTypeEnum = query.getUpdateTimeTypeEnum();
                    String from = DateUtil.format(DateUtil.offsetDay(new Date(), -updateTimeTypeEnum.getCode()), "yyyy-MM-dd HH:mm:ss");
                    log.info("from={}", from);
                    String to = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
                    BoolQueryBuilder filter = QueryBuilders.boolQuery();
                    //gte表示大于等于,
                    filter.must(QueryBuilders.rangeQuery("updateTime").gte(from).lt(to));
                }
    
                /**
                 * 返回的文档必须满足filter子句的条件,但是不会像must一样,参与计算分值
                 */
                if (Objects.nonNull(query.getProductType())) {
                    BoolQueryBuilder filter = QueryBuilders.boolQuery();
                    //termQuery：不会对搜索词进行分词处理，而是作为一个整体与目标字段进行匹配，若完全匹配，则可查询到。
                    filter.must(QueryBuilders.termsQuery("productTypes", new int[]{query.getProductType()}));
    
                    qb.filter(filter);
                }
    
                // 0未删除  1已删除
                qb.filter(QueryBuilders.matchQuery("delFlag", "0"));
                // 上架状态 0-上架；1-下架
                qb.filter(QueryBuilders.matchQuery("shelfStatus", "0"));
                // 审核状态 1 机审通过  2 通过 3 驳回 4待审核
                //qb.mustNot(QueryBuilders.termsQuery("verifyStatus", new Integer[]{3}));
    
                HighlightBuilder highlightBuilder = new HighlightBuilder();
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(qb);
                searchSourceBuilder.highlighter(highlightBuilder);
                searchSourceBuilder.timeout(new TimeValue(5, TimeUnit.SECONDS));
    
                if (CollectionUtil.isNotEmpty(query.getOrderDescs())) {
                    OrderDesc orderDesc = query.getOrderDescs().get(0);
                    SortOrder sortOrder = orderDesc.isAsc() ? SortOrder.ASC : SortOrder.DESC;
                    searchSourceBuilder.sort(orderDesc.getCol(), sortOrder).sort("_score", SortOrder.DESC);
                } else {
                    searchSourceBuilder.sort("_score", SortOrder.DESC)
                            .sort(SortBuilders.fieldSort("updateTime").order(SortOrder.DESC));
                }
    
                //from：未指定，默认值是 0，注意不是1，代表当前页返回数据的起始值。
                Integer from = (query.getPageNum() - 1) * query.getPageSize();
                searchSourceBuilder.from(from);
                //size：未指定，默认值是 10，代表当前页返回数据的条数。
                searchSourceBuilder.size(query.getPageSize());
    
                SearchRequest request = new SearchRequest(bookIndex)
                        .source(searchSourceBuilder);
    
                SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
                if (response == null || response.getHits() == null) {
                    return null;
                }
    
                SearchHits searchHits = response.getHits();
                TotalHits totalHits = searchHits.getTotalHits();
    
                List<BookEs> books = new ArrayList<>();
                for (SearchHit hit : searchHits) {
                    String sourceAsString = hit.getSourceAsString();
                    BookEs bookEs = JSON.parseObject(sourceAsString, BookEs.class);
    
                    books.add(bookEs);
                }
    
                PageResult<BookEs> pageResult = new PageResult<>();
                pageResult.setData(books);
                pageResult.setPageIndex(query.getPageNum());
                pageResult.setPageSize(query.getPageSize());
                pageResult.setTotalCount(Integer.valueOf(totalHits.value + ""));
    
                return pageResult;
            } catch (Exception ex) {
                log.info("BookEsFacade.queryPage#query={}", JSON.toJSONString(query), ex);
                throw new BizException("书籍查询失败");
            }
        }

#### 创建文档测试 ####

es节点所在的磁盘空间不够用的时候，es会将该节点上面的索引标位只读，不能向该索引写入数据，默认当磁盘空间超过85%，我们可以使用下面的命令调整磁盘空间比例为95%

PUT localhost:9200/_cluster/settings
     
    {
      "transient": {
        "cluster.routing.allocation.disk.watermark.low": "90%",
        "cluster.routing.allocation.disk.watermark.high": "95%",
        "cluster.info.update.interval": "1m"
      }
    }

/**
         * 创建文档
         */
        @Test
        public void testCreate() throws Exception {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("title", "今日头条、小红书等多家互联网平台将显示IP属地");
            jsonMap.put("content", "今日头条、小红书等多家互联网平台将显示IP属地");
            jsonMap.put("categoryName", "科技");
            jsonMap.put("view_count", 77);
            jsonMap.put("publishTime", "2022-04-18");
     
            IndexRequest request = new IndexRequest("news")
                    .id(String.valueOf(5))
                    .source(jsonMap)
                    .opType(DocWriteRequest.OpType.CREATE);
     
            IndexResponse response = client.index(request, RequestOptions.DEFAULT);
     
            if (response == null || !Objects.equals(response.getResult(), DocWriteResponse.Result.CREATED)) {
                System.out.println("创建异常");
            } else {
                System.out.println(JSON.toJSONString(response));
            }
        }

输出结果

{
        "fragment":false,
        "id":"5",
        "index":"article",
        "primaryTerm":1,
        "result":"CREATED",
        "seqNo":0,
        "shardId":{
            "fragment":true,
            "id":-1,
            "index":{
                "fragment":false,
                "name":"article",
                "uUID":"_na_"
            },
            "indexName":"article"
        },
        "shardInfo":{
            "failed":0,
            "failures":[
     
            ],
            "fragment":false,
            "successful":1,
            "total":3
        },
        "type":"_doc",
        "version":1
    }

#### 异步创建文档    ####

/**
         * 异步创建文档
         */
        @Test
        public void testAsyncAdd() throws Exception {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("title", "微信朋友圈上线十周年 还记得你第一条发的什么内容吗？");
            jsonMap.put("content", "微信朋友圈上线十周年 还记得你第一条发的什么内容吗？");
            jsonMap.put("categoryName", "科技");
            jsonMap.put("view_count", 88);
            jsonMap.put("publishTime", "2022-04-19");
     
            IndexRequest request = new IndexRequest("news")
                    .id("6")
                    .source(jsonMap);
     
            client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                @Override
                public void onResponse(IndexResponse response) {
                    if (response != null && Objects.equals(response.getResult(), DocWriteResponse.Result.CREATED)) {
                        System.out.println(JSON.toJSONString(response));
                    } else {
                        System.out.println("创建异常");
                    }
                }
     
                @Override
                public void onFailure(Exception e) {
                    System.out.println(e);
                }
            });
     
            try {
                Thread.sleep(4000);
            } catch (Exception ex) {
     
            }
        }

#### 编辑文档 ####

/**
         * 部分更新
         */
        @Test
        public void testUpdate() throws Exception {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("content", "微信朋友圈上线十周年 第一条发的什么内容呢？你知道吗");
     
            UpdateRequest request = new UpdateRequest("news", "5").doc(jsonMap);
     
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            if (response != null && Objects.equals(response.getResult(), DocWriteResponse.Result.UPDATED)) {
                System.out.println(JSON.toJSONString(response));
            } else {
                System.out.println("修改异常");
            }
        }

输出结果：

{
        "fragment":false,
        "id":"5",
        "index":"article",
        "primaryTerm":1,
        "result":"UPDATED",
        "seqNo":1,
        "shardId":{
            "fragment":true,
            "id":-1,
            "index":{
                "fragment":false,
                "name":"article",
                "uUID":"_na_"
            },
            "indexName":"article"
        },
        "shardInfo":{
            "failed":0,
            "failures":[
     
            ],
            "fragment":false,
            "successful":1,
            "total":3
        },
        "type":"_doc",
        "version":2
    }

#### 删除文档 ####

/**
         * 根据文档ID删除
         */
        @Test
        public void testDelete() throws Exception {
            DeleteRequest request = new DeleteRequest("news", "5");
     
            DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
            if (response != null && Objects.equals(DocWriteResponse.Result.DELETED, response.getResult())) {
                System.out.println(JSON.toJSONString(response));
            } else {
                System.out.println("删除异常");
            }
        }

输出结果：

{
        "fragment":false,
        "id":"5",
        "index":"article",
        "primaryTerm":1,
        "result":"DELETED",
        "seqNo":2,
        "shardId":{
            "fragment":true,
            "id":-1,
            "index":{
                "fragment":false,
                "name":"article",
                "uUID":"_na_"
            },
            "indexName":"article"
        },
        "shardInfo":{
            "failed":0,
            "failures":[
     
            ],
            "fragment":false,
            "successful":1,
            "total":3
        },
        "type":"_doc",
        "version":3
    }

#### 创建索引 ####

/**
         * 创建索引
         */
        @Test
        public void testCreateIndex() throws Exception {
            //创建索引对象
            CreateIndexRequest request = new CreateIndexRequest("test_index");
            //设置参数
            request.settings(Settings.builder().put("number_of_shards", 1).put("number_of_replicas", 0));
            Map<String, Object> titleMap = Maps.newHashMap();
            titleMap.put("type", "text");
     
            Map<String, Object> properties = Maps.newHashMap();
            properties.put("title", titleMap);
     
            Map<String, Object> mapping = Maps.newHashMap();
            mapping.put("properties", properties);
            //指定映射
            request.mapping(mapping);
            //指定别名
            request.alias(new Alias("test_index_alias"));
     
            IndicesClient indicesClient = client.indices();
            CreateIndexResponse response = indicesClient.create(request, RequestOptions.DEFAULT);
            System.out.println(JSON.toJSONString(response));
        }

输出：\{"acknowledged":true,"fragment":false,"shardsAcknowledged":true\}

#### 异步创建索引 ####

/**
         * 创建索引
         */
        @Test
        public void testAsyncCreateIndex() throws Exception {
            //创建索引对象
            CreateIndexRequest request = new CreateIndexRequest("test_index_async");
            //设置参数
            request.settings(Settings.builder().put("number_of_shards", 1).put("number_of_replicas", 0));
            Map<String, Object> titleMap = Maps.newHashMap();
            titleMap.put("type", "text");
     
            Map<String, Object> properties = Maps.newHashMap();
            properties.put("title", titleMap);
     
            Map<String, Object> mapping = Maps.newHashMap();
            mapping.put("properties", properties);
            //指定映射
            request.mapping(mapping);
            //指定别名
            request.alias(new Alias("test_index_async_alias"));
     
            IndicesClient indicesClient = client.indices();
            Cancellable cancellable = indicesClient.createAsync(request, RequestOptions.DEFAULT, new ActionListener<CreateIndexResponse>() {
                @Override
                public void onResponse(CreateIndexResponse response) {
                    System.out.println(JSON.toJSONString(response));
                }
     
                @Override
                public void onFailure(Exception e) {
                    System.out.println(e);
                }
            });
     
            try {
                Thread.sleep(4000);
            } catch (Exception ex) {
     
            }
        }

输出：

{
        "acknowledged":true,
        "fragment":false,
        "shardsAcknowledged":true
    }

#### 索引是否存在 ####

/**
         * 索引是否存在
         */
        @Test
        public void testExistsIndex() throws Exception {
            GetIndexRequest request = new GetIndexRequest("test_index");
     
            boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
            System.out.println(exists);
        }

输出：true

#### 删除索引 ####

/**
         * 删除索引
         */
        @Test
        public void testDeleteIndex() throws Exception {
            DeleteIndexRequest request = new DeleteIndexRequest("test_index");
     
            AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
            System.out.println(JSON.toJSONString(response));
        }

输出：

{
        "acknowledged":true,
        "fragment":false
    }