Elasticsearch 简介与使用

Elasticsearch: 权威指南

ES入门

ES6.4官方文档

什么是Elastic Search

​ ElasticSearch 简称 ES ，是基于Apache Lucene构建的开源搜索引擎，是当前流行的企业级搜索引擎。Lucene本身就可以被认为迄今为止性能最好的一款开源搜索引擎工具包，但是lucene的API相对复杂，需要深厚的搜索理论。很难集成到实际的应用中去。但是ES是采用java语言编写，提供了简单易用的RestFul API，开发者可以使用其简单的RestFul API，开发相关的搜索功能，从而避免lucene的复杂性。

同时ES还是一个分布式文档数据库

​

ES的应用场景

​ ES主要以轻量级JSON作为数据存储格式，它被用作全文检索、结构化搜索、分析以及这三个功能的组合（ELK）

​ 我们本次使用主要是使用ES的结构化搜索功能来替代Mysql搜索

ES相关概念介绍

ES	Mysql
Index	数据库
Type	数据表
Document	行
Mapping	Schema

1、索引（_index）

​ 类似MYSQL的数据库

2、类型（_type）

​ 类似MYSQL的表

3、字段（field）

​ 类型里的字段，类似MYSQL的表字段

4、文档（document）

​ 类似MYSQL的行数据

5、映射(mapping)

​ 类似MYSQL的DDL，声明索引与类型的关系、类型与字段的关系，是json格式

6、正向索引

​ 以文档的ID为关键字，表中记录文档中每个字的位置信息，查找时扫描表中每个文档中字的信息直到找出所有包含查询关键字的文档，比如Mysql的B+聚餐索引

7、倒排索引

​ 倒排表以字或词为关键字进行索引，表中关键字所对应的记录表项记录了出现这个字或词的所有文档，一个表项就是一个字表段，它记录该文档的ID和字符在该文档中出现的位置情况，比如ES索引结构

PUT /dangdang             
{
  "mappings": {
    "product": {
      "properties": {
        	"title":    { "type": "text"  },
        	"name":     { "type": "text"  },
       		"age":      { "type": "integer" },
        	"created":  {
         		 "type":   "date"
        		}
      		}
    	}
  	}
}

8、面向文档

​ 应用中的对象很少只是简单的键值列表，更多时候它拥有复杂的数据结构，比如包含日期、地理位置、另一个对象或者数组。
​ 总有一天你会想到把这些对象存储到数据库中。将这些数据保存到由行和列组成的关系数据库中，就好像是把一个丰富，信息表现力强的对象拆散了放入一个非常大的表格中：你不得不拆散对象以适应表模式（通常一列表示一个字段），然后又不得不在查询的时候重建它们。
​ Elasticsearch是面向文档(document oriented)的，这意味着它可以存储整个对象或文档(document)。然而它不仅仅是存储，还会索引(index)每个文档的内容使之可以被搜索。在Elasticsearch中，你可以对文档（而非成行成列的数据）进行索引、搜索、排序、过滤。这种理解数据的方式与以往完全不同，这也是Elasticsearch能够执行复杂的全文搜索的原因之一。

9、节点(node)和集群(node)

Elasticsearch可以作为一个独立的单个搜索服务器。不过，为了能够处理大型数据集，实现容错和高可用性，Elasticsearch可以运行在许多互相合作的服务器上。这些服务器称为集群（cluster），形成集群的每个服务器称为节点（node）。

10、分片(shards )

当有大量的文档时，由于内存的限制、硬盘能力、处理能力不足、无法足够快地响应客户端请求等，一个节点可能不够。在这种情况下，数据可以分为较小的称为分片（shard）的部分（其中每个分片都是一个独立的Apache Lucene索引）。每个分片可以放在不同的服务器上，因此，数据可以在集群的节点中传播。当你查询的索引分布在多个分片上时，Elasticsearch会把查询发送给每个相关的分片，并将结果合并在一起，而应用程序并不知道分片的存在。此外，多个分片可以加快索引。

