Elasticsearch零基础实战

发布时间：2024年01月04日

分享后可优化点（待完成）

java es8 查询如何打印查询入参 ？（直接执行的json）

es自定义分词器 如何实现？

kibana 监控jvm分子分母是什么 ？

es如何 改索引结构？

修改数据原理

分享前罗列大纲，挑出重点

分享时需要讲 原理+应用场景

根据笔记总结分享文档，代码截图等 （不要直接拿笔记出来讲）

任务目标

  1. 基础语法学习
  2. 从es6.8迁移至7.1
  3. 从es6.8迁移至8.7.0
  4. 相关java api升级（从spring boot 封住的es框架到原生es）
  5. 业务操作：订单历史数据从mysql迁移至es
  6. 断路器配置
  7. 监控器配置
  8. 快照配置

windows本地环境搭建（http）

下载es

es下载地址

es集群搭建

解压一个es8.7的zip，然后复制三份，像这样

然后对config下的 elasticsearch.yml 分别进行设置

#节点 1 的配置信息：
# ---------------------------------- Cluster -----------------------------------
# 集群名称，节点之间要保持一致
cluster.name: my-application
# ------------------------------------ Node ------------------------------------
# 节点名称，集群内要唯一
node.name: node-1
# 节点角色 [主节点、数据节点]：注意这一点同7.x版本的区别配置
node.roles: [master,data]
# ---------------------------------- Network -----------------------------------
# 发布ip
network.host: localhost
# http 端口
http.port: 9200
# tcp 监听端口：注意这一点同7.x版本的区别配置
transport.port: 9301
# --------------------------------- Discovery ----------------------------------
# 初始主节点
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
# ---------------------------------- Various -----------------------------------
# 删除索引是否需要指定索引全名（false就是可以用正则）
#action.destructive_requires_name: false
# 跨域配置
http.cors.enabled: true
http.cors.allow-origin: "*"

# 节点 2 的配置信息：
# ---------------------------------- Cluster -----------------------------------
# 集群名称，节点之间要保持一致
cluster.name: my-application
# ------------------------------------ Node ------------------------------------
# 节点名称，集群内要唯一
node.name: node-2
# 节点角色 [主节点、数据节点]
node.roles: [master,data]
# ---------------------------------- Network -----------------------------------
# 发布ip
network.host: localhost
# http 端口
http.port: 9201
# tcp 监听端口
transport.port: 9302
# --------------------------------- Discovery ----------------------------------
# 自动发现集群节点ip
discovery.seed_hosts: ["localhost:9301"]
# discovery.zen.fd.ping_timeout: 1m
# discovery.zen.fd.ping_retries: 5
# 集群内的可以被选为主节点的节点列表
cluster.initial_master_nodes: ["node-1", "node-2","node-3"]
# ---------------------------------- Various -----------------------------------
# 跨域配置
http.cors.enabled: true
http.cors.allow-origin: "*"

#节点 3 的配置信息：
# ---------------------------------- Cluster -----------------------------------
# 集群名称，节点之间要保持一致
cluster.name: my-application
# ------------------------------------ Node ------------------------------------
# 节点名称，集群内要唯一
node.name: node-3
# 节点角色 [主节点、数据节点]
node.roles: [master,data]
# ---------------------------------- Network -----------------------------------
# 发布ip
network.host: localhost
# http 端口
http.port: 9203
# tcp 监听端口
transport.port: 9303
# --------------------------------- Discovery ----------------------------------
# 自动发现集群节点ip
discovery.seed_hosts: ["localhost:9301", "localhost:9302"]
# discovery.zen.fd.ping_timeout: 1m
# discovery.zen.fd.ping_retries: 5
# 集群内的可以被选为主节点的节点列表
cluster.initial_master_nodes: ["node-1", "node-2","node-3"]
# ---------------------------------- Various -----------------------------------
# 跨域配置
http.cors.enabled: true
http.cors.allow-origin: "*"

然后一次性都启动

cd E:\es\node\node-1\bin
.\elasticsearch.bat  

cd E:\es\node\node-2\bin
.\elasticsearch.bat  

cd E:\es\node\node-3\bin
.\elasticsearch.bat

注意点

直接集群启动，es默认走http，

如果不配置集群，单节点，启动，es会在配置文件中添加https相关配置

问题

提示jvm内存不足

修改下内存

找到elasticsearch的安装目录，然后找到config文件夹，里面都是相关的配置文件。

官方不建议直接修改jvm.options，而是复制jvm.options到jvm.options.d目录下，再修改。

其中，jvm.options可以修改es运行时候的内存分配。打开jvm.options文件，我们可以发现默认设置的内存是4g。

我们改成2g

elasticsearch 内存大小设置

忘记密码

可能第一次启动的时候没有保存密码，在bin目录下执行命令：

# 随机密码
./elasticsearch-reset-password -u elastic

注意：在某个节点执行一次即可，所有节点都会生效

参考文档

Windows环境下es8的集群部署

ElasticSearch集群搭建

es8.7官方文档

kibana启动

设置中文

在config的 kibana.yml 中新增

i18n.locale: "zh-CN"

启动

由于es默认开启了security，但是我们又没配置证书，所以kibana直接启动会出错，

需要在三个es节点的配置文件添加如下配置

xpack.security.enabled: false

然后启动

cd E:\es\kibana-8.7.0-windows-x86_64\kibana-8.7.0\bin
.\kibana.bat

大功告成！

windows本地环境搭建（https）

es集群搭建

在http步骤上引入https，之前步骤请查看http部分

证书生成

在bin目录下执行命令，生成证书
./elasticsearch-certutil ca
./elasticsearch-certutil cert --ca elastic-stack-ca.p12

然后将这两个证书文件拷贝到三个node节点的config的certs文件夹下

在配置文件中新增配置（三个节点都需要）

# 开启https协议
xpack.security.enabled: true
# 这行似乎不需要，测试环境copy过来的，先加上
xpack.license.self_generated.type: basic
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate 
xpack.security.transport.ssl.keystore.path: certs/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: certs/elastic-certificates.p12

然后重启es

kibana配置

修改 kibana_system 的密码

在node1节点的bin目录下执行
./elasticsearch-reset-password -u kibana_system

修改 kibana_admin 的密码

./elasticsearch-reset-password -u kibana_admin
密码：

config目录下的 kibana.yml 添加如下配置

server.port: 5601
server.host: "localhost"
server.name: "node-1"
elasticsearch.hosts: ["http://localhost:9200","http://localhost:9201","http://localhost:9203"]
elasticsearch.ssl.verificationMode: none
elasticsearch.requestTimeout: 90000

elasticsearch.username: "kibana_system"
# elasticsearch.password: "Mvwm@n12nal"
elasticsearch.password: "Ql2e3HvwXUL9QfBlb+06"

然后启动，在bin目录执行

./kibana.bat

在浏览器输入网址 http://localhost:5601/

使用 elastic账户登录

参考文档

ES、Kibana 8.0安装

elasticsearch8.2.0 初始化忘记密码重置

kibana设置中文

windows环境下es8的集群部署

linux环境搭建

略

基础语法

注意：es8的时候，已经默认不支持* 或者 _all了

关于删除命令的说明（官网）

查询

# 查询所有节点
GET /_cat/nodes?v

# 查询所有索引
GET /_cat/indices?v
GET /_cat/indices?v&h=health,status,index

# 获取所有的索引mapping信息
GET _all/_mapping

#获取当前索引信息
GET /community_encyclopedia
GET /community_encyclopedia/_settings
GET /community_encyclopedia/_mapping

# 查询当前索引总数
GET /community_encyclopedia/_count
# 查询当前索引数据
GET /community_encyclopedia/_search
# 查询当前索引单条数据
GET /community_encyclopedia/_doc/100
#只获取字段name,age
GET /bamboo/_doc/1?_source=name,age

