分布式搜索引擎ES

这篇具有很好参考价值的文章主要介绍了分布式搜索引擎ES。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

初识elasticsearch

了解ES

elasticsearch的作用

elasticsearch是一款非常强大的开源搜索引擎,具备非常多强大功能,可以帮助我们从海量数据中快速找到需要的内容

例如:

  • 在GitHub搜索代码
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
  • 在电商网站搜索商品
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
  • 在百度搜索答案
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

ELK技术栈
elasticsearch结合kibana、Logstash、Beats,也就是elastic stack(ELK)。被广泛应用在日志数据分析、实时监控等领域:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
而elasticsearch是elastic stack的核心,负责存储、搜索、分析数据
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


elasticsearch和lucene
elasticsearch底层是基于lucene来实现的

Lucene是一个Java语言的搜索引擎类库,是Apache公司的顶级项目,由DougCutting于1999年研发
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
elasticsearch的发展历史:

  • 2004年Shay Banon基于Lucene开发了Compass
  • 2010年Shay Banon 重写了Compass,取名为Elasticsearch
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

为什么不是其他搜索技术?
目前比较知名的搜索引擎技术排名:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
虽然在早期,Apache Solr是最主要的搜索引擎技术,但随着发展elasticsearch已经渐渐超越了Solr,独占鳌头:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

倒排索引

倒排索引的概念是基于MySQL这样的正向索引而言的

正向索引

那么什么是正向索引呢?例如给下表(tb_goods)中的id创建索引:
如果是根据id查询,那么直接走索引,查询速度非常快

但如果是基于title做模糊查询,只能是逐行扫描数据,流程如下:

  • 用户搜索数据,条件是title符合"%手机%"

  • 逐行获取数据,比如id为1的数据

  • 判断数据中的title是否符合用户搜索条件

  • 如果符合则放入结果集,不符合则丢弃。回到步骤1

逐行扫描,也就是全表扫描,随着数据量增加,其查询效率也会越来越低。当数据量达到数百万时,就是一场灾难
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

倒排索引

倒排索引中有两个非常重要的概念:

  • 文档(Document):用来搜索的数据,其中的每一条数据就是一个文档。例如一个网页、一个商品信息
  • 词条(Term):对文档数据或用户搜索数据,利用某种算法分词,得到的具备含义的词语就是词条。例如:我是中国人,就可以分为:我、是、中国人、中国、国人这样的几个词条

创建倒排索引是对正向索引的一种特殊处理,流程如下:

  • 将每一个文档的数据利用算法分词,得到一个个词条
  • 创建表,每行数据包括词条、词条所在文档id、位置等信息
  • 因为词条唯一性,可以给词条创建索引,例如hash表结构索引

如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
倒排索引的搜索流程如下(以搜索"华为手机"为例):

  • 用户输入条件"华为手机"进行搜索
  • 对用户输入内容分词,得到词条:华为手机
  • 拿着词条在倒排索引中查找,可以得到包含词条的文档id:1、2、3
  • 拿着文档id到正向索引中查找具体文档

如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
虽然要先查询倒排索引,再查询倒排索引,但是无论是词条、还是文档id都建立了索引,查询速度非常快!无需全表扫描

正向和倒排

那么为什么一个叫做正向索引,一个叫做倒排索引呢?

  • 正向索引是最传统的,根据id索引的方式。但根据词条查询时,必须先逐条获取每个文档,然后判断文档中是否包含所需要的词条,是根据文档找词条的过程

  • 倒排索引则相反,是先找到用户要搜索的词条,根据词条得到保护词条的文档的id,然后根据id获取文档。是根据词条找文档的过程

是不是恰好反过来了?

那么两者方式的优缺点是什么呢?

正向索引

  • 优点:
    • 可以给多个字段创建索引
    • 根据索引字段搜索、排序速度非常快
  • 缺点:
    • 根据非索引字段,或者索引字段中的部分词条查找时,只能全表扫描。

倒排索引

  • 优点:
    • 根据词条搜索、模糊搜索时,速度非常快
  • 缺点:
    • 只能给词条创建索引,而不是字段
    • 无法根据字段做排序

es的一些概念

elasticsearch中有很多独有的概念,与mysql中略有差别,但也有相似之处

文档和字段

elasticsearch是面向 文档(Document) 存储的,可以是数据库中的一条商品数据,一个订单信息。文档数据会被序列化为json格式后存储在elasticsearch中:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
而Json文档中往往包含很多的字段(Field),类似于数据库中的列

索引和映射

索引(Index):相同类型的文档的集合
映射(mapping):索引中文档的字段约束信息,类似表的结构约束

例如:

  • 所有用户文档,就可以组织在一起,称为用户的索引;
  • 所有商品的文档,可以组织在一起,称为商品的索引;
  • 所有订单的文档,可以组织在一起,称为订单的索引;
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

因此,我们可以把索引当做是数据库中的表

数据库的表会有约束信息,用来定义表的结构、字段的名称、类型等信息。因此,索引库中就有映射(mapping),是索引中文档的字段约束信息,类似表的结构约束

mysql与elasticsearch

我们统一的把mysql与elasticsearch的概念做一下对比:

MySQL Elasticsearch 说明
Table Index 索引(index),就是文档的集合,类似数据库的表(table)
Row Document 文档(Document),就是一条条的数据,类似数据库中的行(Row),文档都是JSON格式
Column Field 字段(Field),就是JSON文档中的字段,类似数据库中的列(Column)
Schema Mapping Mapping(映射)是索引中文档的约束,例如字段类型约束。类似数据库的表结构(Schema)
SQL DSL DSL是elasticsearch提供的JSON风格的请求语句,用来操作elasticsearch,实现CRUD

是不是说,我们学习了elasticsearch就不再需要mysql了呢?

并不是如此,两者各自有自己的擅长支出:

  • Mysql:擅长事务类型操作,可以确保数据的安全和一致性

  • Elasticsearch:擅长海量数据的搜索、分析、计算

因此在企业中,往往是两者结合使用:

  • 对安全性要求较高的写操作,使用mysql实现
  • 对查询性能要求较高的搜索需求,使用elasticsearch实现
  • 两者再基于某种方式,实现数据的同步,保证一致性
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

安装ES

  • 创建网络
    因为我们还需要部署kibana容器,因此需要让es和kibana容器互联。这里先创建一个网络:docker network create es-net

  • 加载镜像
    由于elasticsearch镜像比较大,不建议直接pull,所以我们从官网下载,将其上传到虚拟机中,然后运行命令加载即可:docker load -i es.tar
    同理还有kibana的tar包也需要这样做

  • 运行docker命令,部署单点es:

    docker run -d \
    	--name es \
        -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
        -e "discovery.type=single-node" \
        -v es-data:/usr/share/elasticsearch/data \
        -v es-plugins:/usr/share/elasticsearch/plugins \
        --privileged \
        --network es-net \
        -p 9200:9200 \
        -p 9300:9300 \
    elasticsearch:7.12.1
    

    命令解释:

    • -e "cluster.name=es-docker-cluster":设置集群名称
    • -e "http.host=0.0.0.0":监听的地址,可以外网访问
    • -e "ES_JAVA_OPTS=-Xms512m -Xmx512m":内存大小
    • -e "discovery.type=single-node":非集群模式
    • -v es-data:/usr/share/elasticsearch/data:挂载逻辑卷,绑定es的数据目录
    • -v es-logs:/usr/share/elasticsearch/logs:挂载逻辑卷,绑定es的日志目录
    • -v es-plugins:/usr/share/elasticsearch/plugins:挂载逻辑卷,绑定es的插件目录
    • --privileged:授予逻辑卷访问权
    • --network es-net :加入一个名为es-net的网络中
    • -p 9200:9200:端口映射配置
  • 在浏览器中输入:http://192.168.1.12:9200 即可看到elasticsearch的响应结果:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

部署kibana

kibana可以给我们提供一个elasticsearch的可视化界面,便于我们学习

部署

  • 运行docker命令,部署kibana

    docker run -d \
    --name kibana \
    -e ELASTICSEARCH_HOSTS=http://es:9200 \
    --network=es-net \
    -p 5601:5601  \
    kibana:7.12.1
    
    • --network es-net :加入一个名为es-net的网络中,与elasticsearch在同一个网络中
    • -e ELASTICSEARCH_HOSTS=http://es:9200":设置elasticsearch的地址,因为kibana已经与elasticsearch在一个网络,因此可以用容器名直接访问elasticsearch
    • -p 5601:5601:端口映射配置

    kibana启动一般比较慢,需要多等待一会,可以通过命令:docker logs -f kibana 查看运行日志,当查看到下面的日志,说明成功:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

安装IK分词器

  • 在线安装ik插件(较慢)

    # 进入容器内部
    docker exec -it elasticsearch /bin/bash
    
    # 在线下载并安装
    ./bin/elasticsearch-plugin  install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.12.1/elasticsearch-analysis-ik-7.12.1.zip
    
    #退出
    exit
    #重启容器
    docker restart elasticsearch
    
  • 离线安装ik插件(推荐)

    • 查看数据卷目录
      安装插件需要知道elasticsearch的plugins目录位置,而我们用了数据卷挂载,因此需要查看elasticsearch的数据卷目录,通过下面命令查看:docker volume inspect es-plugins

    • 下载解压缩分词器安装包 ,重命名为ik
      分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

    • 上传到es容器的插件数据卷中:也就是/var/lib/docker/volumes/es-plugins/_data

    • 重启容器:docker restart es

  • 测试:
    IK分词器包含两种模式:

    • ik_smart:最少切分

    • ik_max_word:最细切分

    GET /_analyze
    {
      "analyzer": "ik_max_word",
      "text": "小吴在敲Bug"
    }
    

    结果:

    {
      "tokens" : [
        {
          "token" : "小吴",
          "start_offset" : 0,
          "end_offset" : 2,
          "type" : "CN_WORD",
          "position" : 0
        },
        {
          "token" : "在",
          "start_offset" : 2,
          "end_offset" : 3,
          "type" : "CN_CHAR",
          "position" : 1
        },
        {
          "token" : "敲",
          "start_offset" : 3,
          "end_offset" : 4,
          "type" : "CN_CHAR",
          "position" : 2
        },
        {
          "token" : "bug",
          "start_offset" : 4,
          "end_offset" : 7,
          "type" : "ENGLISH",
          "position" : 3
        }
      ]
    }
    

