最强分布式搜索引擎——ElasticSearch
本篇我们将会介绍到一种特殊的类似数据库存储机制的搜索引擎工具——ES
elasticsearch是一款非常强大的开源搜索引擎，具备非常多强大功能，可以帮助我们从海量数据中快速找到需要的内容
我们会从下面几个角度来讲解ElasticSearch：
- ES概述
- ES索引库操作
- ES文档操作
- IDEA索引库操作
- IDEA文档操作
- ES数据搜索
- IDEA数据搜索
- ES数据聚合
- IDEA数据聚合
- MQ数据同步
ES概述
首先我们先来简单介绍一下ElasticSearch
ES概念
我们首先来简单介绍一下ES：
- ES是一款特殊的搜索引擎工具，它在广大场景都有所使用
- ES的本质是基于倒排索引机制，它可以快速地检索某一个词汇并找到对应的所属位置
ELK技术栈
我们给出ELK的组成部分：
- ELK由四部分组成：elasticsearch，kibana、Logstash和Beats
- kibana：负责将数据可视化展示
- elasticsearch：elastic stack的核心，负责存储、搜索、分析数据
- Logstash，Beats：负责数据的抓取
我们给出一张结构图来表示ELK的整体结构：
ES核心机制
我们如果需要学习ES，那么首先就需要了解ES的核心机制——倒排索引
我们首先来介绍正向索引：
- MySQL数据库中所使用的方法就是正向索引
- MySQL会首先产生一个id，然后根据这个id去生成索引 ，然后根据索引进行数据的查询
- 简单来说：如果我们通过id去查找或者通过索引去查找，速度就会非常快；但是如果我们不是通过索引或者采用模糊查询，速度变慢
首先我们还需要了解倒排索引的一些关键字：
- 文档：我们的一个对象，就被成为文档，类似于MySQL中的一行数据，存在一个唯一id
- 词条：对文档数据或用户搜索数据，利用某种算法分词，得到的具备含义的词语就是词条
那么我们再来介绍倒排索引：
- 倒排索引不将id作为查找字段，而是将保存的数据分割作为查找字段，然后找到该字段后去找对应的对象
- 例如小米手机，华为手机，华为小米充电器等一系列文档，这些文档都有一个唯一id
- 这时就会生成小米，手机，华为，充电器这样的数据内容存放在ES中，这些词汇后会跟着一个id的集合记录哪些文档包含该词条
- 当我们查找时，我们会去直接查找字段，然后查看对应的id号，然后找到该id对应的对象并返回该对象结果
我们可以对两者做出一个简单的比较：
-
正向索引优点：可以给多个字段创建索引；根据索引字段搜索、排序速度非常快
-
正向索引缺点：根据非索引字段，或者索引字段中的部分词条查找时，只能全表扫描。
-
倒排索引优点：根据词条搜索、模糊搜索时，速度非常快
-
倒排索引缺点：只能给词条创建索引，而不是字段；无法根据字段做排序
ES核心概念
我们来介绍一些ES中的核心概念：
- 文档
- ES是面向文档进行存储的，文档数据会被序列化为json格式后存储在elasticsearch中
- 而Json文档中往往包含很多的字段（Field），类似于数据库中的列，这些字段就会被作为搜索条件
- 索引和映射
-
索引实际上对标MySQL的数据库，一个索引就是一个具体的数据库
-
映射实际上对标MySQL的约束信息，用于对索引进行一定条件的限制
-
通俗来讲：索引就是就是相同类型的文档的集合，映射是索引中文档的字段约束信息
ES特点比较
我们将ES和MySQL进行一个简单的对比，我们会发现两者结构上非常相似：
| 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实现；
- 两者再基于某种方式，实现数据的同步，保证一致性，来实现实际开发
ES及相关产品安装
既然要使用ES，那么我们首先需要下载ES：
- 因为我们还需要部署kibana容器，因此需要让es和kibana容器互联。这里先创建一个网络
docker network create es-net - 下载es镜像的tar包，进行加载
# 导入数据 docker load -i es.tar - 采用docker进行部署
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`：端口映射配置 然后我们还需要去部署一个kibana，kibana可以给我们提供一个elasticsearch的可视化界面，便于我们学习：
- 下载kibana镜像的tar包，并进行加载
# 导入数据 docker load -i kibana.tar - 运行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`：端口映射配置 - 此时，在浏览器输入地址访问：http://192.168.150.101:5601，即可看到结果(虚拟机ip:5601)
最后还需要一个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 ES索引库操作
我们首先来介绍对ES索引库的操作
映射属性介绍
我们首先需要去介绍ES索引库的Mapping：
- mapping是对索引库中文档的约束，常见的mapping属性包括很多种
我们下面来一一介绍：
- type字段数据类型
| TYPE名称 | TYPE含义 |
|---|---|
| text | 字符串（可以被划分，可分词的文本） |
| keyword | 字符串（不可被划分，精确值，例如：品牌、国家、ip地址） |
| long、integer、short、byte、double、float | 常见数值类型 |
| boolean | 布尔值 |
| date | 日期值 |
| object | 对象 |
- index索引是否存在
- 默认为true，表示可以作为索引存在
- analyzer分词器
- 后面跟具体的分词器，用于更换分词器种类
- properties子资源
- 该字段的子字段
我们给出一个简单举例：
{ "age": 21,// 类型为 integer；参与搜索，因此需要index为true；无需分词器 "weight": 52.1, "isMarried": false, "info": "河南师范大学",// 类型为字符串，需要分词，因此是text；参与搜索，因此需要index为true；分词器可以用ik_smart "email": "zy@hsd.cn",// 类型为字符串，但是不需要分词，因此是keyword；不参与搜索，因此需要index为false；无需分词器 "score": [99.1, 99.5, 98.9], "name": { "firstName": "云", "lastName": "赵" } } 索引库CURD
这里我们统一使用Kibana编写DSL的方式来演示
创建索引库
下面我们来介绍创建索引库的说明，架构和案例：
/* - 请求方式：PUT - 请求路径：/索引库名，可以自定义 - 请求参数：mapping映射 */ /* 架构 */ PUT /索引库名称 { "mappings": { "properties": { "字段名":{ "type": "text", "analyzer": "ik_smart" }, "字段名2":{ "type": "keyword", "index": "false" }, "字段名3":{ "properties": { "子字段": { "type": "keyword" } } }, } } } /* 案例 */ PUT /qiuluo { "mappings": { "properties": { "info":{ "type": "text", "analyzer": "ik_smart" }, "email":{ "type": "keyword", "index": "falsae" }, "name":{ "properties": { "firstName": { "type": "keyword" } } } } } } 查询索引库
首先我们给出一个简单案例：
/* - 请求方式：GET - 请求路径：/索引库名 - 请求参数：无 */ /* 架构 */ GET /索引库名 /* 案例 */ GET /qiuluo 修改索引库
我们需要注意索引库是无法修改已存在的结构的，但是可以对索引库进行新增操作：
/* - 请求方式：PUT - 请求路径：/索引库名/_mapping - 请求参数：修改内容 */ /* 架构 */ PUT /索引库名/_mapping { "properties": { "新字段名":{ "type": "integer" } } } 删除索引库
删除索引库和前两个的语法基本相似：
/* - 请求方式：DELETE - 请求路径：/索引库名 - 请求参数：无 */ /* 架构 */ DELETE /索引库名 ES文档操作
下面我们来介绍ES的文档操作
ES内容补充
其中在索引库和文档之间原本还有一层Type：
- Type类似于MySQL中的表，在ES 5.X版本中一个索引Index下可以有多个类型Type
- 在ES的后期版本中Type一般只有一个，后期就被默认为doc名称的Type，所以我们后续的操作中会见到doc这个词
文档CURD
这里我们统一使用Kibana编写DSL的方式来演示
新增文档
我们同样直接给出具体的解释和代码：
/* - 请求方式：POST - 请求路径：/索引库名/_doc/文档id - 请求参数：具体的字段值和存储值 */ /* 模板 */ POST /索引库名/_doc/文档id { "字段1": "值1", "字段2": "值2", "字段3": { "子属性1": "值3", "子属性2": "值4" }, // ... } /* 举例 */ POST /qiuluo/_doc/1 { "info": "河南师范大学", "email": "zy@hsd.cn", "name": { "firstName": "云", "lastName": "赵" } } 查询文档
我们同样以DSL语句书写查询文档代码：
/* - 请求方式：GET - 请求路径：/索引库名/_doc/文档id - 请求参数：无 */ /* 模板 */ GET /{索引库名称}/_doc/{id} /* 举例 */ GET /qiuluo/_doc/1 删除文档
删除文档的格式和查询文档的格式基本相同：
/* - 请求方式：DELETE - 请求路径：/索引库名/_doc/文档id - 请求参数：无 */ /* 模板 */ DELETE /{索引库名}/_doc/id值 /* 举例 */ DELETE /qiuluo/_doc/1 修改文档
我们修改文档大致分为两种：全量修改和增量修改：
/* 全量修改的具体步骤： - 根据指定的id删除文档，新增一个相同id的文档 - 请求方式：PUT - 请求路径：/索引库名/_doc/文档id - 请求参数：全部字段内容 */ /* 模板 */ PUT /{索引库名}/_doc/文档id { "字段1": "值1", "字段2": "值2", // ... 略 } /* 举例 */ PUT /qiuluo/_doc/1 { "info": "河南师范大学", "email": "zy@hsd.cn", "name": { "firstName": "云", "lastName": "赵" } } /* 增量修改的具体步骤： - 修改文档中的部分字段 - 请求方式：POST - 请求路径：/索引库名/_update/文档id - 请求参数：只修改需要修改的部分 */ /* 模板 */ POST /{索引库名}/_update/文档id { "doc": { "字段名": "新的值", } } /* 举例 */ POST /heima/_update/1 { "doc": { "email": "ZhaoYun@hsd.cn" } } IDEA索引库操作
下面我们在IDEA上使用API去完成ES的使用
IDEA基本准备
我们在使用ES之前需要先完成几项准备工作：
- 导入数据
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; - 导入项目
- 具体映射分析
/* 下述是我们需要插入的数据，我们需要对其分析并简单了解，其具体思路不再解释 我们需要介绍几个新的内容： 1. geo_point 属于type的一种，表示地理坐标类型，里面包含精度、纬度 geo_point属于由两个数组成的一个点；而geo_shape是由多个geo_point所组成的一条线或一个区域 2. all 一个组合字段，其目的是将多字段的值 利用copy_to合并，提供给用户搜索 all字段在最后进行标明，但在前面的某些字段中我们采用了copy_to字段，后面跟上了all表示将该字段值拷贝一份到all中 也就是说all这个字段是由name，brand，city等多个字段连接起来的，这点是为了帮助我们在后面的按词条快速查询时方便 */ 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", "copy_to": "all" }, "starName":{ "type": "keyword" }, "business":{ "type": "keyword" }, "location":{ "type": "geo_point" }, "pic":{ "type": "keyword", "index": false }, "all":{ "type": "text", "analyzer": "ik_max_word" } } } } - 引入依赖并修改版本
<!-- 我们在IDEA中引用ES去进行操作必定需要借助工具，而这里我们需要借用RestHighLevelClient去完成ES的操作，所以我们需要先去引入依赖 --> <!--引入es的RestHighLevelClient依赖--> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> </dependency> <!--因为SpringBoot默认的ES版本是7.6.2，所以我们需要覆盖默认的ES版本--> <properties> <java.version>1.8</java.version> <elasticsearch.version>7.12.1</elasticsearch.version> </properties> - 生成一个RestHighLevelClient去完成ES操作
// 这里仅是一个代码展示 RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(// 可以看作固定生成语句 HttpHost.create("http://192.168.150.101:9200")// 这里需要给出ES的链接地址，给出你的虚拟机ES端口 )); - 生成一个专门的测试类去完成后面的操作
package cn.itcast.hotel; import org.apache.http.HttpHost; import org.elasticsearch.client.RestHighLevelClient; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.io.IOException; 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(); } } 创建索引库
我们首先需要定义出具体的JSON数据内容：
package cn.itcast.hotel.constants; 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": falsen" + " },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": falsen" + " },n" + " "all":{n" + " "type": "text",n" + " "analyzer": "ik_max_word"n" + " }n" + " }n" + " }n" + "}"; } 然后我们就可以在我们先前定义的测试类中进行ES操作：
@Test void createHotelIndex() throws IOException { // 1.创建Request对象（创建是CreateIndexRequest） CreateIndexRequest request = new CreateIndexRequest("hotel");// 参数是索引名称 // 2.准备请求的参数：DSL语句（对request的source属性进行设置） request.source(MAPPING_TEMPLATE, XContentType.JSON);// 第一个参数是具体JSON，第二个参数是第一个参数类型 // 3.发送请求（client.indices()是一个针对索引的对象，里面封装了各种索引方法） client.indices().create(request, RequestOptions.DEFAULT);// 第一个参数是请求，第二个是默认格式 } 删除索引库
我们直接给出对应的代码展示：
@Test void testDeleteHotelIndex() throws IOException { // 1.创建Request对象（创建是DeleteIndexRequest） DeleteIndexRequest request = new DeleteIndexRequest("hotel");// 参数是索引名称 // 2.发送请求 client.indices().delete(request, RequestOptions.DEFAULT); } 获得索引库
我们这里获得索引库并判断该索引是否存在：
@Test void testExistsHotelIndex() throws IOException { // 1.创建Request对象（创建是GetIndexRequest） GetIndexRequest request = new GetIndexRequest("hotel"); // 2.发送请求(这里调用的请求是exists判断该索引是否存在，最后返回一个boolean值用于判断) boolean exists = client.indices().exists(request, RequestOptions.DEFAULT); // 3.输出 System.err.println(exists ? "索引库已经存在！" : "索引库不存在！"); } 操作总结
我们其实可以将ES的DSL操作和IDEA的操作做一个简单对比：
我们可以注意到：
- 创建Request对象。因为是创建索引库的操作，因此Request是CreateIndexRequest。
- 添加请求参数，就是添加DSL的JSON参数部分，这里是定义了静态字符串常量MAPPING_TEMPLATE
- 发送请求，client.indices()方法的返回值是IndicesClient类型，封装了所有与索引库操作有关的方法。
因而我们可以给出具体的流程：
- 初始化RestHighLevelClient
- 创建IndexRequest，例如GetIndexRequest等
- 当需要DSL数据时，我们提前封装并将其放入到request请求中
- 调用RestHighLevelClient的indices方法获得IndicesClient，并调用其各类方法
IDEA文档操作
下面我们来介绍IDEA的文档操作
IDEA基本准备
我们首先需要准备一个对应的实体类：
// 这次我们主要是针对hotel旅馆进行一个文档信息的填充 // 我们在MySQL数据库中存放了相对应的数据，我们首先给出对应实体类 @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; } // 但是我们需要注意：我们之前索引库中我们存在一个location属性是将longitude和latitude结合起来形成一个坐标 // 因而我们需要一个DTO来完成信息封装 package cn.itcast.hotel.pojo; import lombok.Data; import lombok.NoArgsConstructor; @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(); } } 我们同样提前准备一个简单的测试类：
package cn.itcast.hotel; import cn.itcast.hotel.pojo.Hotel; import cn.itcast.hotel.service.IHotelService; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import java.io.IOException; import java.util.List; @SpringBootTest public class HotelDocumentTest { // 这里是hotel的服务层，我们调用其方法来获得mysql的相关数据来填充进ES的文档中 @Autowired private IHotelService hotelService; // RestHighLevelClient private RestHighLevelClient client; // RestHighLevelClient封装 @BeforeEach void setUp() { this.client = new RestHighLevelClient(RestClient.builder( HttpHost.create("http://192.168.150.101:9200") )); } // RestHighLevelClient释放资源 @AfterEach void tearDown() throws IOException { this.client.close(); } } 新增文档
我们首先给出一个简单的案例展示：
- DSL语法展示
POST /{索引库名}/_doc/1 { "name": "Jack", "age": 21 } - Java语法展示
@Test void testAddDocument() throws IOException { // 1.准备Request对象（这里是IndexRequest，Index类似于新添的概念） IndexRequest request = new IndexRequest("indexName").id(1);// 第一个参数索引名称，后面id跟的是文档id // 2.准备Json文档 request.source("{"name":"jack","age":21}", XContentType.JSON);// 同样是对应的数据信息 // 3.发送请求 client.index(request, RequestOptions.DEFAULT);// 这里文档我们直接采用client调用方法即可，index就是新添操作 } 然后我们再针对MySQL数据库信息将其挪移到ES中：
@Test void testAddDocument() throws IOException { // 这里都是Spring和MyBatisPlus的内容 // 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} 然后我们给出对应的Java代码：
@Test void testGetDocumentById() throws IOException { // 1.准备Request（这里是GetRequest，Get就是获得） GetRequest request = new GetRequest("hotel", "61082");// 第一个参数是索引名称，第二个参数是id // 2.发送请求，得到响应 GetResponse response = client.get(request, RequestOptions.DEFAULT);// get方法获得response // 3.解析响应结果 String json = response.getSourceAsString();// 我们通过response获得对应的数据 // 4.将数据解析获得结果 HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class); System.out.println(hotelDoc); } 删除文档
我们同样给出DSL语句：
DELETE /hotel/_doc/{id} 然后我们给出对应的Java代码：
@Test void testDeleteDocument() throws IOException { // 1.准备Request DeleteRequest request = new DeleteRequest("hotel", "61083"); // 2.发送请求 client.delete(request, RequestOptions.DEFAULT); } 修改文档
修改文档同样可以划分为两种：
- 增量修改：修改文档中的指定字段值
- 全量修改：本质是先根据id删除，再新增；全量修改与新增的API完全一致，判断依据新增或修改依据是ID是否存在
我们直接给出增强修改对应的Java代码：
@Test void testUpdateDocument() throws IOException { // 1.准备Request UpdateRequest request = new UpdateRequest("hotel", "61083"); // 2.准备请求参数(注意：这里采用了doc而不是source) request.doc( "price", "952", "starName", "四钻" ); // 3.发送请求 client.update(request, RequestOptions.DEFAULT); } 批量导入文档
下面我们将会利用BulkRequest批量将数据库数据导入到索引库中：
@Test void testBulkRequest() throws IOException { // 批量查询酒店数据 List<Hotel> hotels = hotelService.list(); // 1.创建Request // BulkRequest实际上只是一个request集合体，它可以添加多种其他类型的request // 包括IndexRequest新增，UpdateRequest修改，DeleteRequest删除三种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); } 操作总结
我们其实可以将ES的DSL操作和IDEA的操作做一个简单对比：
我们可以注意到同样划分为三步：
- 创建Request对象。因为是创建文档的操作，因此Request是IndexRequest。
- 准备请求参数，也就是DSL中的JSON文档
- 发送请求，这里是直接采用client中的方法进行文档操作
因而我们可以给出具体的流程：
- 初始化RestHighLevelClient
- 创建Request。前缀名包括有Index、Get、Update、Delete、Bulk
- 准备参数Index、Update、Bulk的Request请求时需要
- 发送请求。调用RestHighLevelClient的各种方法，包括index、get、update、delete、bulk等方法
- 解析结果，例如get获得数据后将其通过JSON的parseObject转化为Domain实体并输出
ES数据搜索
在前面的章节其实只是完成了ES的一个数据储存功能，但ES的核心功能是数据快速检索查询
数据查询分类
Elasticsearch提供了基于JSON的DSL来定义查询，大致有以下几种查询方式：
- 查询所有：查询出所有数据，一般测试用
- 全文检索查询：利用分词器对用户输入内容分词，然后去倒排索引库中匹配
- 精确查询：根据精确词条值查找数据，一般是查找keyword、数值、日期、boolean等类型字段
- 地理查询：根据经纬度查询
- 复合查询：复合查询可以将上述各种查询条件组合起来，合并查询条件
我们再给出一个基本查询模板：
GET /indexName/_search { "query": { "查询类型": { "查询条件": "条件值" } } } 查询所有
查询所有的关键字是"match_all",无查询条件：
// 查询所有 GET /indexName/_search { "query": { "match_all": { } } } 全文检索查询
首先我们需要了解全文检索查询的基础流程：
- 对用户搜索的内容做分词，得到词条
- 根据词条去倒排索引库中匹配，得到文档id
- 根据文档id找到文档，返回给用户
其中全文检索查询可以大致分为两种：
- match查询：单字段查询
- multi_match查询：多字段查询，任意一个字段符合条件就算符合查询条件
我们分别给出全文检索模板：
// match查询：仅一个字段，一个匹配内容 GET /indexName/_search { "query": { "match": { "FIELD": "TEXT"// FIELD为对应的字段名称，TEXT为查询内容 } } } // multi_match查询：多字段查询，任意一个字段符合条件就算符合查询条件 GET /indexName/_search { "query": { "multi_match": { "query": "TEXT", // TEXT为查询内容 "fields": ["FIELD1", " FIELD12"] // FIELD1，2，3，均为查询字段 } } } 我们同时给出一个简单案例：
// 下述两个全文检索含义相同 // match查询：仅一个字段，一个匹配内容 GET /hotel/_search { "query": { "match": { "all": "外滩" // 这里仅针对all字段进行"外滩"的检索，但是all字段是由多个字段copy_to产生的 } } } // multi_match查询：多字段查询，任意一个字段符合条件就算符合查询条件 GET /hotel/_search { "query": { "multi_match": { "query": "外滩", "fields": ["brand", "name","business"] // 这三个字段均使用了copy_to至all字段，固两个查询含义相同 } } } // 但是查询字段越多其速度越慢，所以match查询的速度是要远高于multi_match的 精准查询
精确查询一般是查找keyword、数值、日期、boolean等类型字段，我们一般会将精准查询分为两部分：
- term：根据词条精确值查询
- range：根据值的范围查询
首先我们先来介绍term查询：
// 模板 GET /indexName/_search { "query": { "term": { // 表示精准查询 "FIELD": { // 字段名 "value": "VALUE" // 查找字段内容 } } } } // 案例 GET /hotel/_search { "query": { "term": { "city": { "value": "北京" // 表示查找地点在北京的宾馆 } } } } 下面我们再来介绍range查询：
// 模板 GET /indexName/_search { "query": { "range": { "FIELD": { // 这里更换字段名 "gte": 10, // 这里的gte代表大于等于，gt则代表大于 "lte": 20 // lte代表小于等于，lt则代表小于 } } } } // 案例 GET /hotel/_search { "query": { "range": { "price": { "gte": 200, "lte": 1000 // 这里表示寻找价格在200~1000之间的宾馆 } } } } 地理查询
所谓的地理坐标查询，其实就是根据经纬度查询，地理查询通常被分为两方面：
- 矩形范围查询：分别规定左上角和右下角来规定矩形范围进行区域划分
- 附近范围查询：查询到指定中心点小于某个距离值的所有文档
我们首先来介绍矩形范围查询：
// geo_bounding_box查询 GET /indexName/_search { "query": { "geo_bounding_box": { "FIELD": { // 这里的FIELD需要修改为字段名 "top_left": { // 左上点 "lat": 31.1, "lon": 121.5 }, "bottom_right": { // 右下点 "lat": 30.9, "lon": 121.7 } } } } } 下面我们再来介绍附近范围查询：
// 模板 GET /indexName/_search { "query": { "geo_distance": { // 表示附近范围查询，其实就是圆形查询 "distance": "15km", // 半径 "FIELD": "31.21,121.5" // 圆心（前面的FIELD需要修改为具体字段名，表示进行匹配） } } } // 案例 GET /hotel/_search { "query": { "geo_distance": { "distance": "15km", "Location": "31.21,121.5" // 圆心修改为Location的值，当距离小于15km时匹配成功 } } } 复合查询
最后我们介绍一下复合查询：
- 复合查询可以将其它简单查询组合起来，实现更复杂的搜索逻辑
复合查询通常被分为两种情况：
- fuction score：算分函数查询，可以控制文档相关性算分，控制文档排名
- bool query：布尔查询，利用逻辑关系组合多个其它的查询，实现复杂搜索
那么在正式介绍复合查询之前，我们需要先去了解一下文档相关性算分：
// 文档相关性算法：档结果会根据与搜索词条的关联度打分（_score），返回结果时按照分值降序排列 [ { "_score" : 17.850193, "_source" : { "name" : "虹桥如家酒店真不错", } }, { "_score" : 12.259849, "_source" : { "name" : "外滩如家酒店真不错", } }, { "_score" : 11.91091, "_source" : { "name" : "迪士尼如家酒店真不错", } } ] 而这种算分机制是由系统去控制，目前所延用的算分机制为BM25算分机制：
那么我们接下来就可以去了解算分函数查询：
// 我们平时所得出的score也许并不能完全满足我们的正常需求 // 例如：百度的广告通常会覆盖掉得分高的空间而被安排到最上层 /* 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的运行流程如下： - 1）根据**原始条件**查询搜索文档，并且计算相关性算分，称为**原始算分**（query score） - 2）根据**过滤条件**，过滤文档 - 3）符合**过滤条件**的文档，基于**算分函数**运算，得到**函数算分**（function score） - 4）将**原始算分**（query score）和**函数算分**（function score）基于**运算模式**做运算，得到最终结果，作为相关性算分。 */ // 我们给出一个简单模板去解释上述内容 GET /indexName/_search { "query": { "function_score": { "query": { .... }, // 原始查询，可以是任意条件 "functions": [ // 算分函数 { "filter": { // 满足的条件 "term": { // 这里假设采用精准查询匹配 "FIELD": "TEXT" // 需要满足FIELD字段为TEXT } }, "weight": N // 算分权重为N } ], "boost_mode": "???" // 原始权重和算分权重的算法：有相加，乘法等 } } } // 我们举一个简单的例子 GET /hotel/_search { "query": { "function_score": { "query": { "match":{ "all":"外滩" } }, // 原始查询，我们这里查询在all字段中带"外滩"的内容 "functions": [ // 算分函数，下面是内容函数 { "filter": { // 满足的条件，品牌必须是如家 "term": { "brand": "如家" } }, "weight": 2 // 算分权重为2 } ], "boost_mode": "sum" // 加权模式，求和 } } } 最后我们还需要介绍一个布尔查询：
// 布尔查询其实就是简单的&&，||查询 /* 布尔查询是一个或多个查询子句的组合，每一个子句就是一个**子查询**。子查询的组合方式有： - must：必须匹配每个子查询，类似“与” - should：选择性匹配子查询，类似“或” - must_not：必须不匹配，**不参与算分**，类似“非” - filter：必须匹配，**不参与算分** */ /* 需要注意的是，搜索时，参与**打分的字段越多，查询的性能也越差**。因此这种多条件查询时，建议这样做： - 搜索框的关键字搜索，是全文检索查询，使用must查询，参与算分 - 其它过滤条件，采用filter查询。不参与算分 */ // 我们首先给出一个模板 GET /indexName/_search { "query": { "bool": { // 表示开启bool复合查询 "must": [ // must必须满足 {"term": {"FIELD": "TEXT" }} ], "should": [// 选择性匹配子查询，类似“或” {"term": {"FIELD": "TEXT" }}, {"term": {"FIELD": "TEXT" }} ], "must_not": [// 不能满足， { "range": { "FIELD": { "lte": TEXT } }} ], "filter": [// 必须满足，但不参与到算分项目中 { "range": {"FIELD": { "gte": TEXT } }} ] } } } // 我们给出一个简单案例： GET /hotel/_search { "query": { "bool": { "must": [// 必须在上海 {"term": {"city": "上海" }} ], "should": [// 品牌必须是皇冠或华美达其中一种 {"term": {"brand": "皇冠假日" }}, {"term": {"brand": "华美达" }} ], "must_not": [// 价格不能低于500 { "range": { "price": { "lte": 500 } }} ], "filter": [// 得分必须高于45 { "range": {"score": { "gte": 45 } }} ] } } } 搜索结果处理
对于GET获得的结果我们还可以对其进行简单处理，其中大致包括有：
- 排序：对搜索结果进行排序操作
- 分页：对搜索结果进行分页操作
- 高亮：对搜索结果进行高亮操作
排序
ES默认是根据相关度算分来排序，但是也支持自定义方式对搜索结果排序，大致分为两种：
- 普通字段排序
- 地理坐标排序
我们首先来介绍普通字段排序：
// 普通字段包括有：keyword、数值、日期类型排序 // 模板 // 排序条件是一个数组，也就是可以写多个排序条件。按照声明的顺序，当第一个条件相等时，再按照第二个条件排序，以此类推 GET /indexName/_search { "query": { "match_all": {} }, "sort": [ { "FIELD": "desc" // 排序字段、排序方式ASC、DESC } ] } // 案例 GET /hotel/_search { "query": { "match_all": {} }, "sort": [ { "score"::"desc" } ,{ "price": "asc" } ] } 我们再来介绍地理坐标排序：
// 地理坐标排序：指定一个坐标，作为目标点，计算每一个文档中，指定字段的坐标 到目标点的距离是多少，根据距离排序 // 模板 GET /indexName/_search { "query": { "match_all": {} }, "sort": [ { "_geo_distance" : { "FIELD" : "纬度，经度", // 文档中geo_point类型的字段名、目标坐标点 "order" : "asc", // 排序方式 "unit" : "km" // 排序的距离单位 } } ] } // 案例 GET /hotel/_search { "query": { "match_all": {} }, "sort": [ { "_geo_distance" : { "location" : "31，121", // 坐标在(31,121)附近的酒店按距离排序 "order" : "asc", // 排序方式 "unit" : "km" // 排序的距离单位 } } ] } 分页
elasticsearch 默认情况下只返回top10的数据，如果希望返回更多只能采用分页模式，分页被划分为两种：
- 基本分页
- 深度分页
我们首先来介绍基本分页：
// 分页主要依赖两个参数：from和size，类似于mysql中的`limit ?, ?` // - from：从第几个文档开始 // - size：总共查询几个文档 // 模板 GET /indexName/_search { "query": { "match_all": {} }, "from": ?, // 分页开始的位置，默认为0 "size": ?, // 期望获取的文档总数 "sort": [ {"price": "asc"} // 排序方式 ] } // 案例 GET /hotel/_search { "query": { "match_all": {} }, "from": 0, // 分页开始的位置，默认为0 "size": 10, // 期望获取的文档总数 "sort": [ {"price": "asc"} ] } 然后我们再来介绍深度分页：
// 首先我们需要去了解一个思想，假设我们获取990~1000的数据，那么我们需要先去查询0~1000的数据然后去截取990~1000 GET /hotel/_search { "query": { "match_all": {} }, "from": 990, // 分页开始的位置，默认为0 "size": 10, // 期望获取的文档总数 "sort": [ {"price": "asc"} ] } // 如果是单点查询，那么我们可以只查找数据并排序截取就可以了 // 但如果集群查询，我们并非说只获取每个节点的TOP200就可以了，因为排序未定，我们需要获取每个节点的TOP1000再重新排序获取 // 就会导致所查询数据过多导致查询缓慢，ES服务器压力较大，因此elasticsearch会禁止from+ size 超过10000的请求 // 针对深度分页，ES提供了两种解决方案，[官方文档]： // - search after：分页时需要排序，原理是从上一次的排序值开始，查询下一页数据。官方推荐使用的方式。 // - scroll：原理将排序后的文档id形成快照，保存在内存。官方已经不推荐使用。 高亮
我们首先介绍一下高亮：
- 当我们在百度查询时，我们的查询词汇通常会在查询内容中高亮显示出来用来确定查询位置
高亮显示的实现分为两步：
- 给文档中的所有关键字都添加一个标签，例如
<em>标签 - 页面给
<em>标签编写CSS样式
我们来简单学习一下高亮：
// 模板 GET /indexName/_search { "query": { "match": { "FIELD": "TEXT" // 查询条件，高亮一定要使用全文检索查询 } }, "highlight": { "fields": { // 指定要高亮的字段 "FIELD": { "pre_tags": "<em>", // 用来标记高亮字段的前置标签 "post_tags": "</em>" // 用来标记高亮字段的后置标签 } } } } // 案例 GET /hotel/_search { "query": { "match": { "all": "如家" // 查询条件，高亮一定要使用全文检索查询 } }, "highlight": { "fields": { // 指定要高亮的字段 "name": { "require_field_match":"false",// 默认无法高亮，需要添加属性 "pre_tags": "<em>", // 用来标记高亮字段的前置标签 "post_tags": "</em>" // 用来标记高亮字段的后置标签 } } } } /* **注意：** - 高亮是对关键字高亮，因此**搜索条件必须带有关键字**，而不能是范围这样的查询。 - 默认情况下，**高亮的字段，必须与搜索指定的字段一致**，否则无法高亮 - 如果要对非搜索字段高亮，则需要添加一个属性：required_field_match=false */ IDEA数据搜索
下面我们来使用Java代码去操作ES完成数据搜索
快速入门
我们首先来简单学习一下使用流程：
- 发起查询请求
@Test void testMatchAll() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel");// 查询统一使用SearchRequest的请求 // 2.准备DSL request.source() .query(QueryBuilders.matchAllQuery()); // request.source()的方法中包含了所有方法：query,sort,size,from,highlighter等 // query()：代表查询条件，利用QueryBuilders.matchAllQuery()构建一个match_all查询的DSL // QueryBuilders：其中包含match、term、function_score、bool等各种查询 // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); } - 对响应进行处理
private void handleResponse(SearchResponse response) { // 4.解析响应 // 通过response.getHits()获取，就是JSON中的最外层的hits，代表命中的结果 SearchHits searchHits = response.getHits(); // 4.1.获取总条数 // `SearchHits#getTotalHits().value`：获取总条数信息 long total = searchHits.getTotalHits().value; System.out.println("共搜索到" + total + "条数据"); // 4.2.文档数组 // `SearchHits#getHits()`：获取SearchHit数组，也就是文档数组 SearchHit[] hits = searchHits.getHits(); // 4.3.遍历 for (SearchHit hit : hits) { // 获取文档source // `SearchHit#getSourceAsString()`：获取文档结果中的_source，也就是原始的json文档数据 String json = hit.getSourceAsString(); // 反序列化 HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class); System.out.println("hotelDoc = " + hotelDoc); } } - 整体代码展示
@Test void testMatchAll() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source() .query(QueryBuilders.matchAllQuery()); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 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); System.out.println("hotelDoc = " + hotelDoc); } } /* 整体步骤： 1. 创建SearchRequest对象 2. 准备Request.source()，也就是DSL。 ① QueryBuilders来构建查询条件 ② 传入Request.source() 的 query() 方法 3. 发送请求，得到结果 4. 解析结果（参考JSON结果，从外到内，逐层解析） */ match查询
我们首先来介绍match查询：
// 全文检索的match和multi_match查询与match_all的API基本一致,Java代码上的差异主要是request.source().query()中的参数 // match查询 @Test void testMatch() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source()// 这里第一个参数是字段，第二个参数是val匹配值 .query(QueryBuilders.matchQuery("all", "如家"));// QueryBuilders的matchQuery // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } // MultiMatch查询 @Test void testMultiMatch() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source()// 这里第一个参数是val匹配值，后面均为字段 .query(QueryBuilders.mutilMatchQuery("如家","brand","name","business"));// mutilMatchQuery方法 // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 精确查询
下面我们来介绍精准查询：
// 精准查询主要分为：term和range，和之前几乎相同，只是采用的API不同 // term查询 @Test void testMatch() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source()// 这里第一个参数是字段，第二个参数是val匹配值 .query(QueryBuilders.termQuery("city", "北京"));// QueryBuilders的termQuery // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } // range查询 @Test void testMatch() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL request.source()// rangeQuery获得对应QueryBuilders，后面采用gte和lte设置条件 .query(QueryBuilders.rangeQuery("price").gte(100).lte(150)); // 3.发送请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 布尔查询
下面我们来介绍布尔查询：
// 布尔查询是用must、must_not、filter等方式组合其它查询 @Test void testBool() throws IOException { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.准备BooleanQuery(由于内容过多，我们提前创建并对其设置，最后添加入request即可) BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); // 2.2.添加term（注意：内部使用的仍是Query，BooleanQuery仅仅是将这些Query结合起来） boolQuery.must(QueryBuilders.termQuery("city", "杭州")); // 2.3.添加range（注意：内部使用的仍是Query，BooleanQuery仅仅是将这些Query结合起来） 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()来设置 @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); // 按price排序，升序 // 2.3.分页 from、size request.source().from((page - 1) * size).size(5);// 分页查询 // 3.发送请求 SearchResponse response P= client.search(request, RequestOptions.DEFAULT); // 4.解析响应 handleResponse(response); } 高亮查询
最后我们介绍一下高亮查询：
// 高亮查询必须使用全文检索查询，并且要有搜索关键字，将来才可以对关键字高亮 // - 查询的DSL：其中除了查询条件，还需要添加高亮条件，同样是与query同级。 // - 结果解析：结果除了要解析_source文档数据，还要解析高亮结果 // 首先我们处理请求问题 @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); } // 我们还需要对高亮结果进行解析 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); } } ES数据聚合
下面我们来学习ES的数据聚合
数据聚合
首先我们需要先来了解数据聚合：
- 聚合可以让我们极其方便的实现对数据的统计、分析、运算
- ES实现这些统计功能的比数据库的sql要方便的多，而且查询速度非常快，可以实现近实时搜索效果
ES的聚合通常被分为三种：
- 桶（Bucket）聚合：用来对文档做分组
- 度量（Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等
- 管道（pipeline）聚合：其它聚合的结果为基础做聚合
注意：参加聚合的字段必须是keyword、日期、数值、布尔类型
桶聚合
首先我们先来了解基本的桶聚合：
- TermAggregation：按照文档字段值分组，例如按照品牌值分组、按照国家分组
- Date Histogram：按照日期阶梯分组，例如一周为一组，或者一月为一组
我们给出一个桶聚合的案例展示：
GET /hotel/_search { "size": 0, // 设置size为0，结果中不包含文档，只包含聚合结果 "aggs": { // 表示开始定义聚合 "brandAgg": { // 聚合名称，自定义即可 "terms": { // 聚合的类型，按照品牌值聚合，所以选择term（TermAggregation聚合） "field": "brand", // 参与聚合的字段，会根据brand品牌进行聚合 "size": 20 // 希望获取的聚合结果数量（默认情况为10，修改可展示数据数量） } } } } 默认情况下，Bucket聚合会统计Bucket内的文档数量，记为_count，并且按照_count降序排序，但是我们可以进行修改：
GET /hotel/_search { "size": 0, "aggs": { "brandAgg": { "terms": { "field": "brand", "order": { // 我们可以在aggs的对应聚合名称内设置order来修改排序方式 "_count": "asc" // 按照_count升序排列 }, "size": 20 } } } } 我们同样可以采用数据搜索的方式来限制聚合的范围大小：
// 很多情况下，我们并非需要聚合所有的数据，而是聚合满足一定条件的数据，那么我们就需要设置限制条件 GET /hotel/_search { // 实际上就是采用最简单的query方法来限制条件 "query": { "range": { "price": { "lte": 200 // 只对200元以下的文档聚合 } } }, "size": 0, // 这里仍旧采用aggs即可 "aggs": { "brandAgg": { "terms": { "field": "brand", "size": 20 } } } } 度量聚合
我们再来介绍一下度量聚合：
- 度量（Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等
- 度量聚合通常是管道聚合，因为度量聚合的数据计算通常是建立在一层数据聚合之后
度量聚合通常会分为四种情况：
- Avg：求平均值
- Max：求最大值
- Min：求最小值
- Stats：同时求max、min、avg、sum等
我们这里同样给出一个简单案例展示：
GET /hotel/_search { // 设置展示数据，设置为0，不允许出聚合数据外的数据展示 "size": 0, // 第一层数据聚合，这层会按brand进行分组并且聚合展示 "aggs": { "brandAgg": { "terms": { "field": "brand", "size": 20 }, // 第二层数据聚合，实际上也是管道聚合！！！ "aggs": { // 是brands聚合的子聚合，也就是分组后对每组分别计算 "score_stats": { // 聚合名称 "stats": { // 聚合类型，这里stats可以计算min、max、avg等（也可以单写min、max、avg其中一种） "field": "score" // 聚合字段，这里是score } } } } } } IDEA数据聚合
下面我们来用Java代码来实现数据聚合
API分析
我们下面会从两方面分别将DSL语句和Java语句进行对比分析：
- 请求信息设置
- 响应数据设置
数据聚合案例
我们将通过一个简单的数据聚合案例来介绍具体API使用：
- 我们希望从ES数据中搜索对应的数据，并将这些数据组合成数组返回到前端进行展示
- 我们希望从ES数据中搜索酒店使用量最多的城市，星级，品牌并进行处理，将其返回到前端页面展示
- 同时我们还需要注意我们的搜索存在一个搜索框，我们所获得的聚合信息必须要满足搜索框条件，也就是Query
下面我们来具体实现其效果：
- 控制层书写Controller
/* 相关前端请求信息： - 请求方式：`POST` - 请求路径：`/hotel/filters` - 请求参数：`RequestParams`，与搜索文档的参数一致 - 返回值类型：`Map<String, List<String>>` */ @PostMapping("filters") public Map<String, List<String>> getFilters(@RequestBody RequestParams params){ return hotelService.getFilters(params); } - 服务层接口Service
Map<String, List<String>> filters(RequestParams params); - 服务层具体实现ServiceImpl
@Override public Map<String, List<String>> filters(RequestParams params) { try { // 1.准备Request SearchRequest request = new SearchRequest("hotel"); // 2.准备DSL // 2.1.query buildBasicQuery(params, request); // 2.2.设置size request.source().size(0); // 2.3.聚合 buildAggregation(request); // 3.发出请求 SearchResponse response = client.search(request, RequestOptions.DEFAULT); // 4.解析结果 Map<String, List<String>> result = new HashMap<>(); Aggregations aggregations = response.getAggregations(); // 4.1.根据品牌名称，获取品牌结果 List<String> brandList = getAggByName(aggregations, "brandAgg"); result.put("品牌", brandList); // 4.2.根据品牌名称，获取品牌结果 List<String> cityList = getAggByName(aggregations, "cityAgg"); result.put("城市", cityList); // 4.3.根据品牌名称，获取品牌结果 List<String> starList = getAggByName(aggregations, "starAgg"); result.put("星级", starList); return result; } catch (IOException e) { throw new RuntimeException(e); } } private void buildAggregation(SearchRequest request) { request.source().aggregation(AggregationBuilders .terms("brandAgg") .field("brand") .size(100) ); request.source().aggregation(AggregationBuilders .terms("cityAgg") .field("city") .size(100) ); request.source().aggregation(AggregationBuilders .terms("starAgg") .field("starName") .size(100) ); } private List<String> getAggByName(Aggregations aggregations, String aggName) { // 4.1.根据聚合名称获取聚合结果 Terms brandTerms = aggregations.get(aggName); // 4.2.获取buckets List<? extends Terms.Bucket> buckets = brandTerms.getBuckets(); // 4.3.遍历 List<String> brandList = new ArrayList<>(); for (Terms.Bucket bucket : buckets) { // 4.4.获取key String key = bucket.getKeyAsString(); brandList.add(key); } return brandList; } MQ数据同步
最后我们来介绍ES和MySQL数据同步的具体实现
数据同步问题
首先我们需要明白为什么要实现数据同步：
- elasticsearch中的酒店数据来自于mysql数据库
- 因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步
数据同步的实现具体来说有三种方式：
- 同步调用
- hotel-demo对外提供接口，用来修改elasticsearch中的数据
- 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口
- 异步通知
- hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息
- hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改
- 监听binlog
- 给mysql开启binlog功能
- mysql完成增、删、改操作都会记录在binlog中
- hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容
但是不同的方式存在有不同的优缺点：
- 同步方式：实现简单，粗暴但业务耦合度高
- 异步方式：低耦合，实现难度一般但依赖于MQ的可靠性
- 监听方式：完全解除服务间耦合但开启binlog增加数据库负担、实现复杂度高
MQ实现数据同步
在ES和MySQL的数据同步问题上异步方式在一定程度优于同步方式且我们之前已经学习过MQ，所以这里采用MQ实现数据同步
在实现数据同步之前我们先来简单介绍一下具体项目内容：
- hotel-admin：宾馆项目的后端处理服务，内部封装了针对hotel的MySQL的数据处理
- hotel-demo：宾馆项目的后端处理服务，内部封装了针对hotel的ES数据处理
下面让我们来逐步完成MQ数据同步操作：
- 思索整体MQ框架
- 引入依赖
<!--注意：在hotel-admin、hotel-demo中引入rabbitmq的依赖--> <!--amqp--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> - 声明队列交换机名称
// 为了保证两个服务的交换机机制相同，我们在两个服务中都声明以下类，采用常量去定义具体交换机和队列以及key package cn.itcast.hotel.constatnts; 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中声明即可 package cn.itcast.hotel.config; import cn.itcast.hotel.constants.MqConstants; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @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的mysql操作中顺便发送MQ信息 @PostMapping public void saveHotel(@RequestBody Hotel hotel){ // 服务层具体实现 hotelService.save(hotel); // 我们主要看这部分，MQ的信息发送(这里仅发送id为了节省MQ内存) rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,OTEL_INSERT_KEY,hotel.getId()); } @PostMapping public void updateHotel(@RequestBody Hotel hotel){ // 服务层具体实现 hotelService.update(hotel); // 由于ES的新增和更新相同，所以这里采用同一个key rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,OTEL_INSERT_KEY,hotel.getId()); } @PostMapping public void deleteHotel(@PathVariable Long id){ // 服务层具体实现 hotelService.delete(hotel); // 这里发送不同的key，进入不同Listener rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,OTEL_DELETE_KEY,id); } - 接收MQ并实现具体逻辑
// 首先在hotel-demo的`cn.qiuluo.hotel.service`包下的`IHotelService`中新增新增、删除业务 void deleteById(Long id); void insertById(Long id); // 给hotel-demo中的`cn.qiuluo.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); } } // 监听类书写 package cn.qiuluo.hotel.mq; import cn.qiuluo.hotel.constants.MqConstants; import cn.qiuluo.hotel.service.IHotelService; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @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以及关于ES的相关API展示，希望能为你带来帮助
这里推荐一篇ElasticSearch的非常详细的博客文章，为我带来很多帮助：Elasticsearch学习笔记_巨輪的博客-CSDN博客
附录
该文章属于学习内容，具体参考B站黑马程序员的微服务课程
这里附上视频链接：01-今日内容介绍7_哔哩哔哩_bilibili