#查询参数
GET /community_encyclopedia/_search 
{
  "query":{
    "bool":{
      "must":[
        {
        "match_phrase": {
          "type":1
          }
        },{
          "match_phrase": {
            "status":1
          }
        }
      ]
    }
  },
  "sort":[
      {
        "normalIndex":"asc"
      },
      {
        "createTime.keyword":"desc"
      }
    ]
}
# 范围查询
#查询参数
GET /t_car_order/_search 
{
  "query":{
    "range": {
      "create_time": {
        "gte": "2017-07-24T21:26:21.000Z",
        "lte": "2017-07-25T00:26:21.000Z"
      }
    }
  }
}

# 聚集查询 id为1,2的数据
GET /bamboo/_doc/_mget
{
"docs":[
   {
       "_id": 2
   },
   {
       "_id": 1
   }
 ]
}

GET /bamboo/_doc/_search 
{"query":{"bool":{"must":[{"match_all":{}}],"must_not":[],"should":[]}},"from":0,"size":10,"sort":[],"aggs":{}}

# 模糊查询
GET /community_encyclopedia/_search 
{
  "query":{
    "bool": {
        "must": [
          {"wildcard": {"managerQuestion.keyword": "*汽车保险包括*"}}
        ]
      }
  }
}

# 多条件查询
GET /t_car_policy/_search 
{
  "query":{
    "bool":{
      "must":[
        {
        "match": {
          "orderNo":"DCARRxFyaf000060147"
          }
        },{
          "match": {
            "appPolicyNo":"0106003202023026408"
          }
        }
      ],
      "filter": [
        {
          "range": {
            "createTime": {
              "lt": "2023-10-23 00:00:00"
            }
          }
        }
      ]
    }
  }
}
上面的语法中，filter不参与评分，也可以直接放在must中：
GET /t_car_policy/_search 
{
  "query":{
    "bool":{
      "must":[
        {
        "match": {
          "orderNo":"DCARRxFyaf000060147"
          }
        },
        {
          "match": {
            "appPolicyNo":"0106003202023026408"
          }
        },
        {
          "range": {
            "createTime": {
              "lt": "2023-10-23 00:00:00"
            }
          }
        }
      ]
    }
  }
}
以上是匹配查询，更建议使用 .keyword 精确查询
#查询参数
GET /t_car_policy/_search 
{
  "query":{
    "bool":{
      "must":[
        {
        "term": {
          "orderNo.keyword":"DCARRxFyaf000060147"
          }
        },
        {
          "term": {
            "appPolicyNo.keyword":"0106003202023026408"
          }
        },
        {
          "range": {
            "createTime": {
              "lt": "2023-10-23 00:00:00"
            }
          }
        }
      ]
    }
  }
}

删除

# 删除所有索引
# DELETE /_all
# 删除部分索引
# DELETE /test-*
# 删除单个索引
# DELETE /bamboo
# 删除某条数据
# DELETE /bamboo/_doc/1

创建

# 创建空索引
PUT /bamboo
# 创建索引和对应的mapping和setting
PUT /bamboo 
{
  "mappings": { 
      "properties": { 
        "title":    { "type": "text"  }, 
        "name":     { "type": "text"  }, 
        "age":      { "type": "integer" },  
        "created":  {
          "type":   "date", 
          "format": "strict_date_optional_time||epoch_millis"
        }
      }
    },
  "settings":{
    "index":{
    "number_of_shards": 5,
    "number_of_replicas": 1
    }
  }
}

#添加一条数据
PUT /bamboo/_doc/1
{
  "name":"zs",
  "title":"张三",
  "age":18,
  "created":"2018-12-25"
}

# 修改一条数据的某个属性值
PUT /bamboo/_doc/1 
{
  "name":"lxs",
  "title":"李小四"
}

# 批量插入多个document,_id不指定则系统生成字符串
POST /bamboo/_doc/_bulk 
{"index":{"_id":2}}
{"name":"ww","title":"王五","age":18,"created":"2018-12-27"}
{"index":{}}
{"name":"zl","title":"赵六","age":25,"created":"2018-12-27"}

# 批量操作（包含修改和删除）
POST /bamboo/_doc/_bulk
{"update":{"_id":"1"}} 
{"doc":{"title":"王小五"}}
{"delete":{"_id":"2"}}

参考链接：语法

各关键字用法及详细示例--java api

es8语法

must, filter, should, must_not, constant_score的区别

and or

should：其查询子句应该被满足，也就是不一定都满足，逻辑相当于 or。

must：其查询子句必须全部被满足，逻辑相当于 and ，并且会计算分数。

filter：与 must 作用一样，但是不会计算分数。在 filter context 下的查询子句不会计算分数且会被缓存。

must_not：其查询子句必须都不被满足。当子句是在 filter context 下时，不会计算分数且会被缓存。

参考文档：

Elasticsearch 中 must, filter, should, must_not, constant_score 的区别

原理解析

查询es的某个索引，原理是怎样的？

引用一段话：

被混淆的概念是，一个 Lucene 索引我们在 Elasticsearch 称作分片 。 而在Elasticsearch中索引是分片的集合。 当 Elasticsearch 在索引中搜索的时候， 他发送查询到每一个属于索引的分片(Lucene 索引)，然后合并每个分片的结果到一个全局的结果集。

总结来说，查询某个索引时，会根据查询算法定位到某些分片，进而缩小查询范围，提高查询效率

7.*中为什么现在要移除type？

  1. 和数据库的库-表类比是错误的，因为es中同一个index中不同type是存储在同一个索引中的（lucene的索引文件），因此不同type中相同名字的字段的定义（mapping）必须一致。
  2. 当您想要索引一个deleted字段在不同的type中数据类型不一样。一个类型中为日期字段，另外一个类型中为布尔字段时，这可能会导致ES的存储失败,因为这影响了ES的初衷设计。
  3. 另外，在一个index中建立很多实体，type，没有相同的字段，会导致数据稀疏，最终结果是干扰了Lucene有效压缩文档的能力，说白了就是影响ES的存储、检索效率。

es6.x和7.x对比

如何分片， 几个节点？

节点数<=主分片数*(副本数)

3个节点，3个主分片，1个副本数

可以简单设置成节点个数的倍数，以便在节点之间均匀分布数据

后续的高级操作可以考虑根据索引内容进行分片，比如根据时间进行分片等。

萌新发问

logging:

config: classpath:logback-spring.xml

elasticsearch:

# 多个IP逗号隔开

hosts: 10.7.176.72:9200,10.7.176.73:9200,10.7.176.74:9200

username: elastic