扩展词词典

随着互联网的发展,“造词运动”也越发的频繁。出现了很多新的词语,在原有的词汇列表中并不存在。比如:“奥力给”,“尊嘟假嘟” 等
所以我们的词汇也需要不断的更新,IK分词器提供了扩展词汇的功能

  • 打开IK分词器config目录在IKAnalyzer.cfg.xml配置文件内容添加:

    <?xml version="1.0" encoding="UTF-8"?>
    <!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
    <properties>
            <comment>IK Analyzer 扩展配置</comment>
            <!--用户可以在这里配置自己的扩展字典 *** 添加扩展词典-->
            <entry key="ext_dict">ext.dic</entry>
    </properties>
    
  • 新建一个 ext.dic,可以参考config目录下复制一个配置文件进行修改

    尊嘟假嘟
    奥力给
    
  • 重启elasticsearch : docker restart es

停用词词典

在互联网项目中,在网络间传输的速度很快,所以很多语言是不允许在网络上传递的,如:关于宗教、政治等敏感词语,那么我们在搜索时也应该忽略当前词汇
IK分词器也提供了强大的停用词功能,让我们在索引时就直接忽略当前的停用词汇表中的内容

  • IKAnalyzer.cfg.xml配置文件内容添加:

    <?xml version="1.0" encoding="UTF-8"?>
    <!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
    <properties>
            <comment>IK Analyzer 扩展配置</comment>
            <!--用户可以在这里配置自己的扩展字典-->
            <entry key="ext_dict">ext.dic</entry>
             <!--用户可以在这里配置自己的扩展停止词字典  *** 添加停用词词典-->
            <entry key="ext_stopwords">stopword.dic</entry>
    </properties>
    
  • 在 stopword.dic 添加停用词

    啊
    哦
    额
    
  • 重启elasticsearch : docker restart es

索引库操作

索引库就类似数据库表,mapping映射就类似表的结构
我们要向es中存储数据,必须先创建“库”和“表”

mapping映射属性

mapping是对索引库中文档的约束,常见的mapping属性包括:

  • type:字段数据类型,常见的简单类型有:
    • 字符串:text(可分词的文本)、keyword(精确值,例如:品牌、国家、ip地址)
    • 数值:long、integer、short、byte、double、float、
    • 布尔:boolean
    • 日期:date
    • 对象:object
  • index:是否创建索引,默认为true
  • analyzer:使用哪种分词器
  • properties:该字段的子字段

例如下面的json文档:

{
    "age": 21,
    "weight": 52.1,
    "isMarried": false,
    "info": "小吴在敲Bug",
    "email": "zy@itcast.cn",
    "score": [99.1, 99.5, 98.9],
    "name": {
        "firstName": "云",
        "lastName": "赵"
    }
}

对应的每个字段映射(mapping):

  • age:类型为 integer;参与搜索,因此需要index为true;无需分词器
  • weight:类型为float;参与搜索,因此需要index为true;无需分词器
  • isMarried:类型为boolean;参与搜索,因此需要index为true;无需分词器
  • info:类型为字符串,需要分词,因此是text;参与搜索,因此需要index为true;分词器可以用ik_smart
  • email:类型为字符串,但是不需要分词,因此是keyword;不参与搜索,因此需要index为false;无需分词器
  • score:虽然是数组,但是我们只看元素的类型,类型为float;参与搜索,因此需要index为true;无需分词器
  • name:类型为object,需要定义多个子属性
    • name.firstName;类型为字符串,但是不需要分词,因此是keyword;参与搜索,因此需要index为true;无需分词器
    • name.lastName;类型为字符串,但是不需要分词,因此是keyword;参与搜索,因此需要index为true;无需分词器

索引库的CRUD

这里我们统一使用Kibana编写DSL的方式来演示

创建索引库和映射

基本语法:

  • 请求方式:PUT
  • 请求路径:/索引库名,可以自定义
  • 请求参数:mapping映射

格式:

PUT /索引库名称
{
  "mappings": {
    "properties": {
      "字段名":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "字段名2":{
        "type": "keyword",
        "index": "false"
      },
      "字段名3":{
        "properties": {
          "子字段": {
            "type": "keyword"
          }
        }
      },
      // ...略
    }
  }
}

查询索引库

基本语法

  • 请求方式:GET

  • 请求路径:/索引库名

  • 请求参数:无

格式GET /索引库名

修改索引库

倒排索引结构虽然不复杂,但是一旦数据结构改变(比如改变了分词器),就需要重新创建倒排索引,这简直是灾难。因此索引库一旦创建,无法修改mapping
虽然无法修改mapping中已有的字段,但是却允许添加新的字段到mapping中,因为不会对倒排索引产生影响

语法说明

PUT /索引库名/_mapping
{
  "properties": {
    "新字段名":{
      "type": "integer"
    }
  }
}

删除索引库

语法:

  • 请求方式:DELETE

  • 请求路径:/索引库名

  • 请求参数:无

格式: DELETE /索引库名

文档操作

  • 新增文档

    POST /索引库名/_doc/文档id
    {
        "字段1": "值1",
        "字段2": "值2",
        "字段3": {
            "子属性1": "值3",
            "子属性2": "值4"
        },
        // ...
    }
    
  • 查询文档:根据rest风格,新增是post,查询应该是get,不过查询一般都需要条件,这里我们把文档id带上
    GET /{索引库名称}/_doc/{id}

  • 删除文档:删除使用DELETE请求,同样,需要根据id进行删除:
    DELETE /{索引库名}/_doc/id值

  • 修改文档
    修改有两种方式:

    • 全量修改:直接覆盖原来的文档

      • 根据指定的id删除文档
      • 新增一个相同id的文档
        注意:如果根据id删除时,id不存在,第二步的新增也会执行,也就从修改变成了新增操作了
      PUT /{索引库名}/_doc/文档id
      {
          "字段1": "值1",
          "字段2": "值2",
          // ... 略
      }
      
    • 增量修改:修改文档中的部分字段
      增量修改是只修改指定id匹配的文档中的部分字段

      POST /{索引库名}/_update/文档id
      {
          "doc": {
               "字段名": "新的值",
          }
      }
      

RestClient操作索引库

ES官方提供了各种不同语言的客户端,用来操作ES。这些客户端的本质就是组装DSL语句,通过http请求发送给ES

其中的Java Rest Client又包括两种:

  • Java Low Level Rest Client
  • Java High Level Rest Client
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    我们学习的是Java High Level Rest Client客户端API

导入Demo工程

  • 导入数据
    首先导入课前资料提供的数据库数据:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    数据结构如下:

    CREATE TABLE `tb_hotel` (
      `id` bigint(20) NOT NULL COMMENT '酒店id',
      `name` varchar(255) NOT NULL COMMENT '酒店名称;例:7天酒店',
      `address` varchar(255) NOT NULL COMMENT '酒店地址;例:航头路',
      `price` int(10) NOT NULL COMMENT '酒店价格;例:329',
      `score` int(2) NOT NULL COMMENT '酒店评分;例:45,就是4.5分',
      `brand` varchar(32) NOT NULL COMMENT '酒店品牌;例:如家',
      `city` varchar(32) NOT NULL COMMENT '所在城市;例:上海',
      `star_name` varchar(16) DEFAULT NULL COMMENT '酒店星级,从低到高分别是:1星到5星,1钻到5钻',
      `business` varchar(255) DEFAULT NULL COMMENT '商圈;例:虹桥',
      `latitude` varchar(32) NOT NULL COMMENT '纬度;例:31.2497',
      `longitude` varchar(32) NOT NULL COMMENT '经度;例:120.3925',
      `pic` varchar(255) DEFAULT NULL COMMENT '酒店图片;例:/img/1.jpg',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
    
  • 导入项目
    然后导入课前资料提供的项目:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    项目结构如图:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

  • mapping映射分析
    创建索引库,最关键的是mapping映射,而mapping映射要考虑的信息包括:

    • 字段名
    • 字段数据类型
    • 是否参与搜索
    • 是否需要分词
    • 如果分词,分词器是什么?

    其中:

    • 字段名、字段数据类型,可以参考数据表结构的名称和类型
    • 是否参与搜索要分析业务来判断,例如图片地址,就无需参与搜索
    • 是否分词呢要看内容,内容如果是一个整体就无需分词,反之则要分词
    • 分词器,我们可以统一使用ik_max_word

    酒店数据的索引库结构:

    PUT /hotel
    {
      "mappings": {
        "properties": {
          "id":{
            "type": "keyword"
          },
          "name":{
            "type": "text",
            "analyzer": "ik_max_word",
            "copy_to": "all"
          },
          "address":{
            "type": "keyword",
            "index": false
          },
          "price":{
            "type": "integer"
          },
          "score":{
            "type": "integer"
          },
          "brand":{
            "type": "keyword",
            "copy_to": "all"
          },
          "city":{
            "type": "keyword"
          },
          "starName":{
            "type": "keyword"
          },
          "business":{
            "type": "keyword",
            "copy_to": "all"
          },
          "location":{
            "type": "geo_point"
          },
          "pic":{
            "type": "keyword",
            "index": false
          },
           "all":{
            "type": "text",
            "analyzer": "ik_max_word"
          }
        }
      }
    }
    

    几个特殊字段说明:

    • location:地理坐标,里面包含精度、纬度
    • all:一个组合字段,其目的是将多字段的值 利用copy_to合并,提供给用户搜索

    地理坐标说明:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    copy_to说明:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

  • 初始化RestClient

    在elasticsearch提供的API中,与elasticsearch一切交互都封装在一个名为RestHighLevelClient的类中,必须先完成这个对象的初始化,建立与elasticsearch的连接

    • 引入es的RestHighLevelClient依赖:

      <dependency>
          <groupId>org.elasticsearch.client</groupId>
          <artifactId>elasticsearch-rest-high-level-client</artifactId>
          <version>7.12.1</version>
      </dependency>
      
    • 因为SpringBoot默认的ES版本是7.6.2,所以我们需要覆盖默认的ES版本:

      <properties>
          <elasticsearch.version>7.12.1</elasticsearch.version>
      </properties>
      
    • 初始化RestHighLevelClient:

      RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
              HttpHost.create("http://192.168.1.12:9200")
      ));
      

      这里为了单元测试方便,我们创建一个测试类HotelIndexTest,然后将初始化的代码编写在@BeforeEach方法中:

      public class HotelIndexTest {
          private RestHighLevelClient client;
      
          @BeforeEach
          void setUp() {
              this.client = new RestHighLevelClient(RestClient.builder(
                      HttpHost.create("http://192.168.150.101:9200")
              ));
          }
      
          @AfterEach
          void tearDown() throws IOException {
              this.client.close();
          }
      }
      