11、副本(replicas)

为了提高查询吞吐量或实现高可用性，可以使用分片副本。副本（replica）只是一个分片的精确复制，每个分片可以有零个或多个副本。换句话说，Elasticsearch可以有许多相同的分片，其中之一被自动选择去更改索引操作。这种特殊的分片称为主分片（primary shard），其余称为副本分片（replica shard）。在主分片丢失时，例如该分片数据所在服务器不可用，集群将副本提升为新的主分片。

12、字段类型

text:
按规则分词，至少分词1个词，做全文检索（不要存放空值，会导致性能浪费）
默认分词器standard
应用场景：用于分词检索场景
keyword:
不分词，仅仅一个词，查询效率高，早期没有这个类型，用String代替，通过属性设置
应用场景：固定文本，无需全文检索，如姓名，省，商品类目
整数：Long（64bit）、Integer（32bit、21亿）、short（16bit）、Byte（8bit）
浮点：Double 、Float 、Half Float 、Scaled float(缩放浮点类型，解决浮点类型计算精度的错误问题)
日期类型 date
日期格式化 format（定义日期类型一定要定义format格式，防止后面一些不确定问题），建议使用UTC时间格式，因为时区问题可能导致查询失败
其他非常用字段：略

ES查询

​ ES官方提供了两中检索方式:一种是通过 URL 参数进行搜索,另一种是通过 DSL(Domain Specified Language) 进行搜索。官方更推荐使用第二种方式第二种方式是基于传递JSON作为请求体(request body)格式与ES进行交互，这种方式更强大，更简洁。

ES数据同步方案

1、业务层同步

​ 在业务层完成数据操作提交事务后，同步数据至ES，常见的做法是在ORM的hooks钩子里编写监听相关表的增删改操作落表记录，发起延时调度触发ES同步

优点：系统自身形成闭环，不依赖新的组件

缺点：ES不支持事务

风险点：从0到1开发，许多技术点待落实

2、基于binlog的日志订阅

介绍：模拟mysql原生主从模式，server端dump binlog并持久化到本地，即使源库down机，client端依然可以从server端正常接收已被持久化的binlog
优点：通过解析mysql的binlog同步数据的变更，适合实时性要求较高的场景
风险点：运维DB交互耗时
Binlogcenter(公司自研组件)：原理类似阿里的Canal

MQ消费方案

3、定时任务根据last_update_time字段同步

缺点：数据的更新存在一定的延迟，数据的删除无法同步更新，数据最终达成一致

优点：实现简单粗暴

使用ES（6.4.0）

Spring-Data接入方式使用demo

1、创建demo实体

@Document(indexName = "user-item",createIndex = false)
@Setting(settingPath = "elasticsearch/settings.json")
@Data
@Accessors(chain = true)
public class EsUserItem{

    @ApiModelProperty(value = "用户管理事项关系(获取详情的id)id")
    @Id
    @Field(type = FieldType.Keyword)
    private String itemUserId;

    @ApiModelProperty(value = "姓名")
    @Field(type = FieldType.Text)
    private String userName;

    @ApiModelProperty(value = "管理事项")
    @Field(type = FieldType.Text)
    private String content;


    @ApiModelProperty(value = "执行状态")
    @Field(type = FieldType.Integer)
    private Integer performState;