password: Mvwm@n12nal
private void createIndexDetail(String indexName) throws IOException {
        // 3.创建索引

        CreateIndexResponse response = elasticsearchClient
                .indices()
                .create(c -> c
                        .index(indexName)
                        .settings(sBuilder -> sBuilder
                                .index(iBuilder -> iBuilder
                                        // 5?个分片
                                        .numberOfShards("5")
                                        // 一个副本
                                        .numberOfReplicas("1")))
                        .mappings(mBuilder -> mBuilder
                                .properties(OrderDetail.ORDER_NO, pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(30)))
                                .properties(OrderDetail.MESSAGE, pBuilder -> pBuilder.text(textPropertyBuilder -> textPropertyBuilder.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                        )
                );
        log.info("createIndexDetail方法，acknowledged={}", response.acknowledged());
    }

这是我在springboot中对es8集群的配置，我想知道当我创建一个索引时，是三个hosts节点都会有我的索引吗？

5个分片，每个分片一个副本分片，所以也有5个副本分片，同一个主分片和副本分片不会在一台host节点上，然后集群会将这些分片打散分布在集群中。

加入我往这个索引中插入100条数据，数据大概是如何分布的呢

假如我往这个索引中插入100条数据，数据大概是如何分布的呢？

大概意思就是说，100条数据在5个主分片上，相当于存放在mysql表中，只是这个表的数据不在一个磁盘上

然后100备份数据放在副本分片上。 相当于主从mysql库

java代码升级

es6.8升级到es7.1.1

采用方式：spring Data Elasticsearch 方式

因为当前springboot版本支持7.1.1，所以采用此方式

版本对应

由于集团使用的是es7.1.1版本，所以我们对应使用spring Data Elasticsearch 4.0.x

相关api

QueryBuilders.matchQuery()：全文查询的参数，会进行分词操作

QueryBuilders.termQuery()：精确查询的参数

参考文档

springboot3.4迁移到4.0官方指南

es6.8升级到es8.7.0

采用方式：es8.7原生spi

由于升级spring boot成本过大，为了和其他微服务的spring boot版本保持一致，故采用此方式

参考文档：

Elasticsearch Java API Client官方文档

Springboot整合ES8(Java API Client)

springboo整合elasticSearch8 java client api

ElasticSearch中文博客

ES8(Java API Client)查询详解

接口示例：

package com.djbx.dh.search.controller.dhwiki;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldSort;
import co.elastic.clients.elasticsearch._types.SortOptions;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.QueryBuilders;
import co.elastic.clients.elasticsearch._types.query_dsl.TermQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.WildcardQuery;
import co.elastic.clients.elasticsearch.core.GetResponse;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.djbx.dh.common.exception.BusinessException;
import com.djbx.dh.common.model.po.ResultCode;
import com.djbx.dh.common.model.vo.ResultEntity;
import com.djbx.dh.search.common.constants.CommonConstants;
import com.djbx.dh.search.common.enums.ValidOrInvalidEnum;
import com.djbx.dh.search.common.enums.YesOrNoEnum;
import com.djbx.dh.search.common.util.UUIDUtil;
import com.djbx.dh.search.entity.dto.IdDTO;
import com.djbx.dh.search.entity.dto.dhwiki.CommunityEncyclopediaQuestionAnswerDTO;
import com.djbx.dh.search.entity.dto.dhwiki.ProductLibraryDTO;
import com.djbx.dh.search.entity.dto.dhwiki.QuestionAnswerUpdateDTO;
import com.djbx.dh.search.entity.po.dhwiki.CommunityEncyclopediaQuestionAnswer;
import com.djbx.dh.search.entity.po.dhwiki.CommunityEncyclopediaQuestionCategory;
import com.djbx.dh.search.entity.vo.dhwiki.CommunityEncyclopediaQuestionAnswerVO;
import com.djbx.dh.search.service.dhwiki.CommunityEncyclopediaQuestionCategoryService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author DJ033979
 */
@Slf4j
@RestController
@RequestMapping("/encyclopedia")
@Api(tags = "搭伙百科")
public class CommunityEncyclopediaController {
    @Resource
    private CommunityEncyclopediaQuestionCategoryService communityEncyclopediaQuestionCategoryService;

    @Resource
    private ElasticsearchClient elasticsearchClient;


    /**
     * 保存数据
     *
     * @param questionAnswer 问答数据
     * @return ResultEntity
     */
    @PostMapping("/saveOrUpdate")
    @ApiOperation(value = "问答数据保存OR更新")
    public ResultEntity<?> saveOrUpdate(@RequestBody CommunityEncyclopediaQuestionAnswer questionAnswer) throws BusinessException, IOException {

        //前置校验，判断索引是否存在
        if (!elasticsearchClient.indices().exists(e -> e.index(CommonConstants.WIKI_INDEX)).value()) {
            // 创建索引
            createIndexDetail();
        }
        // 置顶或推荐判断排序值是否可用
        if (YesOrNoEnum.isYes(questionAnswer.getIsCategoryTop()) && findValidOneByTypeAndStatusAndCategoryAndTopIndex(
                ValidOrInvalidEnum.VALID.getValue(), YesOrNoEnum.YES.getValue(), questionAnswer.getCategory(), questionAnswer.getTopIndex())) {
            log.error("有效的置顶顺序被占用，不能使用");
            return ResultEntity.error("有效的置顶顺序被占用，不能使用");
        }
        if (YesOrNoEnum.isYes(questionAnswer.getIsHotRecommendTop()) && findValidOneByTypeAndStatusAndCategoryAndTopIndex(
                ValidOrInvalidEnum.VALID.getValue(), YesOrNoEnum.YES.getValue(), questionAnswer.getCategory(), questionAnswer.getTopIndex())) {
            log.error("有效的推荐顺序被占用，不能使用");
            return ResultEntity.error("有效的推荐顺序被占用，不能使用");
        }
        // 此字段用于管理后台like查询
        questionAnswer.setManagerQuestion(questionAnswer.getQuestion());
        // 保存
        if (StringUtils.isBlank(questionAnswer.getId())) {
            //设置创建时间
            questionAnswer.setId(UUIDUtil.uuid());
            questionAnswer.setType(YesOrNoEnum.YES.getValue());
            questionAnswer.setCreateTime(new Date());
        } else {// 更新
            // 根据id获取es的数据
            CommunityEncyclopediaQuestionAnswer encyclopediaQuestionAnswer = getCommunityEncyclopediaQuestionAnswerById(questionAnswer.getId());
            // 设置更新数据
            questionAnswer.setCreateTime(encyclopediaQuestionAnswer.getCreateTime());
            questionAnswer.setType(encyclopediaQuestionAnswer.getType());
            questionAnswer.setUpdateTime(new Date());
        }
        try {
            elasticsearchClient.index(s ->
                    s.index(CommonConstants.WIKI_INDEX)
                            .id(questionAnswer.getId())
                            .document(questionAnswer)
            );
        } catch (Exception e) {
            log.error("问答数据保存OR更新至ES报错:{}", e.getMessage());
            return ResultEntity.error("问答数据保存OR更新至ES报错");
        }
        return ResultEntity.success("问答数据保存OR更新至ES成功");
    }



    /**
     * 根据问答ID查询问答数据记录
     *
     * @param dto 问答ID
     * @return CommunityEncyclopediaQuestionAnswer
     */
    @PostMapping("/getById")
    @ApiOperation(value = "根据ID查询")
    public ResultEntity<?> getById(@RequestBody IdDTO dto) throws BusinessException {
        if (null == dto || StringUtils.isBlank(dto.getId())) {
            log.error("百科问答ID不能为NULL");
            return ResultEntity.error("百科问答ID不能为空和NULL");
        }
        return ResultEntity.success(getCommunityEncyclopediaQuestionAnswerById(dto.getId()));
    }

    /**
     * 百科问答上下线操作
     */
    @PostMapping("/updateStatus")
    @ApiOperation(value = "百科问答上下线操作")
    public ResultEntity<?> updateStatus(@RequestBody QuestionAnswerUpdateDTO dto) throws BusinessException, IOException {
        if (null == dto || StringUtils.isBlank(dto.getId()) || null == dto.getStatus()) {
            log.error("百科问答ID或发布状态不能为空和NULL");
            return ResultEntity.error("百科问答ID或发布状态不能为空和NULL");
        }
        String id = dto.getId();
        Integer status = dto.getStatus();
        CommunityEncyclopediaQuestionAnswer encyclopediaQuestionAnswer = getCommunityEncyclopediaQuestionAnswerById(id);

        if (status.equals(encyclopediaQuestionAnswer.getStatus())) {
            log.error("百科问答当前状态已为上下线的操作状态");
            return ResultEntity.error("百科问答当前状态已为上下线的操作状态");
        }

        // 上线时，判断置顶排序和推荐顺序是否占用
        if (YesOrNoEnum.isYes(status) && null != encyclopediaQuestionAnswer.getTopIndex() && findValidOneByTypeAndStatusAndCategoryAndTopIndex(
                ValidOrInvalidEnum.VALID.getValue(), status, encyclopediaQuestionAnswer.getCategory(), encyclopediaQuestionAnswer.getTopIndex())) {
            log.error("有效的置顶顺序被占用，不能使用");
            return ResultEntity.error("有效的置顶顺序被占用，不能使用");
        }
        if (YesOrNoEnum.isYes(status) && null != encyclopediaQuestionAnswer.getRecommendIndex() && findValidOneByTypeAndStatusAndRecommendIndex(
                ValidOrInvalidEnum.VALID.getValue(), status, encyclopediaQuestionAnswer.getRecommendIndex())) {
            log.error("有效的推荐顺序被占用，不能使用");
            return ResultEntity.error("有效的推荐顺序被占用，不能使用");
        }

        // 更新上下线状态
        encyclopediaQuestionAnswer.setStatus(status);
        encyclopediaQuestionAnswer.setUpdateTime(new Date());
        try {
            elasticsearchClient.update(e -> e
                    .index(CommonConstants.WIKI_INDEX)
                    .id(id)
                    .doc(encyclopediaQuestionAnswer), CommunityEncyclopediaQuestionAnswer.class
            );
        } catch (Exception e) {
            log.error("百科问答上下线操作失败:{}", e.getMessage());
            return ResultEntity.error("百科问答上下线操作失败");
        }
        return ResultEntity.success("百科问答上下线操作成功");
    }



    /**
     * 问题分类上下线而更新该分类下所有问答的上下线状态
     */
    @PostMapping("/updateStatusByCategory")
    @ApiOperation(value = "问题分类上下线同步百科问答上下线状态")
    public ResultEntity<?> updateStatusByCategory(@RequestBody Map<String, Object> map) throws BusinessException, IOException {
        Integer category = (Integer) map.get("category");
        Integer status = (Integer) map.get("status");
        List<CommunityEncyclopediaQuestionAnswer> encyclopediaQuestionAnswers = findValidListByTypeAndCategory(
                ValidOrInvalidEnum.VALID.getValue(), category);
        try {
            if (!CollectionUtils.isEmpty(encyclopediaQuestionAnswers)) {
                for (CommunityEncyclopediaQuestionAnswer encyclopediaQuestionAnswer : encyclopediaQuestionAnswers) {
                    // 上线时，判断推荐顺序是否占用,如占用设置为不是热门推荐
                    if (YesOrNoEnum.isYes(status) && null != encyclopediaQuestionAnswer.getRecommendIndex() && findValidOneByTypeAndStatusAndRecommendIndex(
                            ValidOrInvalidEnum.VALID.getValue(), status, encyclopediaQuestionAnswer.getRecommendIndex())) {
                        encyclopediaQuestionAnswer.setIsHotRecommend(YesOrNoEnum.NO.getValue());
                        encyclopediaQuestionAnswer.setIsHotRecommendTop(YesOrNoEnum.NO.getValue());
                        encyclopediaQuestionAnswer.setRecommendIndex(null);
                    }

                    // 更新上下线状态
                    encyclopediaQuestionAnswer.setStatus(status);
                    encyclopediaQuestionAnswer.setUpdateTime(new Date());
                    elasticsearchClient.update(e -> e
                            .index(CommonConstants.WIKI_INDEX)
                            .id(encyclopediaQuestionAnswer.getId())
                            .doc(encyclopediaQuestionAnswer), CommunityEncyclopediaQuestionAnswer.class
                    );
                }
            }
        } catch (Exception e) {
            log.error("该问题分类下的百科问答上下线操作失败:{}", e.getMessage());
            return ResultEntity.error("该问题分类下的百科问答上下线操作失败");
        }
        return ResultEntity.success("该问题分类下的百科问答上下线操作成功");
    }



    /**
     * 根据问答ID删除问答数据记录
     */
    @PostMapping("/delete")
    @ApiOperation(value = "根据ID删除百科问答")
    public ResultEntity<?> delete(@RequestBody IdDTO dto) throws BusinessException {
        if (null == dto || StringUtils.isBlank(dto.getId())) {
            log.error("百科问答ID不能为空和NULL");
            return ResultEntity.error("百科问答ID不能为空和NULL");
        }
        CommunityEncyclopediaQuestionAnswer encyclopediaQuestionAnswer = getCommunityEncyclopediaQuestionAnswerById(dto.getId());
        // 更新有效无效，逻辑删除
        encyclopediaQuestionAnswer.setType(ValidOrInvalidEnum.INVALID.getValue());
        // 去除占用的排序值,设置为默认255
        encyclopediaQuestionAnswer.setTopIndex(null);
        encyclopediaQuestionAnswer.setRecommendIndex(null);
        encyclopediaQuestionAnswer.setUpdateTime(new Date());
        try {
            elasticsearchClient.update(e -> e
                    .index(CommonConstants.WIKI_INDEX)
                    .id(dto.getId())
                    .doc(encyclopediaQuestionAnswer), CommunityEncyclopediaQuestionAnswer.class
            );
        } catch (Exception e) {
            log.error("删除百科问答操作失败:{}", e.getMessage());
            return ResultEntity.error("删除百科问答操作失败");
        }
        return ResultEntity.success("删除百科问答操作成功");
    }

    /**
     * 取消推荐
     */
    @PostMapping("/cancelRecommend")
    @ApiOperation(value = "取消推荐")
    public ResultEntity<?> cancelRecommend(@RequestBody IdDTO dto) throws BusinessException {
        if (null == dto || StringUtils.isBlank(dto.getId())) {
            log.error("百科问答ID不能为空和NULL");
            return ResultEntity.error("百科问答ID不能为空和NULL");
        }
        CommunityEncyclopediaQuestionAnswer encyclopediaQuestionAnswer = getCommunityEncyclopediaQuestionAnswerById(dto.getId());
        // 取消推荐
        encyclopediaQuestionAnswer.setIsHotRecommend(YesOrNoEnum.NO.getValue());
        encyclopediaQuestionAnswer.setIsHotRecommendTop(YesOrNoEnum.NO.getValue());
        encyclopediaQuestionAnswer.setRecommendIndex(null);
        encyclopediaQuestionAnswer.setUpdateTime(new Date());
        try {
            elasticsearchClient.update(e -> e
                    .index(CommonConstants.WIKI_INDEX)
                    .id(dto.getId())
                    .doc(encyclopediaQuestionAnswer), CommunityEncyclopediaQuestionAnswer.class
            );
        } catch (Exception e) {
            log.error("取消推荐操作失败:{}", e.getMessage());
            return ResultEntity.error("取消推荐操作失败");
        }
        return ResultEntity.success("取消推荐操作成功");
    }


    // todo user服务接口指向有问题，应该是 user服务的接口写错了，应和search服务保持一致，待修改
    /**
     * 管理后台查询
     * 多条件查询，分页，过滤，排序
     */
    @PostMapping("/listEncyclopediaQuestionAnswer")
    @ApiOperation(value = "分页查询百科问答列表")
    public ResultEntity<?> listEncyclopediaQuestionAnswer(@RequestBody CommunityEncyclopediaQuestionAnswerDTO questionAnswerDTO) throws IOException {
        BoolQuery.Builder boolQuery = QueryBuilders.bool()
                .must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(1))._toQuery());

        if (!StringUtils.isBlank(questionAnswerDTO.getQuestion())) {
            boolQuery.must(WildcardQuery.of(w -> w.field(CommunityEncyclopediaQuestionAnswer.FIELD_MANAGER_QUESTION).value("*" + questionAnswerDTO.getQuestion() + "*"))._toQuery());
        }
        if (null != questionAnswerDTO.getCategory()) {
            boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY).value(questionAnswerDTO.getCategory()))._toQuery());
        }
        if (null != questionAnswerDTO.getIsCategoryTop()) {
            boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_IS_CATEGORY_TOP).value(questionAnswerDTO.getIsCategoryTop()))._toQuery());
        }
        if (null != questionAnswerDTO.getIsHotRecommend()) {
            boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_IS_HOT_RECOMMEND).value(questionAnswerDTO.getIsHotRecommend()))._toQuery());
        }
        if (null != questionAnswerDTO.getStatus()) {
            boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(questionAnswerDTO.getStatus()))._toQuery());
        }

        SearchResponse<CommunityEncyclopediaQuestionAnswer> search = elasticsearchClient.search(s -> s
                        .index(CommonConstants.WIKI_INDEX)

                        .query(q -> q
                                .bool(boolQuery.build())
                        )
                        .from(questionAnswerDTO.getPageNo() - 1)
                        .size(questionAnswerDTO.getPageSize())
                        .sort(sortOptionsBuilder -> sortOptionsBuilder
                                .field(fieldSortBuilder -> fieldSortBuilder
                                        .field(CommunityEncyclopediaQuestionAnswer.FIELD_TOP_INDEX).order(SortOrder.Asc)
                                        .field(CommunityEncyclopediaQuestionAnswer.FIELD_NORMAL_INDEX).order(SortOrder.Asc)
                                        .field(CommunityEncyclopediaQuestionAnswer.FIELD_CREATE_TIME).order(SortOrder.Desc)))
                , CommunityEncyclopediaQuestionAnswer.class);

        List<CommunityEncyclopediaQuestionAnswer> questionAnswerList = search.hits().hits().stream().map(Hit::source).collect(Collectors.toList());

        List<CommunityEncyclopediaQuestionCategory> questionCategoryList = communityEncyclopediaQuestionCategoryService.listCommunityEncyclopediaQuestionCategory();
        if (null != questionCategoryList && !questionCategoryList.isEmpty()) {
            List<Integer> questionCategoryIdList = questionCategoryList.stream().map(CommunityEncyclopediaQuestionCategory::getId).collect(Collectors.toList());
            questionAnswerList = questionAnswerList.stream()
                    .filter(questionAnswer -> questionCategoryIdList.contains(questionAnswer.getCategory()))
                    .collect(Collectors.toList());
        } else {
            questionAnswerList = new ArrayList<>();
        }

        Map<String, Object> map = new HashMap<>(16);
        map.put("total", Objects.requireNonNull(search.hits().total()).value());
        map.put("dataList", questionAnswerList);
        return ResultEntity.success(map);
    }

    /**
     * APP查询
     * 根据问题关键词和问题类别查询问答列表
     */
    @PostMapping(value = "/listCommunityEncyclopediaQuestionAnswer")
    @ApiOperation(value = "APP百科问答搜索")
    public ResultEntity<?> listCommunityEncyclopediaQuestionAnswer(@RequestBody CommunityEncyclopediaQuestionAnswerDTO questionAnswerDTO) throws IOException {
        // 设置多条件
        BoolQuery.Builder boolQuery = QueryBuilders.bool()
                .must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(YesOrNoEnum.YES.getValue()))._toQuery())
                .must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(YesOrNoEnum.YES.getValue()))._toQuery());

        if (!StringUtils.isBlank(questionAnswerDTO.getQuestion())) {
            boolQuery.must(WildcardQuery.of(w -> w.field(CommunityEncyclopediaQuestionAnswer.FIELD_QUESTION).value(questionAnswerDTO.getQuestion()))._toQuery());
        }
        SortOptions.Builder sortBuilder = new SortOptions.Builder();
        if (null != questionAnswerDTO.getCategory()) {
            // 热门问答
            if (0 == questionAnswerDTO.getCategory()) {
                boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_IS_HOT_RECOMMEND).value(YesOrNoEnum.YES.getValue()))._toQuery());
                sortBuilder.field(FieldSort.of(f -> f.field(CommunityEncyclopediaQuestionAnswer.FIELD_RECOMMEND_INDEX).order(SortOrder.Asc)));
                // 分类别查询
            } else {
                boolQuery.must(TermQuery.of(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY).value(questionAnswerDTO.getCategory()))._toQuery());
                sortBuilder.field(FieldSort.of(f -> f.field(CommunityEncyclopediaQuestionAnswer.FIELD_TOP_INDEX).order(SortOrder.Asc)));
            }
        }

        // 设置排序字段
        sortBuilder.field(FieldSort.of(f -> f.field(CommunityEncyclopediaQuestionAnswer.FIELD_NORMAL_INDEX).order(SortOrder.Asc)));
        sortBuilder.field(FieldSort.of(f -> f.field(CommunityEncyclopediaQuestionAnswer.FIELD_CREATE_TIME).order(SortOrder.Desc)));

        SearchResponse<CommunityEncyclopediaQuestionAnswer> search = elasticsearchClient.search(s -> s
                        .index(CommonConstants.WIKI_INDEX)

                        .query(q -> q
                                .bool(boolQuery.build())
                        )
                        .from(questionAnswerDTO.getPageNo() - 1)
                        .size(questionAnswerDTO.getPageSize())
                        .sort(sortBuilder.build())
                , CommunityEncyclopediaQuestionAnswer.class);

        List<CommunityEncyclopediaQuestionCategory> categoryList = communityEncyclopediaQuestionCategoryService.getAllList();
        List<CommunityEncyclopediaQuestionAnswer> questionAnswerList = search.hits().hits().stream().map(Hit::source).collect(Collectors.toList());

        List<CommunityEncyclopediaQuestionAnswerVO> questionAnswerVOList = questionAnswerList.stream()
                .map(questionAnswer -> {
                    CommunityEncyclopediaQuestionAnswerVO questionAnswerVO = new CommunityEncyclopediaQuestionAnswerVO();
                    questionAnswerVO.setId(questionAnswer.getId());
                    questionAnswerVO.setQuestion(questionAnswer.getQuestion());
                    questionAnswerVO.setAnswer(questionAnswer.getAnswer());

                    if (categoryList != null && !categoryList.isEmpty()) {
                        for (CommunityEncyclopediaQuestionCategory communityEncyclopediaQuestionCategory : categoryList) {
                            if (communityEncyclopediaQuestionCategory.getId().equals(questionAnswer.getCategory())) {
                                questionAnswerVO.setCategoryTitle(communityEncyclopediaQuestionCategory.getTitle());
                                break;
                            }
                        }
                    }
                    return questionAnswerVO;
                }).collect(Collectors.toList());

        Map<String, Object> map = new HashMap<>(16);
        map.put("total", Objects.requireNonNull(search.hits().total()).value());

        map.put("dataList", questionAnswerVOList);
        return ResultEntity.success(map);
    }

    private CommunityEncyclopediaQuestionAnswer getCommunityEncyclopediaQuestionAnswerById(String id) throws BusinessException {
        if (null == id) {
            log.error("问答ID不能为NULL");
            throw new BusinessException(ResultCode.ERROR.getCode(), "问答ID不能为NULL");
        }
        GetResponse<CommunityEncyclopediaQuestionAnswer> optionalById;
        try {
            // 根据问答ID查询到问答数据记录
            optionalById = elasticsearchClient.get(s -> s.index(CommonConstants.WIKI_INDEX).id(id), CommunityEncyclopediaQuestionAnswer.class);
        } catch (Exception e) {
            log.error("es服务异常，请联系管理员：{}", e.getMessage());
            throw new BusinessException(ResultCode.ERROR.getCode(), "es服务异常，请联系管理员");
        }

        if (!optionalById.found()) {
            log.error("根据ID未查询到对应问答数据记录");
            throw new BusinessException(ResultCode.ERROR.getCode(), "根据问答ID未查询到问答数据记录");
        }
        CommunityEncyclopediaQuestionAnswer communityEncyclopediaQuestionAnswer = optionalById.source();
        assert communityEncyclopediaQuestionAnswer != null;
        if (!ValidOrInvalidEnum.isValid(communityEncyclopediaQuestionAnswer.getType())) {
            log.error("根据ID查询的问答数据无效");
            throw new BusinessException(ResultCode.ERROR.getCode(), "根据ID查询的问答数据无效");
        }
        return communityEncyclopediaQuestionAnswer;

    }

    /**
     * 查询有效的百科问答列表
     */
    @PostMapping("/findValidListByType")
    @ApiOperation(value = "查询有效的百科问答列表")
    public ResultEntity<?> findValidList() throws BusinessException {
        try {
            // 根据问答ID查询到问答数据记录
            List<CommunityEncyclopediaQuestionAnswer> encyclopediaQuestionAnswerList = findValidListByType(ValidOrInvalidEnum.VALID.getValue());
            return ResultEntity.success(encyclopediaQuestionAnswerList);
        } catch (Exception e) {
            log.error("es服务异常，请联系管理员：{}", e.getMessage());
            return ResultEntity.error("es服务异常，请联系管理员");
        }
    }




    /**
     * 根据问答分类查询有效上线置顶的百科问答
     */
    @PostMapping("/findValidListByCategory")
    @ApiOperation(value = "根据问答分类查询有效上线置顶的百科问答")
    public ResultEntity<?> findValidListByCategory(@RequestBody CommunityEncyclopediaQuestionAnswerDTO dto) throws BusinessException {
        if (null == dto || null == dto.getCategory()) {
            log.error("百科问答类别不能为NULL");
            return ResultEntity.error("百科问答类别不能为NULL");
        }
        try {
            // 根据问答ID查询到问答数据记录
            List<CommunityEncyclopediaQuestionAnswer> questionAnswerList = findValidListByTypeAndStatusAndCategoryAndIsCategoryTop(ValidOrInvalidEnum.VALID.getValue(),
                            YesOrNoEnum.YES.getValue(), dto.getCategory(), YesOrNoEnum.YES.getValue());
            return ResultEntity.success(questionAnswerList);
        } catch (Exception e) {
            log.error("es服务异常，请联系管理员：{}", e.getMessage());
            return ResultEntity.error("es服务异常，请联系管理员");
        }
    }



    /**
     * 查询有效上线推荐的百科问答
     */
    @PostMapping("/findValidListByRecommend")
    @ApiOperation(value = "查询有效上线推荐的百科问答")
    public ResultEntity<?> findValidListByRecommend() throws BusinessException {
        try {
            List<CommunityEncyclopediaQuestionAnswer> questionAnswerList =
                    findValidListByTypeAndStatusAndIsHotRecommend(ValidOrInvalidEnum.VALID.getValue(),
                            YesOrNoEnum.YES.getValue(), YesOrNoEnum.YES.getValue());
            return ResultEntity.success(questionAnswerList);
        } catch (Exception e) {
            log.error("es服务异常，请联系管理员：{}", e.getMessage());
            return ResultEntity.error("es服务异常，请联系管理员");
        }
    }



    /**
     * 查询有效上线关联产品productCode的百科问答
     */
    @PostMapping("/findValidListByProductCode")
    @ApiOperation(value = "查询有效上线关联产品productCode的百科问答")
    public ResultEntity<?> findValidListByProductCode(@RequestBody ProductLibraryDTO dto) throws BusinessException {
        try {
            List<CommunityEncyclopediaQuestionAnswer> questionAnswerList =
                    findValidListByTypeAndStatusAndProductCode(ValidOrInvalidEnum.VALID.getValue(),
                            YesOrNoEnum.YES.getValue(), dto.getProductCode());
            return ResultEntity.success(questionAnswerList);
        } catch (Exception e) {
            log.error("es服务异常，请联系管理员：{}", e.getMessage());
            return ResultEntity.error("es服务异常，请联系管理员");
        }
    }

    private boolean findValidOneByTypeAndStatusAndCategoryAndTopIndex(Integer type, Integer status, Integer category, Integer topIndex) throws IOException {
        return Objects.requireNonNull(elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(status)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY).value(category)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TOP_INDEX).value(topIndex)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().total()).value() > 0;
    }

    private boolean findValidOneByTypeAndStatusAndRecommendIndex(Integer type, Integer status, Integer recommendIndex) throws IOException {
        return Objects.requireNonNull(elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(status)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_RECOMMEND_INDEX).value(recommendIndex)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().total()).value() > 0;
    }

    private List<CommunityEncyclopediaQuestionAnswer> findValidListByTypeAndCategory(Integer type, Integer category) throws IOException {
        return elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY).value(category)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }
    private List<CommunityEncyclopediaQuestionAnswer> findValidListByType(Integer type) throws IOException {
        return elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }

    private List<CommunityEncyclopediaQuestionAnswer> findValidListByTypeAndStatusAndCategoryAndIsCategoryTop(Integer type, Integer status, Integer category, Integer categoryTop) throws IOException {
        return elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(status)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY).value(category)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_IS_CATEGORY_TOP).value(categoryTop)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }

    private List<CommunityEncyclopediaQuestionAnswer> findValidListByTypeAndStatusAndIsHotRecommend(Integer type, Integer status, Integer isHotRecommend) throws IOException {
        return elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(status)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_IS_HOT_RECOMMEND).value(isHotRecommend)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }

    private List<CommunityEncyclopediaQuestionAnswer> findValidListByTypeAndStatusAndProductCode(Integer type, Integer status, String productCode) throws IOException {
        return elasticsearchClient.search(s -> s
                .index(CommonConstants.WIKI_INDEX)
                .query(q -> q
                        .bool(b -> b
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE).value(type)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS).value(status)))
                                .must(m -> m.term(t -> t.field(CommunityEncyclopediaQuestionAnswer.FIELD_PRODUCT_CODE).value(productCode)))
                        )
                ), CommunityEncyclopediaQuestionAnswer.class).hits().hits().stream().map(Hit::source).collect(Collectors.toList());
    }

    private void createIndexDetail() throws IOException {
        CreateIndexResponse response = elasticsearchClient
                .indices()
                .create(c -> c
                        .index(CommonConstants.WIKI_INDEX)
                        .settings(sBuilder -> sBuilder
                                .index(iBuilder -> iBuilder
                                        // 三个分片
                                        .numberOfShards("5")
                                        // 一个副本
                                        .numberOfReplicas("1")))
                        .mappings(mBuilder -> mBuilder
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_ID,  pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(256)))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_QUESTION,  pBuilder -> pBuilder.text(tBuilder -> tBuilder.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_MANAGER_QUESTION,  pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(256)))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_ANSWER,  pBuilder -> pBuilder.text(tBuilder -> tBuilder.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_CATEGORY,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_STATUS,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_NORMAL_INDEX,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_IS_CATEGORY_TOP,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_TOP_INDEX,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_IS_HOT_RECOMMEND,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_IS_HOT_RECOMMEND_TOP,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_RECOMMEND_INDEX,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_TYPE,  pBuilder -> pBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_OPERATE_REAL_NAME,  pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(256)))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_OPERATE_DJ_CODE,  pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(256)))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_CREATE_TIME,  pBuilder -> pBuilder.date(datePropertyBuilder -> datePropertyBuilder.format("yyyy-MM-dd HH:mm:ss")))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_UPDATE_TIME,  pBuilder -> pBuilder.date(datePropertyBuilder -> datePropertyBuilder.format("yyyy-MM-dd HH:mm:ss")))
                                .properties(CommunityEncyclopediaQuestionAnswer.FIELD_PRODUCT_CODE,  pBuilder -> pBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder.ignoreAbove(256)))
                        )
                );
        log.info("createIndexDetail方法，acknowledged={}", response.acknowledged());
    }
}