创建索引库

创建索引库的API如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
代码分为三步:

  • 创建Request对象。因为是创建索引库的操作,因此Request是CreateIndexRequest
  • 添加请求参数,其实就是DSL的JSON参数部分。因为json字符串很长,这里是定义了静态字符串常量MAPPING_TEMPLATE,让代码看起来更加优雅
  • 发送请求,client.indices()方法的返回值是IndicesClient类型,封装了所有与索引库操作有关的方法

在hotel-demo的cn.itcast.hotel.constants包下,创建一个类,定义mapping映射的JSON字符串常量:

public class HotelConstants {
    public static final String MAPPING_TEMPLATE = "{\n" +
            "  \"mappings\": {\n" +
            "    \"properties\": {\n" +
            "      \"id\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"name\":{\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"address\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"price\":{\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"score\":{\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"brand\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"city\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"starName\":{\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"business\":{\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"location\":{\n" +
            "        \"type\": \"geo_point\"\n" +
            "      },\n" +
            "      \"pic\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"all\":{\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\"\n" +
            "      }\n" +
            "    }\n" +
            "  }\n" +
            "}";
}

在hotel-demo中的HotelIndexTest测试类中,编写单元测试,实现创建索引:

@Test
void createHotelIndex() throws IOException {
    // 1.创建Request对象
    CreateIndexRequest request = new CreateIndexRequest("hotel");
    // 2.准备请求的参数:DSL语句
    request.source(MAPPING_TEMPLATE, XContentType.JSON);
    // 3.发送请求
    client.indices().create(request, RequestOptions.DEFAULT);
}

删除索引库

删除索引库的DSL语句非常简单:DELETE /hotel
与创建索引库相比:

  • 请求方式从PUT变为DELTE
  • 请求路径不变
  • 无请求参数

所以代码的差异,注意体现在Request对象上。依然是三步走:

  • 创建Request对象。这次是DeleteIndexRequest对象
  • 准备参数。这里是无参
  • 发送请求。改用delete方法

在hotel-demo中的HotelIndexTest测试类中,编写单元测试,实现删除索引:

@Test
void testDeleteHotelIndex() throws IOException {
    // 1.创建Request对象
    DeleteIndexRequest request = new DeleteIndexRequest("hotel");
    // 2.发送请求
    client.indices().delete(request, RequestOptions.DEFAULT);
}

判断索引库是否存在

判断索引库是否存在,本质就是查询,对应的DSL是:GET /hotel
因此与删除的Java代码流程是类似的。依然是三步走:

  • 创建Request对象。这次是GetIndexRequest对象
  • 准备参数。这里是无参
  • 发送请求。改用exists方法
@Test
void testExistsHotelIndex() throws IOException {
    // 1.创建Request对象
    GetIndexRequest request = new GetIndexRequest("hotel");
    // 2.发送请求
    boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
    // 3.输出
    System.err.println(exists ? "索引库已经存在!" : "索引库不存在!");
}

RestClient操作文档

为了与索引库操作分离,我们再次参加一个测试类,做两件事情:

  • 初始化RestHighLevelClient
  • 我们的酒店数据在数据库,需要利用IHotelService去查询,所以注入这个接口
@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;

    private RestHighLevelClient client;

    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.1.12:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

新增文档

我们要将数据库的酒店数据查询出来,写入elasticsearch中

  • 索引库实体类
    数据库查询后的结果是一个Hotel类型的对象。结构如下:

    @Data
    @TableName("tb_hotel")
    public class Hotel {
        @TableId(type = IdType.INPUT)
        private Long id;
        private String name;
        private String address;
        private Integer price;
        private Integer score;
        private String brand;
        private String city;
        private String starName;
        private String business;
        private String longitude;
        private String latitude;
        private String pic;
    }
    

    与我们的索引库结构存在差异:

    • longitude和latitude需要合并为location

    因此,我们需要定义一个新的类型,与索引库结构吻合:

    @Data
    @NoArgsConstructor
    public class HotelDoc {
        private Long id;
        private String name;
        private String address;
        private Integer price;
        private Integer score;
        private String brand;
        private String city;
        private String starName;
        private String business;
        private String location;
        private String pic;
    
        public HotelDoc(Hotel hotel) {
            this.id = hotel.getId();
            this.name = hotel.getName();
            this.address = hotel.getAddress();
            this.price = hotel.getPrice();
            this.score = hotel.getScore();
            this.brand = hotel.getBrand();
            this.city = hotel.getCity();
            this.starName = hotel.getStarName();
            this.business = hotel.getBusiness();
            this.location = hotel.getLatitude() + ", " + hotel.getLongitude();
            this.pic = hotel.getPic();
        }
    }
    
  • 语法说明
    新增文档的DSL语句如下:

    POST /{索引库名}/_doc/1
    {
        "name": "Jack",
        "age": 21
    }
    

    对应的java代码如图:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    可以看到与创建索引库类似,同样是三步走:

    • 创建Request对象
    • 准备请求参数,也就是DSL中的JSON文档
    • 发送请求

    变化的地方在于,这里直接使用client.xxx()的API,不再需要client.indices()了

  • 完整代码
    我们导入酒店数据,基本流程一致,但是需要考虑几点变化:

    • 酒店数据来自于数据库,我们需要先查询出来,得到hotel对象
    • hotel对象需要转为HotelDoc对象
    • HotelDoc需要序列化为json格式

    因此,代码整体步骤如下:

    • 根据id查询酒店数据Hotel
    • 将Hotel封装为HotelDoc
    • 将HotelDoc序列化为JSON
    • 创建IndexRequest,指定索引库名和id
    • 准备请求参数,也就是JSON文档
    • 发送请求

    在hotel-demo的HotelDocumentTest测试类中,编写单元测试:

    @Test
    void testAddDocument() throws IOException {
        // 1.根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        // 2.转换为文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);
        // 3.将HotelDoc转json
        String json = JSON.toJSONString(hotelDoc);
    
        // 1.准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());
        // 2.准备Json文档
        request.source(json, XContentType.JSON);
        // 3.发送请求
        client.index(request, RequestOptions.DEFAULT);
    }
    

查询文档

查询的DSL语句如下:GET /hotel/_doc/{id}
非常简单,因此代码大概分两步:

  • 准备Request对象
  • 发送请求

不过查询的目的是得到结果,解析为HotelDoc,因此难点是结果的解析。完整代码如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
可以看到,结果是一个JSON,其中文档放在一个_source属性中,因此解析就是拿到_source,反序列化为Java对象即可。

与之前类似,也是三步走:

  • 准备Request对象。这次是查询,所以是GetRequest
  • 发送请求,得到结果。因为是查询,这里调用client.get()方法
  • 解析结果,就是对JSON做反序列化

完整代码

@Test
void testGetDocumentById() throws IOException {
    // 1.准备Request
    GetRequest request = new GetRequest("hotel", "61082");
    // 2.发送请求,得到响应
    GetResponse response = client.get(request, RequestOptions.DEFAULT);
    // 3.解析响应结果
    String json = response.getSourceAsString();

    HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
    System.out.println(hotelDoc);
}

修改文档

修改我们讲过两种方式:

  • 全量修改:本质是先根据id删除,再新增
  • 增量修改:修改文档中的指定字段值

在RestClient的API中,全量修改与新增的API完全一致,判断依据是ID:

  • 如果新增时,ID已经存在,则修改
  • 如果新增时,ID不存在,则新增

这里不再赘述,我们主要关注增量修改。

代码示例如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
与之前类似,也是三步走:

  • 准备Request对象。这次是修改,所以是UpdateRequest
  • 准备参数。也就是JSON文档,里面包含要修改的字段
  • 更新文档。这里调用client.update()方法

完整代码
在hotel-demo的HotelDocumentTest测试类中,编写单元测试:

@Test
void testUpdateDocument() throws IOException {
    // 1.准备Request
    UpdateRequest request = new UpdateRequest("hotel", "61083");
    // 2.准备请求参数
    request.doc(
        "price", "952",
        "starName", "四钻"
    );
    // 3.发送请求
    client.update(request, RequestOptions.DEFAULT);
}

删除文档

删除的DSL为是这样的:DELETE /hotel/_doc/{id}