    @ApiModelProperty("执行开始日期")
    @Field(type = FieldType.Date, format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS")
    private Date performStartDate;

    @ApiModelProperty("执行结束日期)")
    @Field(type = FieldType.Date, format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss")
    private Date performEndDate;

}

2、创建索引+设置映射+查询语句（DSL）

PUT /mss_user_item
{
  "mappings": {
    "UserItem": {
      "properties": {
        "itemUserId": {
          "type": "keyword"
        },
        "userName": {
          "type": "keyword"
        },
        "content": {
          "type": "text"
        },
        "performState": {
          "type": "integer"
        },
        "performStartDate": {
          "type": "date",
          "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS||epoch_millis"
        },
        "performEndDate": {
          "type": "date",
          "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd HH:mm:ss.SSS||epoch_millis"
        }
      }
    }
  }
}

GET mss-user-item/UserItem/_search?
{
  "query": {
    "bool": {
      "must": [
        {
          "terms": {
            "performState": [
              2,
              3,
              4,
              5,
              6,
              1
            ],
            "boost": 1
          }
        },
        {
          "terms": {
            "userName": [
              "李1",
              "李2",
              "李3",
              "李4",
              "李5",
              "李6"
            ],
            "boost": 1
          }
        },
        {
          "prefix": {
            "userName": {
              "value": "李"
            }
          }
        },
        {
         "wildcard": {
           "userName": {
             "value": "李*"
           }
         }
        }
      ],
      "filter": {
        "range": {
          "performStartDate": {
          
            "lte": "2022-01-06 19:26:18"
          }
        }
      }
    }
  },
  "sort": [
    {
      "itemUserId": {
        "order": "asc"
      }
    }
  ]
}

索引文档的简单地增删改查–使用spring-data面向对象编程

public interface ItemRepository extends ElasticsearchRepository<UserItem,Long> {
}

 public void testInsertById(String indexName) throws IOException{
        UserItem userItem = new UserItem().setItemUserId(1).setUserName("李").setContent("lallala")
                .setPerformEndDate(new Date())
                .setPerformStartDate(new Date()).setPerformState(1);
        UserItem save = itemRepository.save(userItem);
        UserItem UserItem = itemRepository.findById("1").orElse(null);
        log.info(UserItem.toString());
    }

XXX列表页查询接入使用

查询入参

{
    "deptId": [
        4102,
        4107
    ],
    "gradeOrRoleId": [
        498,
        488
    ],
    "leaderUserId": [
        28,
        32
    ],
    "pageIndex": 1,
    "pageSize": 20,
    "performState": [
        10,
        5,
        15,
        20,
        30,
        35
    ],
    "performTimeEnd": "2022-01-01",
    "performTimeStart": "2021-12-01",
    "userId": 28,
    "employeeId": "",
    "isPersonMark": 0,
    "isWorkMark": 0,
    "ehrEmployeeId": "111111"
}

查询SQL：略

根据前端查询检索入参和原SQL查询条件得到索引类型结构

1、梳理用户点检项列表的索引字段

itemUserId --用户点检id
ogDeptId --部门id
leaderUserId --上级id 
manage_grade_id --管理职级id
is_exist_personal_remark --是否存在个人备注
is_exist_work_remark  --是否存在工作备注
execute_role_id --执行角色id
perform_start_time --执行开始时间
perform_end_time --执行结束时间
perform_state --点检项状态

RestHighLevelClient接入&封装

​ Java 高级 REST 客户端现在是 Elasticsearch 的默认客户端，它提供了对 TransportClient 的直接替代，因为它接受并返回完全相同的请求/响应对象，因此依赖于 Elasticsearch 核心项目。异步调用在客户端管理的线程池上进行操作，并且需要在请求完成时通知回调，总之SpringBoot项目使用它不会有那么多垃圾兼容问题

1、说在前面

1、因为博主公司对日志输出有限制，只有info级别以上的日志才会上传Kibana，所以需要对ES的日志打印进行调整

2、RestHighLevelClient没有提供链式调用的封装方法和字段名设置解耦（像spring-data-es封装），所以为了后续字段修改扩展和方便调用，需要对API进行一定的封装

2、开始引入

1、maven依赖引入

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

2、封装ES Service操作接口

public interface EsIService<T> {
    List<T> search(SearchRequest searchRequest) throws IOException;

    MssRestHighLevelClient getRestHighLevelClient();

    default BaseEsQueryChainWrapper<T> lambdaQuery(){
        BaseEsQueryChainWrapper<T> wrapper = new BaseEsQueryChainWrapper<>();
        wrapper.setIndex(getIndex());
        wrapper.setType(getType());
        wrapper.setClazz(getClazz());
        wrapper.setRestHighLevelClient(getRestHighLevelClient());
        return wrapper;
    }
    /**
     * 获取索引
     * @return
     */
    String getIndex();

    /**
     * 获取类型
     * @return
     */
    String getType();

    Class<T> getClazz();
}

public class EsIServiceImpl<T> implements EsIService<T> {
	//记录T的Class
    private Class<T> clazz;

    @Resource
    private MssRestHighLevelClient mssRestHighLevelClient;

    public EsIServiceImpl() {
      clazz = (Class)((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
    }

    @Override
    public List<T> doSearch(SearchRequest searchRequest) throws IOException {

        SearchResponse search = mssRestHighLevelClient.search(searchRequest);
        List<T> list = Stream.of(search.getHits().getHits()).map(e -> new JSONObject(e.getSourceAsMap()).toJavaObject(clazz)).collect(Collectors.toList());
        return list;
    }

    @Override
    public MssRestHighLevelClient getRestHighLevelClient() {
        return mssRestHighLevelClient;
    }

    @Override
    public String getIndex() {
        return "mss-user-item";
    }

    @Override
    public String getType() {
        return "mssesuseritem";
    }

    @Override
    public Class<T> getClazz() {
        return clazz;
    }
}

3、封装Lambda调用(举例封装must、should、filter)

@FunctionalInterface
public interface ColumnFunction<T, R> extends Function<T, R>, Serializable{

}
//根据传进来的get方法获取字段属性名
public class ColumnUtil{

    public static final String GET = "get";

    public static String getColumn(ColumnFunction<?,?> fn){
        Method writeReplaceMethod;
        try {
            writeReplaceMethod = fn.getClass().getDeclaredMethod("writeReplace");
        } catch (Exception e) {
            throw new MssBizException("es入参序列化失败,e",e);
        }

        ReflectionUtils.makeAccessible(writeReplaceMethod);
        SerializedLambda serializedLambda;
        try {
            serializedLambda = (SerializedLambda) writeReplaceMethod.invoke(fn);
        } catch (Exception e) {
            throw new MssBizException("es入参序列化失败,e",e);
        }

        // 从lambda信息取出methodName
        String implMethodName = serializedLambda.getImplMethodName();
        // 确保方法是符合规范的get方法
        if (!implMethodName.startsWith(GET)) {
            throw new MssBizException("只允许传入get方法: " + implMethodName);
        }

        String fieldName = implMethodName.substring(GET.length());
        String firstChar = fieldName.substring(0, 1);
        fieldName = fieldName.replaceFirst(firstChar, firstChar.toLowerCase());
        return fieldName;
    }
}
//链式调用构造SearchRequest
public class BaseEsQueryChainWrapper<T>{
    protected SearchSourceBuilder builder = new SearchSourceBuilder();
    protected MssRestHighLevelClient restHighLevelClient;

    protected String index;
    public Class<T> clazz;

    protected String type;

    public void setClazz(Class<T> clazz) {
        this.clazz = clazz;
    }

    public String getColumn(ColumnFunction<T,?> fn){
        String column = ColumnUtil.getColumn(fn);
        return column;
    }
    public void setRestHighLevelClient(MssRestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }
    public String getIndex() {
        return index;
    }
    public String getType() {
        return type;
    }
    public void setIndex(String index) {
        this.index = index;
    }
    public void setType(String type) {
        this.type = type;
    }
    public BaseEsQueryChainWrapper() {
        builder.query(QueryBuilders.boolQuery());
    }
    public BaseEsQueryChainWrapper<T> mustEq(ColumnFunction<T, ?> column , Object value){
        BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder) builder.query();
        boolQueryBuilder.must(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }
    public BaseEsQueryChainWrapper<T> filterEq(ColumnFunction<T, ?> column , Object value){
        BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder) builder.query();
        boolQueryBuilder.filter(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }
    public BaseEsQueryChainWrapper<T> shouldEq(ColumnFunction<T, ?> column , Object value){
        BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder) builder.query();
        boolQueryBuilder.should(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }

    public BaseEsQueryChainWrapper<T> nestedMustEq(BoolQueryBuilder nestedQueryBuilder){
        BoolQueryBuilder boolQueryBuilder = (BoolQueryBuilder) builder.query();
        boolQueryBuilder.must(nestedQueryBuilder);
        return this;
    }

    public SearchRequest getSearchRequest() throws IOException {
        SearchRequest request = new SearchRequest(index);
        request.types(type);
        request.source(builder);
        return request;
    }

}
//嵌套链式调用
public class BaseNestedEsQueryChainWrapper<T>{
    private BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); ;

    public Class<T> clazz;


    public void setClazz(Class<T> clazz) {
        this.clazz = clazz;
    }

    public String getColumn(ColumnFunction<T,?> fn){
        String column = ColumnUtil.getColumn(fn);
        return column;
    }

    public BaseNestedEsQueryChainWrapper<T> mustEq(ColumnFunction<T, ?> column , Object value){
        boolQueryBuilder.must(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }

    public BaseNestedEsQueryChainWrapper<T> filterEq(ColumnFunction<T, ?> column , Object value){
        boolQueryBuilder.filter(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }

    public BaseNestedEsQueryChainWrapper<T> shouldEq(ColumnFunction<T, ?> column , Object value){
        boolQueryBuilder.should(QueryBuilders.termQuery(getColumn(column),value));
        return this;
    }

    public BoolQueryBuilder getBoolQueryBuilder() {
        return boolQueryBuilder;
    }
}

4、日志切面

@Slf4j
@Aspect
@Component
@Order
@Profile({"local", "dev", "test", "ys"})
public class EsLogAspect {

    private final static Integer PRINT_LIMIT = 10;

    //切EsIServiceImpl的do开头的方法
    @Pointcut("execution(public * *.EsIServiceImpl.do*(..))")
    public void pointcut() {}

    @Around("pointcut()")
    public Object around(ProceedingJoinPoint point) {
        // 获取切入点的方法参数
        Object[] args = point.getArgs();
        for (Object arg : args) {
            if (arg instanceof SearchRequest) {
                SearchRequest searchRequest = (SearchRequest) arg;
                log.info("ES查询参数：{}", searchRequest);
            }
        }

        Object ret = null;
        try {
            long start = System.currentTimeMillis();
            ret = point.proceed();
            long end = System.currentTimeMillis();
            if (ret instanceof SearchResponse) {
                SearchResponse searchResponse = (SearchResponse) ret;
                Aggregations aggs = searchResponse.getAggregations();
                SearchHit[] hits = searchResponse.getHits().getHits();
                // 防止查询出来的记录数太多打印不了，最多打印10条
                int toIndex = hits.length > PRINT_LIMIT? PRINT_LIMIT:  hits.length;
                List<SearchHit> hitList = Arrays.stream(hits).limit(toIndex).collect(Collectors.toList());
                String data = hitList.stream().map(SearchHit::getSourceAsString).collect(Collectors.joining(","));
                log.info("ES查询耗时：{}ms, ES查询结果：contents={}, 聚合结果：aggregations={}",
                        end - start, data, aggs == null ? "" : JSON.toJSON(aggs.asList()));
            }else{
                log.info("ES查询耗时：{}ms, ES查询结果：contents={}",end - start,ret);
            }
        } catch (Throwable e) {
            log.error("ES请求错误：{}", e);
            throw new MssBizException("ES请求错误");
        }

        return ret;
    }

}

5、测试类

@Slf4j
@Service
public class ItemEsMapper extends EsIServiceImpl<MssEsUserItem> {
    public void test(String index,String type) throws IOException {
        //原生构造
        SearchRequest request = new SearchRequest(index);
        request.types(type);
        SearchSourceBuilder builder = new SearchSourceBuilder();
        BoolQueryBuilder query = QueryBuilders.boolQuery();
        query.filter(QueryBuilders.termsQuery("itemUserId", Arrays.asList(1,2,3,4,5,6,7)));
        query.must(QueryBuilders.prefixQuery("userName", "李"));
        query.must(QueryBuilders.rangeQuery("itemUserId").lte(9).gte(1));
        builder.query(query);
        builder.from(0).size(10);
        builder.sort("itemUserId");
        builder.aggregation(AggregationBuilders.count("userName").field("userName"));
        request.source(builder);
        List<MssEsUserItem> search = itemEsMapper.doSearch(request);
        //链式构造
        BoolQueryBuilder nested = itemEsMapper.lambdaNestedQuery().shouldEq(MssEsUserItem::getUserName, "李1").shouldEq(MssEsUserItem::getUserName, "李2").getBoolQueryBuilder();
        SearchRequest searchRequest = itemEsMapper.lambdaQuery()
                .mustEq(MssEsUserItem::getItemUserId, 1L).nestedMustEq(nested).getSearchRequest();
        List<MssEsUserItem> list = itemEsMapper.doSearch(searchRequest);

    }
    
}

//测试接口
@PostMapping(value = "/testQuery")
@Menu(MenuConfig.DATA_REPAIR)
@ApiOperation(value = "测试查询", notes = " 测试查询")
public Response<PageResult<MssEsUserItem>> testQuery(@RequestBody VisitTagDTO visitTagDTO) throws IOException {
    esTestService.test("mss-user-item","mssesuseritem");
    return RestResponse.buildSuccessInfo(null);
}

注意事项

1、单次查询数量 form+size不大于5k

假如每页是 10 条数据，现在要查询第 100 页的10条数据，实际上是会把每个 shard 上存储的前 1000 条数据都查到一个协调节点上。如果有个 5 个 shard，那么就有 5000 条数据，接着协调节点对这 5000 条数据进行一些合并、处理，再获取到最终第 100 页的 10 条数据。

ES作为分布式程序，要查第 100 页的 10 条数据，不可能说从 5 个 shard，每个 shard 就查 2 条数据，最后到协调节点合并成 10 条数据，必须得从每个 shard 都查 1000 条数据过来，然后根据的需求进行排序、筛选等等操作，最后再次分页，拿到里面第 100 页的数据。翻页的时候，翻的越深，每个 shard 返回的数据就越多，而且协调节点处理的时间越长，所以用 es 做分页的时候，会发现越翻到后面，就越是慢。

解决方案：

1. 不允许深度分页。（目前使用的是这种方法）

2. scroll api

   scroll 会一次性给你生成所有数据的一个快照，然后每次滑动向后翻页就是通过游标 scroll_id移动，获取下一页下一页这样子，性能会比上面说的那种分页性能要高很多很多，基本上都是毫秒级的。但是这一种不能随意跳到任何一页。

3. search_after

   search_after 的思想是使用前一页的结果来帮助检索下一页的数据，显然，这种方式也不允许你随意翻页，你只能一页页往后翻。初始化时，需要使用一个唯一值的字段作为 sort 字段。

2、禁用 wildcard（keyword前后模糊必须使用）

3、充分利用倒排索引机制，能 keyword 类型尽量 keyword

4、尽可能使用filter而不是query，但是要避免filter cache过大导致jvm飙升

5、做好索引重建的准备（起别名）

6、 java使用es的客户端时，不要使用TransportClient sdk啦。直接使用high level rest client以及low level rest client就好了。（如果不看文档，很多人直接选择 TransportClient ，里面好多query dsl的java类，刚刚接触可能选择这个了，但，这个7.0时就被废弃了；8.0之后直接就删除了）