es8代码升级遇到的问题

pom.xml 关于 jackson 的版本不兼容问题

引入依赖后服务无法正常启动

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.13.3</version>
</dependency>

通过mvn命名进行排查

mvn dependency:tree

观察依赖树发现，可能是 jackson 2.13版本下其他依赖可能和已有版本存在冲突

解决方案

将 依赖保持和已有依赖一致

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.11.4</version>
</dependency>
ElasticSearch查询报错JsonpMappingException

说明某些字段无法转换，根据报错信息排查

重新导入es数据，删除version、timestamp 两个额外字段即可

或者使用兜底方案（不一定成）：

在实体类上加注解

@JsonIgnoreProperties(ignoreUnknown=true)

ElasticSearch查询报错JsonpMappingException

WildcardQuery 查询不到数据
Query wildcardQuery = WildcardQuery.of(w -> w.field("managerQuestion").value("*" + questionAnswerDTO.getQuestion() + "*"))._toQuery();
SearchResponse<CommunityEncyclopediaQuestionAnswer> search = elasticsearchClient.search(s -> s
    .index("community_encyclopedia")

    .query(q -> q
            .bool(b -> b
                    .must(type)
                    .must(wildcardQuery)
            )
    ), CommunityEncyclopediaQuestionAnswer.class);