与查询相比,仅仅是请求方式从DELETE变成GET,可以想象Java代码应该依然是三步走:

  • 准备Request对象,因为是删除,这次是DeleteRequest对象。要指定索引库名和id
  • 准备参数,无参
  • 发送请求。因为是删除,所以是client.delete()方法

完整代码
在hotel-demo的HotelDocumentTest测试类中,编写单元测试:

@Test
void testDeleteDocument() throws IOException {
    // 1.准备Request
    DeleteRequest request = new DeleteRequest("hotel", "61083");
    // 2.发送请求
    client.delete(request, RequestOptions.DEFAULT);
}

批量导入文档

案例需求:利用BulkRequest批量将数据库数据导入到索引库中。

步骤如下:

  • 利用mybatis-plus查询酒店数据

  • 将查询到的酒店数据(Hotel)转换为文档类型数据(HotelDoc)

  • 利用JavaRestClient中的BulkRequest批处理,实现批量新增文档

批量处理BulkRequest,其本质就是将多个普通的CRUD请求组合在一起发送
其中提供了一个add方法,用来添加其他请求:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
可以看到,能添加的请求包括:

  • IndexRequest,也就是新增
  • UpdateRequest,也就是修改
  • DeleteRequest,也就是删除

因此Bulk中添加了多个IndexRequest,就是批量新增功能了。示例:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
其实还是三步走:

  • 创建Request对象。这里是BulkRequest
  • 准备参数。批处理的参数,就是其它Request对象,这里就是多个IndexRequest
  • 发起请求。这里是批处理,调用的方法为client.bulk()方法

完整代码
我们在导入酒店数据时,将上述代码改造成for循环处理即可
在hotel-demo的HotelDocumentTest测试类中,编写单元测试:

@Test
void testBulkRequest() throws IOException {
    // 批量查询酒店数据
    List<Hotel> hotels = hotelService.list();

    // 1.创建Request
    BulkRequest request = new BulkRequest();
    // 2.准备参数,添加多个新增的Request
    for (Hotel hotel : hotels) {
        // 2.1.转换为文档类型HotelDoc
        HotelDoc hotelDoc = new HotelDoc(hotel);
        // 2.2.创建新增文档的Request对象
        request.add(new IndexRequest("hotel")
                    .id(hotelDoc.getId().toString())
                    .source(JSON.toJSONString(hotelDoc), XContentType.JSON));
    }
    // 3.发送请求
    client.bulk(request, RequestOptions.DEFAULT);
}

DSL查询文档

elasticsearch的查询依然是基于JSON风格的DSL来实现的

DSL查询分类

Elasticsearch提供了基于JSON的DSL(Domain Specific Language)来定义查询。常见的查询类型包括:

  • 查询所有:查询出所有数据,一般测试用。例如:match_all

  • 全文检索(full text)查询:利用分词器对用户输入内容分词,然后去倒排索引库中匹配。例如:

    • match_query
    • multi_match_query
  • 精确查询:根据精确词条值查找数据,一般是查找keyword、数值、日期、boolean等类型字段。例如:

    • ids
    • range
    • term
  • 地理(geo)查询:根据经纬度查询。例如:

    • geo_distance
    • geo_bounding_box
  • 复合(compound)查询:复合查询可以将上述各种查询条件组合起来,合并查询条件。例如:

    • bool
    • function_score

查询的语法基本一致:

GET /indexName/_search
{
  "query": {
    "查询类型": {
      "查询条件": "条件值"
    }
  }
}

我们以查询所有为例,其中:

  • 查询类型为match_all
  • 没有查询条件
// 查询所有
GET /indexName/_search
{
  "query": {
    "match_all": {
    }
  }
}

其它查询无非就是查询类型查询条件的变化

全文检索查询

使用场景
全文检索查询的基本流程如下:

  • 对用户搜索的内容做分词,得到词条
  • 根据词条去倒排索引库中匹配,得到文档id
  • 根据文档id找到文档,返回给用户

比较常用的场景包括:

  • 商城的输入框搜索
  • 百度输入框搜索

例如京东:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
因为是拿着词条去匹配,因此参与搜索的字段也必须是可分词的text类型的字段


基本语法
常见的全文检索查询包括:

  • match查询:单字段查询
  • multi_match查询:多字段查询,任意一个字段符合条件就算符合查询条件

match查询语法如下:

GET /indexName/_search
{
  "query": {
    "match": {
      "字段名": "查询的关键字"
    }
  }
}

mulit_match语法如下:

GET /indexName/_search
{
  "query": {
    "multi_match": {
      "query": "查询的关键字",
      "fields": ["字段名1", "字段名2"]
    }
  }
}

精准查询

精确查询一般是查找keyword、数值、日期、boolean等类型字段。所以不会对搜索条件分词。常见的有:

  • term:根据词条精确值查询
  • range:根据值的范围查询

term查询

因为精确查询的字段搜是不分词的字段,因此查询的条件也必须是不分词的词条。查询时,用户输入的内容跟自动值完全匹配时才认为符合条件。如果用户输入的内容过多,反而搜索不到数据

语法说明:

// term查询
GET /indexName/_search
{
  "query": {
    "term": {
      "字段名": {
        "value": "关键字"
      }
    }
  }
}

range查询

范围查询,一般应用在对数值类型做范围过滤的时候。比如做价格范围过滤

基本语法:

// range查询
GET /indexName/_search
{
  "query": {
    "range": {
      "字段名": {
        "gte": 10, // 这里的gte代表大于等于,gt则代表大于
        "lte": 20 // lte代表小于等于,lt则代表小于
      }
    }
  }
}

理坐标查询

所谓的地理坐标查询,其实就是根据经纬度查询,官方文档

常见的使用场景包括:

  • 携程:搜索我附近的酒店
  • 滴滴:搜索我附近的出租车
  • 微信:搜索我附近的人

矩形范围查询
矩形范围查询,也就是geo_bounding_box查询,查询坐标落在某个矩形范围的所有文档:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
查询时,需要指定矩形的左上右下两个点的坐标,然后画出一个矩形,落在该矩形内的都是符合条件的点

语法如下:

GET /indexName/_search
{
  "query": {
    "geo_bounding_box": {
      "字段名": {
        "top_left": { // 左上点
          "lat": 31.1,
          "lon": 121.5
        },
        "bottom_right": { // 右下点
          "lat": 30.9,
          "lon": 121.7
        }
      }
    }
  }
}

附近查询
附近查询,也叫做距离查询(geo_distance):查询到指定中心点小于某个距离值的所有文档

换句话来说,在地图上找一个点作为圆心,以指定距离为半径,画一个圆,落在圆内的坐标都算符合条件:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
语法说明:

GET /indexName/_search
{
  "query": {
    "geo_distance": {
      "distance": "15km", // 半径
      "字段名": "31.21,121.5" // 圆心
    }
  }
}

复合查询

复合(compound)查询:复合查询可以将其它简单查询组合起来,实现更复杂的搜索逻辑。常见的有两种:

  • fuction score:算分函数查询,可以控制文档相关性算分,控制文档排名
  • bool query:布尔查询,利用逻辑关系组合多个其它的查询,实现复杂搜索

算分函数查询

根据相关度打分是比较合理的需求,但合理的不一定是产品经理需要

以百度为例,你搜索的结果中,并不是相关度越高排名越靠前,而是谁掏的钱多排名就越靠前。如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
要想人为控制相关性算分,就需要利用elasticsearch中的function score 查询了

语法说明
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
function score 查询中包含四部分内容:

  • 原始查询条件:query部分,基于这个条件搜索文档,并且基于BM25算法给文档打分,原始算分(query score)
  • 过滤条件:filter部分,符合该条件的文档才会重新算分
  • 算分函数:符合filter条件的文档要根据这个函数做运算,得到的函数算分(function score),有四种函数
    • weight:函数结果是常量
    • field_value_factor:以文档中的某个字段值作为函数结果
    • random_score:以随机数作为函数结果
    • script_score:自定义算分函数算法
  • 运算模式:算分函数的结果、原始查询的相关性算分,两者之间的运算方式,包括:
    • multiply:相乘
    • replace:用function score替换query score
    • 其它,例如:sum、avg、max、min

function score的运行流程如下:

  • 根据原始条件查询搜索文档,并且计算相关性算分,称为原始算分(query score)
  • 根据过滤条件,过滤文档
  • 符合过滤条件的文档,基于算分函数运算,得到函数算分(function score)
  • 原始算分(query score)和函数算分(function score)基于运算模式做运算,得到最终结果,作为相关性算分。

因此,其中的关键点是:

  • 过滤条件:决定哪些文档的算分被修改
  • 算分函数:决定函数算分的算法
  • 运算模式:决定最终算分结果

示例

需求:给“如家”这个品牌的酒店排名靠前一些

翻译一下这个需求,转换为之前说的四个要点:

  • 原始条件:不确定,可以任意变化
  • 过滤条件:brand = “如家”
  • 算分函数:可以简单粗暴,直接给固定的算分结果,weight
  • 运算模式:比如求和

因此最终的DSL语句如下:

GET /hotel/_search
{
  "query": {
    "function_score": {
      "query": {  .... }, // 原始查询,可以是任意条件
      "functions": [ // 算分函数
        {
          "filter": { // 满足的条件,品牌必须是如家
            "term": {
              "brand": "如家"
            }
          },
          "weight": 2 // 算分权重为2
        }
      ],
      "boost_mode": "sum" // 加权模式,求和
    }
  }
}

布尔查询

布尔查询是一个或多个查询子句的组合,每一个子句就是一个子查询。子查询的组合方式有:

  • must:必须匹配每个子查询,类似“与”
  • should:选择性匹配子查询,类似“或”
  • must_not:必须不匹配,不参与算分,类似“非”
  • filter:必须匹配,不参与算分

比如在搜索酒店时,除了关键字搜索外,我们还可能根据品牌、价格、城市等字段做过滤:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
每一个不同的字段,其查询的条件、方式都不一样,必须是多个不同的查询,而要组合这些查询,就必须用bool查询了。

需要注意的是,搜索时,参与打分的字段越多,查询的性能也越差。因此这种多条件查询时,建议这样做:

  • 搜索框的关键字搜索,是全文检索查询,使用must查询,参与算分
  • 其它过滤条件,采用filter查询。不参与算分

语法示例:

GET /hotel/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"city": "上海" }}
      ],
      "should": [
        {"term": {"brand": "皇冠假日" }},
        {"term": {"brand": "华美达" }}
      ],
      "must_not": [
        { "range": { "price": { "lte": 500 } }}
      ],
      "filter": [
        { "range": {"score": { "gte": 45 } }}
      ]
    }
  }
}

示例
需求:搜索名字包含“如家”,价格不高于400,在坐标31.21,121.5周围10km范围内的酒店。

分析:

  • 名称搜索,属于全文检索查询,应该参与算分。放到must中
  • 价格不高于400,用range查询,属于过滤条件,不参与算分。放到must_not中
  • 周围10km范围内,用geo_distance查询,属于过滤条件,不参与算分。放到filter中

分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

搜索结果处理

搜索的结果可以按照用户指定的方式去处理或展示

排序

elasticsearch默认是根据相关度算分(_score)来排序,但是也支持自定义方式对搜索结果排序。可以排序字段类型有:keyword类型、数值类型、地理坐标类型、日期类型等

普通字段排序
keyword、数值、日期类型排序的语法基本一致
语法

GET /indexName/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "字段名": "desc"  // 排序字段、排序方式ASC、DESC
    }
  ]
}

排序条件是一个数组,也就是可以写多个排序条件。按照声明的顺序,当第一个条件相等时,再按照第二个条件排序,以此类推


地理坐标排序
语法说明

GET /indexName/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "_geo_distance" : {
          "字段名" : "纬度,经度", // 文档中geo_point类型的字段名、目标坐标点
          "order" : "asc", // 排序方式
          "unit" : "km" // 排序的距离单位
      }
    }
  ]
}

这个查询的含义是:

  • 指定一个坐标,作为目标点
  • 计算每一个文档中,指定字段(必须是geo_point类型)的坐标 到目标点的距离是多少
  • 根据距离排序

分页

elasticsearch
默认情况下只返回top10的数据。而如果要查询更多数据就需要修改分页参数了。elasticsearch中通过修改from、size参数来控制要返回的分页结果:

  • from:从第几个文档开始
  • size:总共查询几个文档

类似于mysql中的limit ?, ?

基本的分页

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "from": 0, // 分页开始的位置,默认为0
  "size": 10, // 期望获取的文档总数
  "sort": [
    {"price": "asc"}
  ]
}

深度分页问题
现在,我要查询990~1000的数据,查询逻辑要这么写:

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "from": 990, // 分页开始的位置,默认为0
  "size": 10, // 期望获取的文档总数
  "sort": [
    {"price": "asc"}
  ]
}

这里是查询990开始的数据,也就是 第990~第1000条 数据

不过,elasticsearch内部分页时,必须先查询 0~1000条,然后截取其中的990 ~ 1000的这10条:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
查询TOP1000,如果es是单点模式,这并无太大影响
但是elasticsearch将来一定是集群,例如我集群有5个节点,我要查询TOP1000的数据,并不是每个节点查询200条就可以了

因为节点A的TOP200,在另一个节点可能排到10000名以外了
因此要想获取整个集群的TOP1000,必须先查询出每个节点的TOP1000,汇总结果后,重新排名,重新截取TOP1000
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
那如果我要查询9900~10000的数据呢?是不是要先查询TOP10000呢?那每个节点都要查询10000条?汇总到内存中?
当查询分页深度较大时,汇总数据过多,对内存和CPU会产生非常大的压力,因此elasticsearch会禁止from+ size 超过10000的请求

针对深度分页,ES提供了两种解决方案,官方文档:

  • search after:分页时需要排序,原理是从上一次的排序值开始,查询下一页数据。官方推荐使用的方式。
  • scroll:原理将排序后的文档id形成快照,保存在内存。官方已经不推荐使用。

高亮

高亮原理
什么是高亮显示呢?

我们在百度,京东搜索时,关键字会变成红色,比较醒目,这叫高亮显示:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
高亮显示的实现分为两步:

  • 给文档中的所有关键字都添加一个标签,例如<em>标签
  • 页面给<em>标签编写CSS样式

实现高亮

高亮的语法

GET /hotel/_search
{
  "query": {
    "match": {
      "字段名": "关键字" // 查询条件,高亮一定要使用全文检索查询
    }
  },
  "highlight": {
    "fields": { // 指定要高亮的字段
      "字段名": {
        "pre_tags": "<em>",  // 用来标记高亮字段的前置标签
        "post_tags": "</em>" // 用来标记高亮字段的后置标签
      }
    }
  }
}

注意:

  • 高亮是对关键字高亮,因此搜索条件必须带有关键字,而不能是范围这样的查询。
  • 默认情况下,高亮的字段,必须与搜索指定的字段一致,否则无法高亮
  • 如果要对非搜索字段高亮,则需要添加一个属性:required_field_match=false

RestClient查询文档

文档的查询同样适用昨天学习的 RestHighLevelClient对象,基本步骤包括:

  • 准备Request对象
  • 准备请求参数
  • 发起请求
  • 解析响应

快速入门

发起查询请求
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

代码解读:

  • 第一步,创建SearchRequest对象,指定索引库名

  • 第二步,利用request.source()构建DSL,DSL中可以包含查询、分页、排序、高亮等

    • query():代表查询条件,利用QueryBuilders.matchAllQuery()构建一个match_all查询的DSL
  • 第三步,利用client.search()发送请求,得到响应

    这里关键的API有两个,一个是request.source(),其中包含了查询、排序、分页、高亮等所有功能:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    另一个是QueryBuilders,其中包含match、term、function_score、bool等各种查询:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


解析响应
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
elasticsearch返回的结果是一个JSON字符串,结构包含:

  • hits:命中的结果
    • total:总条数,其中的value是具体的总条数值
    • max_score:所有结果中得分最高的文档的相关性算分
    • hits:搜索结果的文档数组,其中的每个文档都是一个json对象
      • _source:文档中的原始数据,也是json对象

因此,我们解析响应结果,就是逐层解析JSON字符串,流程如下:

  • SearchHits:通过response.getHits()获取,就是JSON中的最外层的hits,代表命中的结果
    • SearchHits#getTotalHits().value:获取总条数信息
    • SearchHits#getHits():获取SearchHit数组,也就是文档数组
      • SearchHit#getSourceAsString():获取文档结果中的_source,也就是原始的json文档数据

match查询

全文检索的match和multi_match查询与match_all的API基本一致。差别是查询条件,也就是query的部分
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

因此,Java代码上的差异主要是request.source().query()中的参数了。同样是利用QueryBuilders提供的方法:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
而结果解析代码则完全一致,可以抽取并共享


完整代码如下:

@Test
void testMatch() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    request.source()
        .query(QueryBuilders.matchQuery("all", "如家"));
    // 3.发送请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    handleResponse(response);

}

精确查询

精确查询主要是两者:

  • term:词条精确匹配
  • range:范围查询

与之前的查询相比,差异同样在查询条件,其它都一样

查询条件构造的API如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

复合查询

布尔查询是用must、must_not、filter等方式组合其它查询,代码示例如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
可以看到,API与其它查询的差别同样是在查询条件的构建,QueryBuilders,结果解析等其他代码完全不变


完整代码如下:

@Test
void testBool() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    // 2.1.准备BooleanQuery
    BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
    // 2.2.添加term
    boolQuery.must(QueryBuilders.termQuery("city", "杭州"));
    // 2.3.添加range
    boolQuery.filter(QueryBuilders.rangeQuery("price").lte(250));

    request.source().query(boolQuery);
    // 3.发送请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    handleResponse(response);

}

排序、分页

搜索结果的排序和分页是与query同级的参数,因此同样是使用request.source()来设置

对应的API如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

完整代码示例:

@Test
void testPageAndSort() throws IOException {
    // 页码,每页大小
    int page = 1, size = 5;

    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    // 2.1.query
    request.source().query(QueryBuilders.matchAllQuery());
    // 2.2.排序 sort
    request.source().sort("price", SortOrder.ASC);
    // 2.3.分页 from、size
    request.source().from((page - 1) * size).size(5);
    // 3.发送请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    handleResponse(response);
}

高亮

高亮的代码与之前代码差异较大,有两点:

  • 查询的DSL:其中除了查询条件,还需要添加高亮条件,同样是与query同级。
  • 结果解析:结果除了要解析_source文档数据,还要解析高亮结果

高亮请求构建
高亮请求的构建API如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
上述代码省略了查询条件部分,但是大家不要忘了:高亮查询必须使用全文检索查询,并且要有搜索关键字,将来才可以对关键字高亮

完整代码如下:

@Test
void testHighlight() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    // 2.1.query
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 2.2.高亮
    request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));
    // 3.发送请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    handleResponse(response);
}

高亮结果解析
高亮的结果与查询的文档结果默认是分离的,并不在一起

因此解析高亮的代码需要额外处理:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
代码解读:

  • 第一步:从结果中获取source。hit.getSourceAsString(),这部分是非高亮结果,json字符串。还需要反序列为HotelDoc对象
  • 第二步:获取高亮结果。hit.getHighlightFields(),返回值是一个Map,key是高亮字段名称,值是HighlightField对象,代表高亮值
  • 第三步:从map中根据高亮字段名称,获取高亮字段值对象HighlightField
  • 第四步:从HighlightField中获取Fragments,并且转为字符串。这部分就是真正的高亮字符串了
  • 第五步:用高亮的结果替换HotelDoc中的非高亮结果