将字段 managerQuestion 改为 managerQuestion.keyword

查询后Date类型 的 @JsonFormat 未生效

待解决

es间数据迁移

迁移方式对比

参考链接：三种常用的 Elasticsearch 数据迁移方案

es6.8到7.1.1

采用方式

ElasticSearch-dump

安装

npm install elasticdump -g

在 D:\environment\nodejs\node_global\node_modules\elasticdump\bin 目录下运行命令

相关命令

在线迁移单个索引
elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/community_encyclopedia --output=http://localhost:9200/community_encyclopedia --type=mapping
elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/community_encyclopedia --output=http://localhost:9200/community_encyclopedia --type=data


elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/orgnization --output=http://localhost:9200/orgnization --type=mapping
elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/orgnization --output=http://localhost:9200/orgnization --type=data

elasticdump  --input=http://localhost:9200/dragon --output=http://"elastic:pB0@uI"@10.221.50.102:9200/dragon --type=mapping
elasticdump  --input=http://localhost:9200/dragon --output=http://"elastic:pB0@uI"@10.221.50.102:9200/dragon --type=data

elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/community_encyclopedia --output=http://"elastic:pB0@uI"@localhost:9200/community_encyclopedia --type=mapping
elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/community_encyclopedia --output=https://"elastic:pB0@uI"@localhost:9200/community_encyclopedia --type=data