完整代码如下:

private void handleResponse(SearchResponse response) {
    // 4.解析响应
    SearchHits searchHits = response.getHits();
    // 4.1.获取总条数
    long total = searchHits.getTotalHits().value;
    System.out.println("共搜索到" + total + "条数据");
    // 4.2.文档数组
    SearchHit[] hits = searchHits.getHits();
    // 4.3.遍历
    for (SearchHit hit : hits) {
        // 获取文档source
        String json = hit.getSourceAsString();
        // 反序列化
        HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
        // 获取高亮结果
        Map<String, HighlightField> highlightFields = hit.getHighlightFields();
        if (!CollectionUtils.isEmpty(highlightFields)) {
            // 根据字段名获取高亮结果
            HighlightField highlightField = highlightFields.get("name");
            if (highlightField != null) {
                // 获取高亮值
                String name = highlightField.getFragments()[0].string();
                // 覆盖非高亮结果
                hotelDoc.setName(name);
            }
        }
        System.out.println("hotelDoc = " + hotelDoc);
    }
}

数据聚合

聚合(aggregations) 可以让我们极其方便的实现对数据的统计、分析、运算。例如:

  • 什么品牌的手机最受欢迎?
  • 这些手机的平均价格、最高价格、最低价格?
  • 这些手机每月的销售情况如何?

实现这些统计功能的比数据库的sql要方便的多,而且查询速度非常快,可以实现近实时搜索效果。


聚合常见的有三类:

注意: 参加聚合的字段必须是keyword、日期、数值、布尔类型

  • 桶(Bucket) 聚合:用来对文档做分组

    • TermAggregation:按照文档字段值分组,例如按照品牌值分组、按照国家分组
    • Date Histogram:按照日期阶梯分组,例如一周为一组,或者一月为一组
  • 度量(Metric) 聚合:用以计算一些值,比如:最大值、最小值、平均值等

    • Avg:求平均值
    • Max:求最大值
    • Min:求最小值
    • Stats:同时求max、min、avg、sum等
  • 管道(pipeline) 聚合:其它聚合的结果为基础做聚合

DSL实现聚合

现在,我们要统计所有数据中的酒店品牌有几种,其实就是按照品牌对数据分组。此时可以根据酒店品牌的名称做聚合,也就是Bucket聚合


Bucket聚合语法
语法如下:

GET /hotel/_search
{
  "size": 0,  // 设置size为0,结果中不包含文档,只包含聚合结果
  "aggs": { // 定义聚合
    "brandAgg": { //给聚合起个名字
      "terms": { // 聚合的类型,按照品牌值聚合,所以选择term
        "field": "brand", // 参与聚合的字段
        "size": 20 // 希望获取的聚合结果数量
      }
    }
  }
}

结果如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


聚合结果排序

默认情况下,Bucket聚合会统计Bucket内的文档数量,记为_count,并且按照_count降序排序

我们可以指定order属性,自定义聚合的排序方式:

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "order": {
          "_count": "asc" // 按照_count升序排列
        },
        "size": 20
      }
    }
  }
}

限定聚合范围

默认情况下,Bucket聚合是对索引库的所有文档做聚合,但真实场景下,用户会输入搜索条件,因此聚合必须是对搜索结果聚合。那么聚合必须添加限定条件

我们可以限定要聚合的文档范围,只要添加query条件即可:

GET /hotel/_search
{
  "query": {
    "range": {
      "price": {
        "lte": 200 // 只对200元以下的文档聚合
      }
    }
  }, 
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "size": 20
      }
    }
  }
}

这次,聚合得到的品牌明显变少了:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


Metric聚合语法

上面,我们对酒店按照品牌分组,形成了一个个桶。现在我们需要对桶内的酒店做运算,获取每个品牌的用户评分的min、max、avg等值

这就要用到Metric聚合了,例如stat聚合:就可以获取min、max、avg等结果

语法如下:

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": { 
      "terms": { 
        "field": "brand", 
        "size": 20
      },
      "aggs": { // 是brands聚合的子聚合,也就是分组后对每组分别计算
        "score_stats": { // 聚合名称
          "stats": { // 聚合类型,这里stats可以计算min、max、avg等
            "field": "score" // 聚合字段,这里是score
          }
        }
      }
    }
  }
}

这次的score_stats聚合是在brandAgg的聚合内部嵌套的子聚合。因为我们需要在每个桶分别计算

另外,我们还可以给聚合结果做个排序,例如按照每个桶的酒店平均分做排序:分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

RestAPI实现聚合

API语法
聚合条件与query条件同级别,因此需要使用request.source()来指定聚合条件

聚合条件的语法:

分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
聚合的结果也与查询结果不同,API也比较特殊。不过同样是JSON逐层解析:分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

自动补全

当用户在搜索框输入字符时,我们应该提示出与该字符有关的搜索项,如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
这种根据用户输入的字母,提示完整词条的功能,就是自动补全了

因为需要根据拼音字母来推断,因此要用到拼音分词功能

拼音分词器

要实现根据字母做补全,就必须对文档按照拼音分词。在GitHub上恰好有elasticsearch的拼音分词插件
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


安装方式与IK分词器一样,分三步:

  • 下载解压

  • 上传到虚拟机中,elasticsearch的plugin目录

  • 重启elasticsearch

  • 测试用法如下:

    POST /_analyze
    {
      "text": "如家酒店还不错",
      "analyzer": "pinyin"
    }
    

    结果:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

自定义分词器

默认的拼音分词器会将每个汉字单独分为拼音,而我们希望的是每个词条形成一组拼音,需要对拼音分词器做个性化定制,形成自定义分词器

elasticsearch中分词器(analyzer)的组成包含三部分:

  • character filters:在tokenizer之前对文本进行处理。例如删除字符、替换字符
  • tokenizer:将文本按照一定的规则切割成词条(term)。例如keyword,就是不分词;还有ik_smart
  • tokenizer filter:将tokenizer输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等

文档分词时会依次由这三部分来处理文档:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
声明自定义分词器的语法如下:

PUT /test
{
  "settings": {
    "analysis": {
      "analyzer": { 
        "my_analyzer": { 
          "tokenizer": "ik_max_word",
          "filter": "py"
        }
      },
      "filter": {
        "py": { 
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name":{
        "type": "text",
        "analyzer": "my_analyzer",
        "search_analyzer": "ik_smart"
      }
    }
  }
}

测试:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

自动补全查询

elasticsearch提供了Completion Suggester查询来实现自动补全功能。这个查询会匹配以用户输入内容开头的词条并返回。为了提高补全查询的效率,对于文档中字段的类型有一些约束:

  • 参与补全查询的字段必须是completion类型
  • 字段的内容一般是用来补全的多个词条形成的数组

比如,一个这样的索引库:

# 创建索引库
PUT test
{
  "mappings": {
    "properties": {
      "title":{
        "type": "completion"
      }
    }
  }
}

然后插入下面的数据:

# 示例数据
POST test/_doc
{
  "title": ["Sony", "WH-1000XM3"]
}
POST test/_doc
{
  "title": ["SK-II", "PITERA"]
}
POST test/_doc
{
  "title": ["Nintendo", "switch"]
}

查询的DSL语句如下:

# 自动补全查询
GET /test/_search
{
  "suggest": {
    "title_suggest": {
      "text": "s", // 关键字
      "completion": {
        "field": "title", // 补全查询的字段
        "skip_duplicates": true, // 跳过重复的
        "size": 10 // 获取前10条结果
      }
    }
  }
}

实现酒店搜索框自动补全

现在,我们的hotel索引库还没有设置拼音分词器,需要修改索引库中的配置。但是我们知道索引库是无法修改的,只能删除然后重新创建

另外,我们需要添加一个字段,用来做自动补全,将brand、suggestion、city等都放进去,作为自动补全的提示

因此,总结一下,我们需要做的事情包括:

  1. 修改hotel索引库结构,设置自定义拼音分词器

  2. 修改索引库的name、all字段,使用自定义分词器

  3. 索引库添加一个新字段suggestion,类型为completion类型,使用自定义的分词器

  4. 给HotelDoc类添加suggestion字段,内容包含brand、business

  5. 重新导入数据到hotel库


修改酒店映射结构

// 酒店数据索引库
PUT /hotel
{
  "settings": {
    "analysis": {
      "analyzer": {
        "text_anlyzer": {
          "tokenizer": "ik_max_word",
          "filter": "py"
        },
        "completion_analyzer": {
          "tokenizer": "keyword",
          "filter": "py"
        }
      },
      "filter": {
        "py": {
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "id":{
        "type": "keyword"
      },
      "name":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart",
        "copy_to": "all"
      },
      "address":{
        "type": "keyword",
        "index": false
      },
      "price":{
        "type": "integer"
      },
      "score":{
        "type": "integer"
      },
      "brand":{
        "type": "keyword",
        "copy_to": "all"
      },
      "city":{
        "type": "keyword"
      },
      "starName":{
        "type": "keyword"
      },
      "business":{
        "type": "keyword",
        "copy_to": "all"
      },
      "location":{
        "type": "geo_point"
      },
      "pic":{
        "type": "keyword",
        "index": false
      },
      "all":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart"
      },
      "suggestion":{
          "type": "completion",
          "analyzer": "completion_analyzer"
      }
    }
  }
}

修改HotelDoc实体
HotelDoc中要添加一个字段,用来做自动补全,内容可以是酒店品牌、城市、商圈等信息。按照自动补全字段的要求,最好是这些字段的数组

因此我们在HotelDoc中添加一个suggestion字段,类型为List<String>,然后将brand、city、business等信息放到里面

@Data
@NoArgsConstructor
public class HotelDoc {
    private Long id;
    private String name;
    private String address;
    private Integer price;
    private Integer score;
    private String brand;
    private String city;
    private String starName;
    private String business;
    private String location;
    private String pic;
    private Object distance;
    private Boolean isAD;
    private List<String> suggestion;

    public HotelDoc(Hotel hotel) {
        this.id = hotel.getId();
        this.name = hotel.getName();
        this.address = hotel.getAddress();
        this.price = hotel.getPrice();
        this.score = hotel.getScore();
        this.brand = hotel.getBrand();
        this.city = hotel.getCity();
        this.starName = hotel.getStarName();
        this.business = hotel.getBusiness();
        this.location = hotel.getLatitude() + ", " + hotel.getLongitude();
        this.pic = hotel.getPic();
        // 组装suggestion
        if(this.business.contains("/")){
            // business有多个值,需要切割
            String[] arr = this.business.split("/");
            // 添加元素
            this.suggestion = new ArrayList<>();
            this.suggestion.add(this.brand);
            Collections.addAll(this.suggestion, arr);
        }else {
            this.suggestion = Arrays.asList(this.brand, this.business);
        }
    }
}

重新导入
重新执行之前编写的导入数据功能,可以看到新的酒店数据中包含了suggestion:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


自动补全查询的JavaAPI
之前我们学习了自动补全查询的DSL,而没有学习对应的JavaAPI,这里给出一个示例:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
而自动补全的结果也比较特殊,解析的代码如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


实现搜索框自动补全
查看前端页面,可以发现当我们在输入框键入时,前端会发起ajax请求:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
返回值是补全词条的集合,类型为List<String>

  • cn.itcast.hotel.web包下的HotelController中添加新接口,接收新的请求:

    @GetMapping("suggestion")
    public List<String> getSuggestions(@RequestParam("key") String prefix) {
        return hotelService.getSuggestions(prefix);
    }
    
  • cn.itcast.hotel.service包下的IhotelService中添加方法:

    List<String> getSuggestions(String prefix);
    
  • cn.itcast.hotel.service.impl.HotelService中实现该方法:

    @Override
    public List<String> getSuggestions(String prefix) {
        try {
            // 1.准备Request
            SearchRequest request = new SearchRequest("hotel");
            // 2.准备DSL
            request.source().suggest(new SuggestBuilder().addSuggestion(
                "suggestions",
                SuggestBuilders.completionSuggestion("suggestion")
                .prefix(prefix)
                .skipDuplicates(true)
                .size(10)
            ));
            // 3.发起请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 4.解析结果
            Suggest suggest = response.getSuggest();
            // 4.1.根据补全查询名称,获取补全结果
            CompletionSuggestion suggestions = suggest.getSuggestion("suggestions");
            // 4.2.获取options
            List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions();
            // 4.3.遍历
            List<String> list = new ArrayList<>(options.size());
            for (CompletionSuggestion.Entry.Option option : options) {
                String text = option.getText().toString();
                list.add(text);
            }
            return list;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    

数据同步

elasticsearch中的酒店数据来自于mysql数据库,因此mysql数据发生改变时,elasticsearch也必须跟着改变,这个就是elasticsearch与mysql之间的数据同步
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

思路分析

常见的数据同步方案有三种:

  • 同步调用
  • 异步通知
  • 监听binlog

同步调用:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
基本步骤如下:

  • hotel-demo对外提供接口,用来修改elasticsearch中的数据
  • 酒店管理服务在完成数据库操作后,直接调用hotel-demo提供的接口

异步通知:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
流程如下:

  • hotel-admin对mysql数据库数据完成增、删、改后,发送MQ消息
  • hotel-demo监听MQ,接收到消息后完成elasticsearch数据修改

监听binlog:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
流程如下:

  • 给mysql开启binlog功能
  • mysql完成增、删、改操作都会记录在binlog中
  • hotel-demo基于canal监听binlog变化,实时更新elasticsearch中的内容

选择:
方式一:同步调用

  • 优点:实现简单,粗暴
  • 缺点:业务耦合度高

方式二:异步通知

  • 优点:低耦合,实现难度一般
  • 缺点:依赖mq的可靠性

方式三:监听binlog

  • 优点:完全解除服务间耦合
  • 缺点:开启binlog增加数据库负担、实现复杂度高

实现数据同步

思路:
利用课前资料提供的hotel-admin项目作为酒店管理的微服务。当酒店数据发生增、删、改时,要求对elasticsearch中数据也要完成相同操作。

步骤:

  • 导入课前资料提供的hotel-admin项目,启动并测试酒店数据的CRUD

  • 声明exchange、queue、RoutingKey

  • 在hotel-admin中的增、删、改业务中完成消息发送

  • 在hotel-demo中完成消息监听,并更新elasticsearch中数据

  • 启动并测试数据同步功能


导入demo
导入课前资料提供的hotel-admin项目:
运行后,访问 http://localhost:8099
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
其中包含了酒店的CRUD功能:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


声明交换机、队列:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

  • 引入依赖
    在hotel-admin、hotel-demo中引入rabbitmq的依赖:

    <!--amqp-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    
  • 配置mq地址

    spring:
      rabbitmq:
        host: 192.168.1.12
        port: 5672
        username: xiaowu
        password: 123321
        virtual-host: /
    
  • 声明队列交换机名称
    在hotel-admin和hotel-demo中的cn.itcast.hotel.constatnts包下新建一个类MqConstants

     public class MqConstants {
        /**
         * 交换机
         */
        public final static String HOTEL_EXCHANGE = "hotel.topic";
        /**
         * 监听新增和修改的队列
         */
        public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
        /**
         * 监听删除的队列
         */
        public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
        /**
         * 新增或修改的RoutingKey
         */
        public final static String HOTEL_INSERT_KEY = "hotel.insert";
        /**
         * 删除的RoutingKey
         */
        public final static String HOTEL_DELETE_KEY = "hotel.delete";
    }
    
  • 声明队列交换机
    在hotel-demo中,定义配置类,声明队列、交换机:

    @Configuration
    public class MqConfig {
        @Bean
        public TopicExchange topicExchange(){
            return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
        }
    
        @Bean
        public Queue insertQueue(){
            return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
        }
    
        @Bean
        public Queue deleteQueue(){
            return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
        }
    
        @Bean
        public Binding insertQueueBinding(){
            return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
        }
    
        @Bean
        public Binding deleteQueueBinding(){
            return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
        }
    }
    
  • 发送MQ消息
    在hotel-admin中的增、删、改业务中分别发送MQ消息
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

  • 接收MQ消息
    hotel-demo接收到MQ消息要做的事情包括:

    • 新增消息:根据传递的hotel的id查询hotel信息,然后新增一条数据到索引库
    • 删除消息:根据传递的hotel的id删除索引库中的一条数据

    首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、删除业务

    void deleteById(Long id);
    
    void insertById(Long id);
    

    给hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中实现业务:

    @Override
    public void deleteById(Long id) {
        try {
            // 1.准备Request
            DeleteRequest request = new DeleteRequest("hotel", id.toString());
            // 2.发送请求
            client.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    
    @Override
    public void insertById(Long id) {
        try {
            // 0.根据id查询酒店数据
            Hotel hotel = getById(id);
            // 转换为文档类型
            HotelDoc hotelDoc = new HotelDoc(hotel);
    
            // 1.准备Request对象
            IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
            // 2.准备Json文档
            request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
            // 3.发送请求
            client.index(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    

    编写监听器:在hotel-demo中的cn.itcast.hotel.mq包新增一个类:

    @Component
    public class HotelListener {
    
        @Autowired
        private IHotelService hotelService;
    
        /**
         * 监听酒店新增或修改的业务
         * @param id 酒店id
         */
        @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
        public void listenHotelInsertOrUpdate(Long id){
            hotelService.insertById(id);
        }
    
        /**
         * 监听酒店删除的业务
         * @param id 酒店id
         */
        @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
        public void listenHotelDelete(Long id){
            hotelService.deleteById(id);
        }
    }
    

ES集群

单机的elasticsearch做数据存储,必然面临两个问题:海量数据存储问题、单点故障问题。

  • 海量数据存储问题:将索引库从逻辑上拆分为N个分片(shard),存储到多个节点
  • 单点故障问题:将分片数据在不同节点备份(replica )

ES集群相关概念:

  • 集群(cluster):一组拥有共同的 cluster name 的 节点
  • 节点(node) :集群中的一个 Elasticearch 实例
  • 分片(shard):索引可以被拆分为不同的部分进行存储,称为分片。在集群环境下,一个索引的不同分片可以拆分到不同的节点中
    解决问题:数据量太大,单点存储量有限的问题
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    此处,我们把数据分成3片:shard0、shard1、shard2
  • 主分片(Primary shard):相对于副本分片的定义
  • 副本分片(Replica shard)每个主分片可以有一个或者多个副本,数据和主分片一样

数据备份可以保证高可用,但是每个分片备份一份,所需要的节点数量就会翻一倍,成本实在是太高了!

为了在高可用和成本间寻求平衡,我们可以这样做:

  • 首先对数据分片,存储到不同节点
  • 然后对每个分片进行备份,放到对方节点,完成互相备份

这样可以大大减少所需要的服务节点数量,如图,我们以3分片,每个分片备份一份为例:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
现在,每个分片都有1个备份,存储在3个节点:

  • node0:保存了分片0和1
  • node1:保存了分片0和2
  • node2:保存了分片1和2

搭建集群

创建es集群
首先编写一个docker-compose文件,内容如下:

version: '2.2'
services:
  es01:
    image: elasticsearch:7.12.1 #镜像
    container_name: es01	#容器名称
    environment:	#环境变量
      - node.name=es01	#节点名称
      - cluster.name=es-docker-cluster #集群名称
      - discovery.seed_hosts=es02,es03	#集群中其他节点的地址
      - cluster.initial_master_nodes=es01,es02,es03 #初始化的主节点
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m" #JVM堆内存大小
    volumes:
      - data01:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
    networks:
      - elastic
  es02:
    image: elasticsearch:7.12.1
    container_name: es02
    environment:
      - node.name=es02
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - data02:/usr/share/elasticsearch/data
    ports:
      - 9201:9200
    networks:
      - elastic
  es03:
    image: elasticsearch:7.12.1
    container_name: es03
    environment:
      - node.name=es03
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es02
      - cluster.initial_master_nodes=es01,es02,es03
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - data03:/usr/share/elasticsearch/data
    networks:
      - elastic
    ports:
      - 9202:9200
volumes:
  data01:
    driver: local
  data02:
    driver: local
  data03:
    driver: local

networks:
  elastic:
    driver: bridge

es运行需要修改一些linux系统权限,修改/etc/sysctl.conf文件

vi /etc/sysctl.conf

添加下面的内容:

vm.max_map_count=262144

然后执行命令,让配置生效:

sysctl -p

通过docker-compose启动集群:

docker-compose up -d

集群状态监控
kibana可以监控es集群,不过新版本需要依赖es的x-pack 功能,配置比较复杂

这里推荐使用cerebro来监控es集群状态,解压即可使用,非常方便
解压好的目录如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
进入对应的bin目录:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
双击其中的cerebro.bat文件即可启动服务
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
访问http://localhost:9000 即可进入管理界面:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
输入你的elasticsearch的任意节点的地址和端口,点击connect即可:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
绿色的条,代表集群处于绿色(健康状态)

利用kibana的DevTools创建索引库
在DevTools中输入指令:

PUT /itcast
{
  "settings": {
    "number_of_shards": 3, // 分片数量
    "number_of_replicas": 1 // 副本数量
  },
  "mappings": {
    "properties": {
      // mapping映射定义 ...
    }
  }
}

集群脑裂问题

集群脑裂问题
elasticsearch中集群节点有不同的职责划分:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
默认情况下,集群中的任何一个节点都同时具备上述四种角色

但是真实的集群一定要将集群职责分离:

  • master节点:对CPU要求高,但是内存要求低
  • data节点:对CPU和内存要求都高
  • coordinating节点:对网络带宽、CPU要求高

职责分离可以让我们根据不同节点的需求分配不同的硬件去部署。而且避免业务之间的互相干扰。

一个典型的es集群职责划分如图:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


脑裂问题
脑裂是因为集群中的节点失联导致的
例如一个集群中,主节点与其它节点失联:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
此时,node2和node3认为node1宕机,就会重新选主:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
当node3当选后,集群继续对外提供服务,node2和node3自成集群,node1自成集群,两个集群数据不同步,出现数据差异

当网络恢复后,因为集群中有两个master节点,集群状态的不一致,出现脑裂的情况:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
解决脑裂的方案是,要求选票超过 ( eligible节点数量 + 1 )/ 2 才能当选为主,因此eligible节点数量最好是奇数。对应配置项是discovery.zen.minimum_master_nodes,在es7.0以后,已经成为默认配置,因此一般不会发生脑裂问题

例如:3个节点形成的集群,选票必须超过 (3 + 1) / 2 ,也就是2票。node3得到node2和node3的选票,当选为主。node1只有自己1票,没有当选。集群中依然只有1个主节点,没有出现脑裂

集群分布式存储

当新增文档时,应该保存到不同分片,保证数据均衡,那么coordinating node如何确定数据该存储到哪个分片呢?

分片存储测试
插入三条数据:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
测试可以看到,三条数据分别在不同分片:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
结果:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch


分片存储原理
elasticsearch会通过hash算法来计算文档应该存储到哪个分片:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
说明:

  • _routing默认是文档的id
  • 算法与分片数量有关,因此索引库一旦创建,分片数量不能修改!

新增文档的流程如下:
分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

集群分布式查询

elasticsearch的查询分成两个阶段:

  • scatter phase:分散阶段,coordinating node会把请求分发到每一个分片

  • gather phase:聚集阶段,coordinating node汇总data node的搜索结果,并处理为最终结果集返回给用户
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

集群故障转移

集群的master节点会监控集群中的节点状态,如果发现有节点宕机,会立即将宕机节点的分片数据迁移到其它节点,确保数据安全,这个叫做故障转移文章来源地址https://www.toymoban.com/news/detail-759384.html

  • 例如一个集群结构如图:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    现在,node1是主节点,其它两个节点是从节点
  • 突然,node1发生了故障:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    宕机后的第一件事,需要重新选主,例如选中了node2:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch
    node2成为主节点后,会检测集群监控状态,发现:shard-1、shard-0没有副本节点。因此需要将node1上的数据迁移到node2、node3:
    分布式搜索引擎ES,Java笔记,搜索引擎,分布式,elasticsearch

到了这里,关于分布式搜索引擎ES的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 搜索引擎ElasticSearch分布式搜索和分析引擎学习,SpringBoot整合ES个人心得

    Elasticsearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎,基于RESTful web接口。Elasticsearch是用Java语言开发的,并作为Apache许可条款下的开放源码发布,是一种流行的企业级搜索引擎。Elasticsearch用于云计算中,能够达到实时搜索,稳定,可靠,

    2024年02月04日
    浏览(72)
  • elasticsearch(ES)分布式搜索引擎03——(RestClient查询文档,ES旅游案例实战)

    文档的查询同样适用昨天学习的 RestHighLevelClient对象,基本步骤包括: 1)准备Request对象 2)准备请求参数 3)发起请求 4)解析响应 我们以match_all查询为例 3.1.1.发起查询请求 代码解读: 第一步,创建 SearchRequest 对象,指定索引库名 第二步,利用 request.source() 构建DSL,DSL中可

    2024年02月07日
    浏览(53)
  • elasticsearch(ES)分布式搜索引擎04——(数据聚合,自动补全,数据同步,ES集群)

    **聚合(aggregations)**可以让我们极其方便的实现对数据的统计、分析、运算。例如: 什么品牌的手机最受欢迎? 这些手机的平均价格、最高价格、最低价格? 这些手机每月的销售情况如何? 实现这些统计功能的比数据库的sql要方便的多,而且查询速度非常快,可以实现近

    2024年02月08日
    浏览(51)
  • SpringCloud分布式搜索引擎、数据聚合、ES和MQ的结合使用、ES集群的问题

    目录 数据聚合 聚合的分类 ​编辑 DSL实现Bucket聚合 ​编辑  DSL实现Metrics聚合​编辑 RestAPI实现聚合  对接前端接口​编辑  自定义分词器​编辑 Completion suggester查询 Completion suggester查询 酒店数据自动补全 实现酒店搜索框界面输入框的自动补全  数据同步问题分析​编辑 同

    2024年02月16日
    浏览(49)
  • 项目1在线交流平台-6.Elasticsearch分布式搜索引擎-3.ES结合Kafka应用-开发社区搜索功能

    参考牛客网高级项目教程 狂神说Elasticsearch教程笔记 尚硅谷Elasticsearch教程笔记 1.在业务层处理好搜索帖子的服务 包括保存帖子到ES服务器 从服务器中删除帖子 从服务器中查询帖子 2.发布事件 在controller层,结合kafka,发布帖子、增加评论时,数据放入消息队列 异步消费消息

    2024年02月02日
    浏览(56)
  • elasticsearch(ES)分布式搜索引擎01——(初识ES,索引库操作和文档操作,RestClient操作索引库和文档)

    1.1.1.elasticsearch的作用 elasticsearch是一款非常强大的开源搜索引擎,具备非常多强大功能,可以帮助我们从海量数据中快速找到需要的内容 1.1.2.ELK技术栈 elasticsearch结合kibana、Logstash、Beats,也就是elastic stack(ELK)。被广泛应用在日志数据分析、实时监控等领域: 而elasticsearc

    2024年02月07日
    浏览(62)
  • Java远程连接本地开源分布式搜索引擎ElasticSearch

    简单几步,结合Cpolar内网穿透工具实现Java远程连接操作本地Elasticsearch。 什么是elasticsearch?一个开源的分布式搜索引擎,具备非常多强大功能,可以用来实现搜索、日志统计、分析、系统监控等功能,可以帮助我们从海量数据中快速找到需要的内容。 Cpolar内网穿透提供了更高

    2024年02月05日
    浏览(50)
  • 第七章-分布式搜索引擎-ES:全文查询、分词查询、精确查询、地理坐标查询、组合查询(bool、funtion_score)以及RestApi

    DSL查询分类 全文查询、分词查询、非分词查询、地理坐标查询、组合查询 match_all 查询所有,不需要查询条件,固定写法_search 第一个hits就是命中的数据 ,total就是条数,第二个hits是source嘞   全文检索查询 我们不要整多个字段查询,参与的字段越多,查询速度越慢,如果有

    2024年01月16日
    浏览(81)
  • 分布式搜索引擎

    elasticsearch的查询依然是基于JSON风格的DSL来实现的。 Elasticsearch提供了基于JSON的DSL(Domain Specific Language)来定义查询。常见的查询类型包括: 查询所有 :查询出所有数据,一般测试用。例如:match_all 全文检索(full text)查询 :利用分词器对用户输入内容分词,然后去倒排索

    2024年02月10日
    浏览(41)
  • 分布式搜索引擎ElasticSearch——搜索功能

    DSL查询分类 DSL官方文档 全文检索查询 精确查询 地理查询 复合查询 Function Score Query function score query Boolean Query 排序 分页 官方文档 高亮 快速入门 match,term,range,bool查询 排序和分页 高亮显示 就是在前面抽取的解析代码中进一步添加关于高亮的解析部分,因为highlight和so

    2024年02月01日
    浏览(54)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包