elasticdump  --input=http://"elastic:pB0@uI"@10.221.50.102:9200/community_encyclopedia --output=http://"elastic:pB0@uI"@10.7.176.72:9200/community_encyclopedia --type=mapping

http://10.7.176.72:9200/
dragon

离线迁移全部索引（需要手动创建目录）  -- 未测试
# 导出
multielasticdump --direction=dump --match='^.*$' --input=http://"elastic:pB0@uI"@10.221.50.102:9200 --output=/tmp/es_backup --includeType='data,mapping' --limit=2000
# 导入
multielasticdump --direction=load --match='^.*$' --input=/tmp/es_backup --output=http://localhost:9200 --includeType='data,mapping' --limit=2000 

multielasticdump --direction=dump --match='^.*$' --input=http://"elastic:pB0@uI"@10.221.50.102:9200 --output=E:\file\back_test --includeType='data,mapping' --limit=2000
multielasticdump --direction=load --match='^.*$' --input=E:\file\back_test --output=http://localhost:9200 --includeType='data,mapping' --limit=2000
参数含义
  • --input: 源地址，可为 ES 集群 URL、文件或 stdin,可指定索引，格式为：{protocol}://{host}:{port}/{index}--input-index: 源 ES 集群中的索引
  • --output: 目标地址，可为 ES 集群地址 URL、文件或 stdout，可指定索引，格式为：{protocol}://{host}:{port}/{index}
  • --output-index: 目标 ES 集群的索引
  • --type: 迁移类型，默认为 data，表明只迁移数据，可选 settings, analyzer, data, mapping, alias

参考链接

elasticsearch-dump工具：GitHub - elasticsearch-dump/elasticsearch-dump: Import and export tools for elasticsearch & opensearch

使用该工具迁移文档：使用ElasticSearch-dump进行数据迁移、备份_elasticsearchdump 大于4g_刘李404not found的博客-CSDN博客

es6.8 到 8.7.0（本地）

采用方式

ElasticSearch-dump（需要配置安全证书，暂时不采用这种方式）

logstash

启动方式

在logstash新建 mysql/logstash-es.conf 文件

然后用一下命令启动

bin\logstash.bat -f mysql\logstash-es.conf

配置文件

#logstash输入配置
input {
   elasticsearch {
		hosts => ["10.221.50.102:9200"]
		index => "community_encyclopedia"
    user => "elastic"
    password => "pB0@uI"
    #设置为true，将会提取ES文档的元数据信息，例如index、type和id。貌似没用
    # docinfo => true
	}
}
#logstash输出配置
output {
  # stdout { codec =>  json_lines}
  stdout { codec =>  rubydebug}
  elasticsearch {
		hosts => ["localhost:9200"]
    user => "elastic"
    password => "pB0@uI"
    ssl => true
    ssl_certificate_verification => false 
		index => "community_encyclopedia"
		document_id => "%{id}"
	}
}

es6.8 到 8.7.0（测试环境）

特别说明

为保证es中索引结构相同，请先在代码端/页面 运行下保存相关接口，在es中创建索引后再进行数据迁移

采用方式

ElasticSearch-dump（需要配置安全证书，暂时不采用这种方式）

logstash

启动方式

在logstash新建 mysql/logstash-es.conf 文件

然后用一下命令启动

bin\logstash.bat -f mysql\logstash-es.conf

[linux]: 
sudo ./bin/logstash -f mysql_test/logstash-es.conf
# sudo 加不加都可
./bin/logstash -f extend/job/logstash-t_settle_order_info.conf

配置文件

#logstash输入配置
input {
   elasticsearch {
		hosts => ["10.221.50.102:9200"]
		index => "community_encyclopedia"
    user => "elastic"
    password => "pB0@uI"
    #设置为true，将会提取ES文档的元数据信息，例如index、type和id。貌似没用
    # docinfo => true
	}
}
filter {
  mutate {
    remove_field => ["@version", "@timestamp"]
  }
}
#logstash输出配置
output {
  # stdout { codec =>  json_lines}
  stdout { codec =>  rubydebug}
  elasticsearch {
		# hosts => ["10.7.176.72:9200"]
		hosts => ["10.7.176.72:9200","10.7.176.73:9200","10.7.176.74:9200"]
		# hosts => ["10.7.176.72:9300"]
    user => "elastic"
    # password => "pB0@uI"
    password => "Mvwm@n12nal"
    # ssl => true
    # ssl_certificate_verification => false 
		index => "community_encyclopedia"
		document_id => "%{id}"
	}
}

mysql到es数据迁移

订单历史数据从mysql迁移至es(单表测试)

搭伙app订单查询接口分析

url:https://agentd.djbx.com/order/orderInfo/orderInfoList
method:POST
requestTime:2023.06.28-14:02:56:820
responseTime:2023.06.28-14:02:57:31
duration:211ms
body:{
  "pageNo": "1",
  "pageSize": "10",
  "param": {
    "memberId": "4OnzMs73",
    "productType": [],
    "status": "3",
    "policyStatus": null,
    "startDate": null,
    "endDate": null,
    "appntName": null,
    "lcnNo": null
  }
}
params:{}
header:{content-type: application/json; charset=utf-8, systemType: android, systemVersion: 9, deviceID: PQ3A.190705.003, bundleId: com.djcx.dahuo, appVersion: 4.3.10, phoneModel: Xiaomi 2203121C, token: 7pyvLZQfD6UUP0d4yPTgkXTgdq8, content-length: 174}

采用方式

logstash

启动运行

在【程序目录】目录执行以下命令启动：

【windows】bin\logstash.bat -f mysql\logstash-db-sync.conf
【linux】nohup ./bin/logstash -f mysql_test/logstash-es.conf &   (后台)
sudo ./bin/logstash -f mysql_test/logstash-es.conf

需求

从mysql：dh.order 的 t_car_order 中迁移已支付和已取消的订单到es

条件：ORDER_STATUS in (3,4)

数据量

一个月30w，一年300w（占用磁盘大约1GB）

索引类创建方案

按月分索引：数据量太少，不建议

按年划分索引：感觉可行，试一下

都试一下，看看查询效率，删除效率如何（什么情况下会删除数据呢？）

logstash相关配置

#logstash输入配置
input {
  #jdbc输入配置，用来指定mysql中需要同步的数据查询SQL及同步周期
  jdbc {
    type => "jdbc"
    jdbc_connection_string => "jdbc:mysql://10.221.50.106:3306/dh_order?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false"
    # 数据库连接账号密码；
		jdbc_user => "dh_test"
		jdbc_password => "Y2017dh123"
    # MySQL依赖包路径；
		jdbc_driver_library => "mysql/mysql-connector-java-5.1.49.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    # 数据库重连尝试次数
		connection_retry_attempts => "3"
    # 判断数据库连接是否可用，默认false不开启
		jdbc_validate_connection => "true"
    # 数据库连接可用校验超时时间，默认3600S
		jdbc_validation_timeout => "3600"
    # 是否开启分页
    jdbc_paging_enabled => true
    # statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC"
    # statement => "SELECT * FROM `t_car_order`"
    # statement => "SELECT DATE_FORMAT(tco.create_time, '%Y.%m') index_date, tco.* FROM t_car_order tco limit 200"
    statement => "SELECT DATE_FORMAT(tco.create_time, '%Y') index_year, tco.* FROM t_car_order tco"
    # statement => "SELECT ID, order_no, order_type, MEMBER_ID, owner_id, team_id, CHANNEL_CODE, pay_type, PAY_NO, TOTAL_AMOUNT, PAY_AMOUNT, DISCOUNT_AMOUNT, DISCOUNT_RATE, ROLE_TYPE, PRODUCT_VALUE, ORDER_STATUS, SUB_ORDER_STATUS, step_url, total_annual_prem, paidTime, subject, transfer, identity, msg, date_format(update_time,'%Y-%m-%d %H:%i:%s') update_time, date_format(create_time,'%Y-%m-%d %H:%i:%s') create_time, uuid, renewal_redis_key, clue_id, union_type, system_type, operation_code, cx_order_no, fin_typ 
    # FROM t_car_order limit 10"
    # statement => "SELECT * FROM t_car_order limit 1"
    # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		# lowercase_column_names => false
    # Value can be any of: fatal,error,warn,info,debug，默认info；
		# sql_log_level => warn
		sql_log_level => debug
    # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
		# record_last_run => true
		 # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
		# use_column_value => true
		 # 需要记录的字段，用于增量同步，需是数据库字段
		# tracking_column => "ModifyTime"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		# tracking_column_type => timestamp
		 # record_last_run上次数据存放位置；
		# last_run_metadata_path => "mysql/last_id.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		# clean_run => false
		# 设置定时任务间隔  含义：分、时、天、月、年，全部为*默认含义为每分钟跑一次任务，这里设置为每5分钟同步一次
    # schedule => "*/5 * * * * *"

    # 用来控制增量更新的字段，一般是自增id或者创建、更新时间，注意这里要采用sql语句中select采用的字段别名
    # tracking_column => "unix_ts_in_secs"
    # tracking_column 对应字段的类型
    # tracking_column_type => "numeric"
    # timezone => "Asia/Shanghai" # 你的时区
    


  }
}
#logstash输入数据的字段匹配和数据过滤
# filter {
#   mutate {
#     copy => { "id" => "[@metadata][_id]"}
#     remove_field => ["id", "@version", "unix_ts_in_secs"]
#   }
# }
filter {
  # date {
  #   match => ["update_time", "yyyy-MM-dd HH:mm:ss"]
  #   target => "update_time"
  # }
  # grok {
	#   match => {"message" => "(?<create_time>(?:%{YEAR}-%{MONTHNUM2}-%{MONTHDAY} %{HOUR}:%{MINUTE}:%{SECOND}))"}
	# }
  # mutate {
  #       add_field => {"temp_ts" => "%{create_time}"}
  #   }
  # date {
  #       match => ["temp_ts", "yyyy-MM-dd HH:mm:ss"]
  #       target => "@timestamp"
  #   }
  # date {
  #   match => ["create_time", "yyyy-MM-dd HH:mm:ss"]
  #   target => "create_time"
  # }
  
  
  # ruby {
  #   code => 'event.set("create_time", event.get("create_time").strftime("%Y-%m-%d %H:%M:%S"))'
  # }
  #  date {
  #   # match => ["create_time", "yyyy-MM-dd HH:mm:ss"]
  #   match => ["create_time", "ISO8601"]
  #   target => "create_time"
  #   timezone => "Asia/Shanghai" # 你的时区
  #   # timezone => "America/New_York" # 你的时区
  # }
  # mutate {
  #   add_field => { "index_date" => "%{create_time}" }
  # }
  # mutate {
  #   rename => { "create_time_string" => "index_date" }
  # }
  #  date {
  #   # match => ["index_date", "ISO8601"]
  #   match => ["index_date", "ISO8601"]
  #   # target => "index_date"
  # }
  
    
  # }
  # date {
  #   match => ["index_date", "yyyy-MM-dd HH:mm:ss"]
  #   # target => "index_date"
  #   # target => "index_date"
  # }
  # mutate {
  #   add_field => {
  #     "index_date1" => "%{index_date}"
  #   }
  # mutate  {
  #     remove_field => ["@version","index_date"]
  # }
  

  ruby { 
        code => "event.set('@timestamp', event.get('@timestamp').time.localtime + 8*60*60)" 
    }

  ruby { 
        code => "event.set('update_time', event.get('update_time').time.localtime + 8*60*60)" 
    }

  ruby { 
        code => "event.set('create_time', event.get('create_time').time.localtime + 8*60*60)" 
    }

  # ruby {
  #     code => "event.set('create_time',event.get('timestamp'))"
  # }

  # mutate {
  #   add_field => { "[@metadata][index_date]" => "%{index_date}" }
  #   remove_field => ["@version", "index_date"]
  # }
  mutate {
    add_field => { "[@metadata][index_year]" => "%{index_year}" }
    remove_field => ["@version", "index_year"]
  }
 
}
#logstash输出配置
output {
  # 采用stdout可以将同步数据输出到控制台，主要是调试阶段使用
  # stdout { codec =>  json_lines}
  stdout { codec =>  rubydebug}

  # 指定输出到ES的具体索引
  # elasticsearch {
  #     index => "rdbms_sync_idx"
  #     document_id => "%{[@metadata][_id]}"
  # }
  elasticsearch {
		 # host => "192.168.1.1"
		 # port => "9200"
		 # 配置ES集群地址
		# hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
		hosts => ["localhost:9200"]
		 # 索引名字，必须小写
		# index => "t_car_order-%{+YYYY.MM.dd}"
		# index => "t_car_order-%{index_date}"
		# index => "t_car_order-%{[@metadata][index_date]}"
		# index => "t_car_order_@timestamp"
		index => "t_car_order"
		# index => "t_car_order-%{[@metadata][index_year]}"
		 # 数据唯一索引（建议使用数据库KeyID）
		# document_id => "%{KeyId}"
		document_id => "%{id}"
		# document_id => "ID"
	}
}

参考链接

使用logstash同步MySQL数据到ES

使用logstash实现mysql到ES的数据迁移

通过Logstash实现mysql数据定时增量同步到ES

通过Logstash将RDS MySQL数据同步至Elasticsearch

日期插件语法（官网）

问题

logstash日期格式转换问题（暂未解决）

多个logstash版本测试

日期转换不可用，sql层面也不行，打算在接口层处理。

日期格式问题，暂缓。。。 心累了，估计是日期插件问题，全网没有找到解决方案，走sql修改吧

提了个bug：无法将date类型进行格式化操作 · Issue #158 · logstash-plugins/logstash-filter-date · GitHub

参考链接

Logstash:获取业务时间并替换@timestamp

Elasticsearch 滞后8个小时等时区问题，一网打尽！

logstash 7.x 中时间问题，@timestamp 与本地时间相差 8个小时

现有订单数据保存到es方案

修改保存接口，保存到mysql改为es，或者可以考虑消息队列中间件

订单历史数据从mysql迁移至es(实际业务场景)

背景1：由于无法在单sql中查询数据，因此无法使用logstash工具

背景2：订单迁移分为订单详情和订单列表两大类

背景3：订单信息涉及多张表，无法进行单表简单迁移

迁移方案

订单信息获取

search服务设置一个定时任务，通过feign调用car服务查询接口，进行全量和增量查询需要的订单信息

订单信息迁移至es

分两大类创建索引：订单详情和订单列表，每一类按照年维度进行索引构建，用别名进行查询

es结构按订单号为id，报文信息为message

定时任务详情

参数

订单创建日期开始日期，订单创建日期截止日期

默认参数：最近三个月

如何获取符合条件的增量数据的订单号？通过binlog日志

通过时间段（每三个月跑一次）

参考链接

java与es8实战之六：用JSON创建请求对象(比builder pattern更加直观简洁)

参考文档

Elasticsearch7.x中文文档

es7.0版本升级变化(官网)

ELK Stack

文章来源:https://blog.csdn.net/RAXCL/article/details/135387914
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
最新文章