Lucene&ElasticSeach&Kafka

Lucene&ElasticSeach

1 什么是全文检索

1.1 数据分类

生活中的数据总体分为两种：结构化数据和非结构化数据。

结构化数据：指具有固定格式或有限长度的数据，如数据库，元数据等

非结构化数据：指不定长或无固定格式的数据，如邮件，word 文档等磁盘上的文件

1.2 结构化数据搜索

常见的结构化数据也就是数据库中的数据

在数据库中搜索很容易实现，通常都是使用 sql语句进行查询，而且能很快的得到查询结果。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Xu0ldly8-1637500558715)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–什么是全文检索–结构化数据搜索)]

为什么数据库搜索很容易？

因为数据库中的数据存储是有规律的，有行有列而且数据格式、数据长度都是固定的。

1.3 非结构化数据查询方法

（1）顺序扫描法(Serial Scanning)

用户搜索----->文件

所谓顺序扫描，比如要找内容包含某一个字符串的文件，就是一个文档一个文档的看，对于每一个文档，从头看到尾，如果此文档包含此字符串，则此文档为我们要找的文件，接着看下一个文件，直到扫描完所有的文件。如利用 windows 的搜索也可以搜索文件内容，只是相当的慢。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qFUPn7LR-1637500558732)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–什么是全文检索-- 非结构化数据查询方法–顺序扫描法(Serial Scanning)])

（2）全文检索(Full-text Search)

用户通过查询索引库---->生成索引----->文档

全文检索是指计算机索引程序通过扫描文章中的每一个词，对每一个词建立一个索引，指明该词在文章中出现的次数和位置，当用户查询时，检索程序就根据事先建立的索引进行查找，并将查找的结果反馈给用户的检索方法。这个过程类似于通过字典的目录查字的过程。

将非结构化数据中的一部分信息提取出来，重新组织，使其变得有一定结构，然后对此有一定结构的数据进行搜索，从而达到搜索相对较快的目的。这部分从非结构化数据中提取出的然后重新组织的信息，我们称之索引。

例如：字典。字典的拼音表和部首检字表就相当于字典的索引，对每一个字的解释是非结构化的，如果字典没有音节表和部首检字表，在茫茫辞海中找一个字只能顺序扫描。然而字的某些信息可以提取出来进行结构化处理，比如读音，就比较结构化，分声母和韵母，分别只有几种可以一一列举，于是将读音拿出来按一定的顺序排列，每一项读音都指向此字的详细解释的页数。我们搜索时按结构化的拼音搜到读音，然后按其指向的页数，便可找到我们的非结构化数据——也即对字的解释。

这种先建立索引，再对索引进行搜索的过程就叫全文检索(Full-Text Search) 。虽然创建索引的过程也是非常耗时的，但是索引一旦创建就可以多次使用，全文检索主要处理的是查询，所以耗时间创建索引是值得的。

建立索引

检索索引

1.4 如何实现全文检索

可以使用 Lucene 实现全文检索。Lucene 是 apache 下的一个开放源代码的全文检索引擎工具包。提供了完整的查询引擎和索引引擎，部分文本分析引擎（英文与德文两种西方语言）。Lucene 的目的是为软件开发人员提供一个简单易用的工具包，以方便的在目标系统中实现全文检索的功能。

Lucene适用场景：

在应用中为数据库中的数据提供全文检索实现。

开发独立的搜索引擎服务、系统

Lucene的特性：

1.稳定、索引性能高

每小时能够索引150GB以上的数据

对内存的要求小，只需要1MB的堆内存

增量索引和批量索引一样快

索引的大小约为索引文本大小的20%~30%

2.高效、准确、高性能的搜索算法

良好的搜索排序

强大的查询方式支持：短语查询、通配符查询、临近查询、范围查询等

支持字段搜索（如标题、作者、内容）可根据任意字段排序

支持多个索引查询结果合并

支持更新操作和查询操作同时进行

支持高亮、join、分组结果功能

速度快

可扩展排序模块，内置包含向量空间模型、BM25模型可选

可配置存储引擎

3.跨平台

纯java编写

作为Apache开源许可下的开源项目，你可以在商业或开源项目中使用

Lucene有多种语言实现版（如C，C++、Python等），不仅仅是JAVA

Lucene架构：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nQ1BPJt6-1637500558737)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene架构)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QxIpw7EC-1637500558742)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene架构02)]

1.5 全文检索的应用场景

对于数据量大、数据结构不固定的数据可采用全文检索方式搜索。

单机软件的搜索：word、markdown

站内搜索：京东、淘宝、拉勾，索引源是数据库

搜索引擎：百度、Google，索引源是爬虫程序抓取的数据

2 Lucene 实现全文检索的流程说明

2.1 索引和搜索流程图

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bVL5YbYY-1637500558744)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明–索引和搜索流程图)]

1、绿色表示索引过程，对要搜索的原始内容进行索引构建一个索引库，索引过程包括：

确定原始内容即要搜索的内容–>采集文档–>创建文档–>分析文档–>索引文档

2、红色表示搜索过程，从索引库中搜索内容，搜索过程包括：

用户通过搜索界面–>创建查询–>执行搜索，从索引库搜索–>渲染搜索结果

2.2 创建索引

核心概念：

Document：

用户提供的源是一条条记录，它们可以是文本文件、字符串或者数据库表的一条记录等等。一条记录经过索引之后，就是以一个Document的形式存储在索引文件中的。用户进行搜索，也是以Document列表的形式返回。

Field：

一个Document可以包含多个信息域，例如一篇文章可以包含“标题”、“正文”、“最后修改时间”等信息域，这些信息域就是通过Field在Document中存储的。

Field有两个属性可选：存储和索引。通过存储属性你可以控制是否对这个Field进行存储；通过索引属性你可以控制是否对该Field进行索引。

如果对标题和正文进行全文搜索，所以我们要把索引属性设置为真，同时我们希望能直接从搜索结果中提取文章标题，所以我们把标题域的存储属性设置为真，但是由于正文域太大了，我们为了缩小索引文件大小，将正文域的存储属性设置为假，当需要时再直接读取文件；我们只是希望能从搜索解果中提取最后修改时间，不需要对它进行搜索，所以我们把最后修改时间域的存储属性设置为真，索引属性设置为假。上面的三个域涵盖了两个属性的三种组合，还有一种全为假的没有用到，事实上Field不允许你那么设置，因为既不存储又不索引的域是没有意义的。

Term：

Term是搜索的最小单位，它表示文档的一个词语，Term由两部分组成：它表示的词语和这个词语所出现的Field的名称。

我们以拉勾招聘网站的搜索为例，在网站上输入关键字搜索显示的内容不是直接从数据库中来的，而是从索引库中获取的，网站的索引数据需要提前创建的。以下是创建的过程：

第一步：获得原始文档：就是从mysql数据库中通过sql语句查询需要创建索引的数据

第二步：创建文档对象（Document），把查询的内容构建成lucene能识别的Document对象，获取原始内容的目的是为了索引，在索引前需要将原始内容创建成文档，文档中包括一个一个的域（Field），这个域对应就是表中的列。

注意：每个Document可以有多个Field，不同的Document可以有不同的Field，同一个Document可以有相同的 Field（域名和域值都相同）。每个文档都有一个唯一的编号，就是文档id。

第三步：分析文档

将原始内容创建为包含域（Field）的文档（document），需要再对域中的内容进行分析，分析的过程是经过对原始文档提取单词、将字母转为小写、去除标点符号、去除停用词等过程生成最终的语汇单元，可以将语汇单元理解为一个一个的单词。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-saLmjNAv-1637500558747)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明–索引和搜索流程图–创建索引)]

分好的词会组成索引库中最小的单元：term，一个term由域名和词组成

第四步：创建索

对所有文档分析得出的语汇单元进行索引，索引的目的是为了搜索，最终要实现只搜索被索引的语汇单元从而找到 Document（文档）。

注意：创建索引是对语汇单元索引，通过词语找文档，这种索引的结构叫倒排索引结构。倒排索引结构是根据内容（词语）找文档，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-L9Jp0bT7-1637500558750)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明–索引和搜索流程图–创建索引02)]

倒排索引结构也叫反向索引结构，包括索引和文档两部分，索引即词汇表，它的规模较小，而文档集合较大。

2.3 倒排索引

倒排索引记录每个词条出现在哪些文档，及在文档中的位置，可以根据词条快速定位到包含这个词条的文档及出现的位置。

文档：索引库中的每一条原始数据，例如一个商品信息、一个职位信息

词条：原始数据按照分词算法进行分词，得到的每一个词

创建倒排索引，分为以下几步：

1）创建文档列表：

lucene首先对原始文档数据进行编号（DocID），形成列表，就是一个文档列表

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LA4AgeRJ-1637500558753)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明–索引和搜索流程图–倒排索引–创建文档列表)]

2）创建倒排索引列表

对文档中数据进行分词，得到词条（分词后的一个又一个词）。对词条进行编号，以词条创建索引。然后记录下包含该词条的所有文档编号（及其它信息）。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WUA9vnFC-1637500558755)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明–索引和搜索流程图–倒排索引–创建倒排索引列表)]

3）搜索的过程：

当用户输入任意的词条时，首先对用户输入的数据进行分词，得到用户要搜索的所有词条，然后拿着这些词条去倒排索引列表中进行匹配。找到这些词条就能找到包含这些词条的所有文档的编号。然后根据这些编号去文档列表中找到文档

2.4 查询索引

查询索引也是搜索的过程。搜索就是用户输入关键字，从索引（index）中进行搜索的过程。根据关键字搜索索引，根据索引找到对应的文档

第一步：创建用户接口：用户输入关键字的地方

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-i94qKT4u-1637500558757)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明-- 查询索引01)]

第二步：创建查询指定查询的域名和关键字

第三步：执行查询

第四步：渲染结果（结果内容显示到页面上关键字需要高亮）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-j62sKdwj-1637500558758)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach-- Lucene 实现全文检索的流程说明-- 查询索引02)]

3 Lucene实战

3.1 需求说明

生成职位信息索引库，从索引库检索数据

创建数据库es，将sql脚本导入数据库执行

3.2 准备开发环境

第一步：创建一个maven工程 ,已经学过Spring Boot，我们就创建一个SpringBoot项目

第二步：导入依赖

<!--spring boot 父启动器依赖-->
<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.6.RELEASE</version>
</parent>
<dependencies><!--web依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--测试依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!--lombok工具--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.4</version><scope>provided</scope></dependency><!--热部署--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><optional>true</optional></dependency><!--mybatis-plus--><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.3.2</version></dependency><!--pojo持久化使用--><dependency><groupId>javax.persistence</groupId><artifactId>javax.persistence-api</artifactId><version>2.2</version></dependency><!--mysql驱动--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><scope>runtime</scope></dependency><!--引入Lucene核心包及分词器包--><dependency><groupId>org.apache.lucene</groupId><artifactId>lucene-core</artifactId><version>4.10.3</version></dependency><dependency><groupId>org.apache.lucene</groupId><artifactId>lucene-analyzers-common</artifactId><version>4.10.3</version></dependency>
</dependencies>
<build><plugins><!--编译插件--><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>11</source><target>11</target><encoding>utf-8</encoding></configuration></plugin><!--打包插件--><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><executions><execution><goals><goal>repackage</goal></goals></execution></executions></plugin></plugins>
</build>

第三步：创建引导类

@SpringBootApplication
@MapperScan("com.lagou.lucene.mapper")
public class LuceneApplication {public static void main(String[] args) {SpringApplication.run(LuceneApplication.class, args);}}

第四步：配置properties文件

server:port: 9000
spring:application:name: lagou-lucenedatasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/es?useUnicode=true&characterEncoding=utf8&serverTimezone=UTCusername: rootpassword: 123456#开启驼峰命名匹配映射
mybatis-plus:configuration:map-underscore-to-camel-case: true

第五步：创建实体类、mapper、service

实体类

@Data
@Table(name = "job_info")
public class JobInfo {@Idprivate long id;private String companyName;private String companyAddr;private String companyInfo;private String jobName;private String jobAddr;private String jobInfo;private int salaryMin;private int salaryMax;private String url;private String time;
}

mapper

// 继承BaseMapper
public interface JobInfoMapper extends BaseMapper<JobInfo> {
}

service （只是测试，这个可以省略）

@Service
public class JobInfoServiceImpl implements JobInfoService {@Autowiredprivate JobInfoMapper jobInfoMapper;@Overridepublic JobInfo selectById(long id) {return jobInfoMapper.selectById(id);}@Overridepublic List<JobInfo> selectAll() {QueryWrapper<JobInfo> queryWrapper = new QueryWrapper<>();List<JobInfo> jobInfoList = jobInfoMapper.selectList(queryWrapper);return jobInfoList;}
}

3.3创建索引

在test下创建一个包com.lucene

@RunWith(SpringRunner.class)
@SpringBootTest
class LuceneIndexTest {@Autowiredprivate JobInfoService jobInfoService;/*** 创建索引*/@Testvoid create() throws Exception {// 1.指定索引文件存储的位置，索引具体的表现形式就是一组有规则的文件Directory directory = FSDirectory.open(new File("E:/class/index"));// 2.配置版本及其分词器//Analyzer analyzer = new StandardAnalyzer();Analyzer analyzer = new StandardAnalyzer();IndexWriterConfig config = new IndexWriterConfig(Version.LATEST, analyzer);// 3.创建IndexWriter对象，作用就是创建索引IndexWriter indexWriter = new IndexWriter(directory, config);// 先删除已经存在的索引库indexWriter.deleteAll();// 4.获得索引源/原始数据List<JobInfo> jobInfoList = jobInfoService.selectAll();// 5.遍历jobInfoList，每次遍历创建一个Document对象for (JobInfo jobInfo : jobInfoList) {// 创建Document对象Document document = new Document();// 创建Field对象，添加到document对象中document.add(new LongField("id",jobInfo.getId(), Field.Store.YES));// 且分词、索引、存储document.add(new TextField("companyName",jobInfo.getCompanyName(),Field.Store.YES));document.add(new TextField("companyAddr",jobInfo.getCompanyAddr(),Field.Store.YES));document.add(new TextField("companyInfo",jobInfo.getCompanyInfo(),Field.Store.YES));document.add(new TextField("jobName",jobInfo.getJobName(),Field.Store.YES));document.add(new TextField("jobAddr",jobInfo.getJobAddr(),Field.Store.YES));document.add(new TextField("jobInfo",jobInfo.getJobInfo(),Field.Store.YES));document.add(new IntField("salaryMin",jobInfo.getSalaryMin(),Field.Store.YES));document.add(new IntField("salaryMax",jobInfo.getSalaryMax(),Field.Store.YES));document.add(new StringField("url",jobInfo.getUrl(),Field.Store.YES));// 将文档追加到索引库中indexWriter.addDocument(document);}// 关闭资源indexWriter.close();System.out.println("create index success!");}}

生成的索引目录：E:\class\index

索引(Index)：

在Lucene中一个索引是放在一个文件夹中的。

如下图，同一文件夹中的所有的文件构成一个Lucene索引。

段(Segment)：

按层次保存了从索引，一直到词的包含关系：索引(Index) –> 段(segment) –> 文档 (Document) –> 域(Field) –> 词(Term)
也即此索引包含了那些段，每个段包含了那些文档，每个文档包含了那些域，每个域包含了那些词
一个索引可以包含多个段，段与段之间是独立的，添加新文档可以生成新的段，不同的段可以合并
如上图，具有相同前缀文件的属同一个段，图中共一个段 “_0”
segments.gen和segments_1是段的元数据文件，也即它们保存了段的属性信息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ewNukbuO-1637500558761)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene实战–创建索引)]

Field的特性：

Document(文档)是Field(域)的承载体, 一个Document由多个Field组成。Field由名称和值两部分组成。Field的值是要索引的内容, 也是要搜索的内容.

是否分词(tokenized)

是：将Field的值进行分词处理，分词的目的是为了索引。如: 商品名称, 商品描述。这些内容用户会通过输入关键词进行查询，由于内容多样，需要进行分词处理建立索引。

否: 不做分词处理。如：订单编号，身份证号，是一个整体，分词以后就失去了意义，故不需要分词。

是否索引(indexed)

是：将Field内容进行分词处理后得到的词(或整体Field内容)建立索引，存储到索引域。索引的目的是为了搜索。如: 商品名称，商品描述需要分词建立索引。订单编号，身份证号作为整体建立索引。只要可能作为用户查询条件的词，都需要索引。

否：不索引。如：商品图片路径，不会作为查询条件，不需要建立索引。

是否存储(stored)

是：将Field值保存到Document中。如：商品名称，商品价格。凡是将来在搜索结果页面展现给用户的内容，都需要存储。

否：不存储。如：商品描述。内容多格式大，不需要直接在搜索结果页面展现，不做存储。需要的时候可以从关系数据库取。

常用的Field类型：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DwGxrgg5-1637500558763)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene实战–常用的Field类型)]

3.4查询索引

/*** 查询索引* @throws Exception*/
@Test
public void query() throws Exception {// 1.指定索引文件存储的位置，索引具体的表现形式就是一组有规则的文件Directory directory = FSDirectory.open(new File("E:/class/index"));// 2.IndexReader对象DirectoryReader indexReader = DirectoryReader.open(directory);// 3.创建查询对象，IndexSearcherIndexSearcher indexSearcher = new IndexSearcher(indexReader);// 4.使用term查询，查询公司名称中包含“北京”的所有的文档对象TermQuery query = new TermQuery(new Term("companyName","北"));TopDocs topDocs = indexSearcher.search(query, 100);// 5.获得符合查询条件的文档数int totalHits = topDocs.totalHits;System.out.println("符合条件的文档数：" + totalHits);// 6.获得命中的文档 ScoreDoc封装了文档id信息ScoreDoc[] scoreDocs = topDocs.scoreDocs;for (ScoreDoc scoreDoc : scoreDocs) {// 文档idint docId = scoreDoc.doc;// 通过文档id获得文档对象Document doc = indexSearcher.doc(docId);System.out.println("id："+doc.get("id"));System.out.println("companyName："+doc.get("companyName"));System.out.println("companyAddr："+doc.get("companyAddr"));System.out.println("companyInfo："+doc.get("companyInfo"));System.out.println("jobName："+doc.get("jobName"));System.out.println("jobInfo："+doc.get("jobInfo"));System.out.println("********************************************");}// 释放资源indexReader.close();
}

查看结果你会发现，居然没有数据，如果把查询的关键字“北京”那里改为“北”或“京”就可以，原因是因为中文会一个字一个字的分词，显然是不合适的，所以我们需要使用可以合理分词的分词器，其中最有名的是IKAnalyzer分词器

3.5中文分词器的使用

使用方式：

第一步：导依赖

<!--IK中文分词器-->
<dependency><groupId>com.janeluo</groupId><artifactId>ikanalyzer</artifactId><version>2012_u6</version>
</dependency>

第二步：可以添加配置文件

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-trMQongl-1637500558765)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene实战–中文分词器的使用01)]

放入到resources文件夹中。

第三步创建索引时使用IKanalyzer

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-u0KwYgua-1637500558767)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Lucene实战–中文分词器的使用02)]

把原来的索引数据删除，再重新生成索引文件，再使用关键字“北京”就可以查询到结果了

考虑一个问题：一个大型网站中的索引数据会很庞大的，所以使用lucene这种原生的写代码的方式就不合适了，所以需要借助一个成熟的项目或软件来实现，目前比较有名是solr和elasticSearch，所以接下来我们学习elasticSearch的使用。

4 Elastic search介绍和安装

Elasticsearch是一个需要安装配置的软件

ELK技术栈说明

Elastic有一条完整的产品线：Elasticsearch、Logstash、Kibana等，前面说的三个就是大家常说的ELK技术栈（开源实时日志分析平台）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-q7Gu6kGc-1637500558770)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–ELK技术栈说明)]

Logstash 的作用就是一个数据收集器，将各种格式各种渠道的数据通过它收集解析之后格式化输出到 Elasticsearch ，最后再由Kibana提供的比较友好的Web界面进行汇总、分析、搜索

ELK 内部实际就是个管道结构，数据从Logstash到Elasticsearch再到Kibana做可视化展示。这三个组件各自也可以单独使用，比如Logstash不仅可以将数据输出到Elasticsearch ，也可以到数据库、缓存等

4.1.简介

4.1.1.Elastic

Elastic官网：https://www.elastic.co/cn/

Elastic有一条完整的产品线：Elasticsearch、Logstash、Kibana等，前面说的三个就是大家常说的ELK技术栈

4.1.2.Elasticsearch

Elasticsearch官网：https://www.elastic.co/cn/products/elasticsearch

功能：

分布式的搜索引擎：百度、Google、站内搜索

全文检索：提供模糊搜索等自动度很高的查询方式，并进行相关性排名，高亮等功能

数据分析引擎（分组聚合）：电商网站—一周内手机销量Top10

对海量数据进行近乎实时处理：水平扩展，每秒钟可处理海量事件，同时能够自动管理索引和查询在集群中的分布方式，以实现极其流畅的操作。

如上所述，Elasticsearch具备以下特点：

高速、扩展性、最相关的搜索结果

分布式：节点对外表现对等，每个节点都可以作为入门，加入节点自动负载均衡
JSON：输入输出格式是JSON
Restful风格，一切API都遵循Rest原则，容易上手
近实时搜索，数据更新在Elasticsearch中几乎是完全同步的，数据检索近乎实时
安装方便：没有其它依赖，下载后安装很方便，简单修改几个参数就可以搭建集群
支持超大数据：可以扩展到PB级别的结构化和非结构化数据

4.1.3.版本

目前Elasticsearch最新的版本是7.x，企业内目前用的比较多是6.x，我们以6.2.4进行讲解，需要JDK1.8及以上

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1KThRSZ5-1637500558772)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–.版本)]

4.2.安装和配置

为了快速看到效果我们直接在本地window下安装Elasticsearch。环境要求：JDK8及以上版本

第一步：把今天资料文件夹中准备好的软件放到一个没有中文没有空格的位置，解压即可\

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-IOkzEQK2-1637500558774)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–.安装和配置01)]

第二步：修改配置文件

1、修改索引数据和日志数据存储的路径

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PViXh1CN-1637500558776)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–.安装和配置02)]

第33行和37行，修改完记得把注释打开

path.data: d:\class\es\data
#
# Path to log files:
#
path.logs: d:\class\es\log

第三步：进入bin目录中直接双击后缀名为.bat的命令文件

如果启动失败（估计好多同学都会启动失败的），需要修改虚拟机内存的大小

找到jvm.options文件如图修改

Xms 是指设定程序启动时占用内存大小。一般来讲，大点，程序会启动的快一点，但是也可能会导致机器暂时间变慢。

Xmx 是指设定程序运行期间最大可占用的内存大小。如果程序运行需要占用更多的内存，超出了这个设置值，就会抛出OutOfMemory异常。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LWFcDDlk-1637500558777)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–.安装和配置03)]

4.3.访问

启动后台输出如下

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-XGx7zIwZ-1637500558779)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–访问01)]

可以看到绑定了两个端口:

9300：集群节点间通讯接口，接收tcp协议

9200：客户端访问接口，接收Http协议

我们在浏览器中访问：http://127.0.0.1:9200

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mq86T8Xp-1637500558781)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–访问02)]

4.4.安装kibana

4.4.1.什么是Kibana

Kibana是一个基于Node.js的Elasticsearch索引库数据统计工具，可以利用Elasticsearch的聚合功能，生成各种图表，如柱形图，线状图，饼图等。

而且还提供了操作Elasticsearch索引数据的控制台，并且提供了一定的API提示，非常有利于我们学习 Elasticsearch的语法。

4.4.2.安装

因为Kibana依赖于node，需要在windows下先安装Node.js，双击运行课前资料提供的node.js的安装包：

node-v10.15.0-x64.msi

一路下一步即可安装成功，然后在任意DOS窗口输入名：

node -v

可以查看到node版本，如下：

v10.15.0

然后安装kibana，最新版本与elasticsearch保持一致，也是6.2.4

解压即可

4.4.3.配置运行

进入安装目录下的config目录，修改kibana.yml文件的第21行（注释放开即可）：

修改elasticsearch服务器的地址：

elasticsearch.url: "http://127.0.0.1:9200"

双击运行：kibana.bat

发现kibana的监听端口是5601

我们访问：http://127.0.0.1:5601

4.4.4.控制台

选择左侧的DevTools菜单，即可进入控制台页面：

在页面右侧，我们就可以输入请求，访问Elasticsearch了。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LP28gNwW-1637500558785)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–安装kibana–控制台)]

4.5.安装ik分词器

Lucene的IK分词器早在2012年已经没有维护了，现在我们要使用的是在其基础上维护升级的版本，并且开发为Elasticsearch的集成插件了，与Elasticsearch一起维护升级，版本也保持一致

https://github.com/medcl/elasticsearch-analysis-ik

4.5.1.安装

1.解压elasticsearch-analysis-ik-6.2.4.zip后，将解压后的文件夹拷贝到elasticsearch-6.2.4\plugins下，并重命名文件夹为ik

2.重新启动ElasticSearch，即可加载IK分词器

4.5.2.测试

大家先不管语法，我们先测试一波。

在kibana控制台输入下面的请求：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZTJJMNA6-1637500558786)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elastic search介绍和安装–安装ik分词器)]

GET /_analyze
{"analyzer": "ik_max_word","text": "我是中国人"
}

运行得到结果：

{"tokens": [{"token": "我","start_offset": 0,"end_offset": 1,"type": "CN_CHAR","position": 0},{"token": "是","start_offset": 1,"end_offset": 2,"type": "CN_CHAR","position": 1},{"token": "中国人","start_offset": 2,"end_offset": 5,"type": "CN_WORD","position": 2},{"token": "中国","start_offset": 2,"end_offset": 4,"type": "CN_WORD","position": 3},{"token": "国人","start_offset": 3,"end_offset": 5,"type": "CN_WORD","position": 4}]
}

GET /_analyze
{"analyzer": "ik_smart","text": "我是中国人"
}

结果：

{"tokens": [{"token": "我","start_offset": 0,"end_offset": 1,"type": "CN_CHAR","position": 0},{"token": "是","start_offset": 1,"end_offset": 2,"type": "CN_CHAR","position": 1},{"token": "中国人","start_offset": 2,"end_offset": 5,"type": "CN_WORD","position": 2}]
}

4.6 安装Head插件

4.6.1 elasticsearch-head 简介

elasticsearch-head是一个界面化的集群操作和管理工具，可以对集群进行傻瓜式操作。你可以通过插件把它集成到es（首选方式）,也可以安装成一个独立webapp。

es-head主要有三个方面的操作：

1.显示集群的拓扑，并且能够执行索引和节点级别操作

2.搜索接口能够查询集群中原始json或表格格式的检索数据

3.能够快速访问并显示集群的状态

官方的文档： https://github.com/mobz/elasticsearch-head

4.6.2 elasticsearch-head 安装（基于谷歌浏览器）

（1）直接下载压缩包，地址：https://files.cnblogs.com/files/sanduzxcvbnm/elasticsearch-head.7z

（2）解压

（3）在谷歌浏览器中点击“加载已解压的压缩程序”，找到elasticsearch-head文件夹，点击打开即可进行安装。

5 使用kibana对索引库操作

5.1.基本概念

5.1.1.节点、集群、分片及副本

1、节点（node）

一个节点是一个Elasticsearch的实例。

在服务器上启动Elasticsearch之后，就拥有了一个节点。如果在另一台服务器上启动Elasticsearch，这就是另一个节点。甚至可以通过启动多个Elasticsearch进程，在同一台服务器上拥有多个节点。

2、集群（cluster）

多个协同工作的Elasticsearch节点的集合被称为集群。

在多节点的集群上，同样的数据可以在多台服务器上传播。这有助于性能。这同样有助于稳定性，如果每个分片至少有一个副本分片，那么任何一个节点宕机后，Elasticsearch依然可以进行服务，返回所有数据。

但是它也有缺点：必须确定节点之间能够足够快速地通信，并且不会产生脑裂效应（集群的2个部分不能彼此交流，都认为对方宕机了）。

3、分片（shard）

索引可能会存储大量数据，这些数据可能超过单个节点的硬件限制。例如，十亿个文档的单个索引占用了1TB的磁盘空间，可能不适合单个节点的磁盘，或者可能太慢而无法单独满足来自单个节点的搜索请求。

为了解决此问题，Elasticsearch提供了将索引细分为多个碎片的功能。创建索引时，只需定义所需的分片数量即可。每个分片本身就是一个功能齐全且独立的“索引”，可以托管在群集中的任何节点上。

分片很重要，主要有两个原因：

它允许您水平分割/缩放内容量

它允许您跨碎片（可能在多个节点上）分布和并行化操作，从而提高性能/吞吐量

分片如何分布以及其文档如何聚合回到搜索请求中的机制完全由Elasticsearch管理，并且对您作为用户是透明的。

在随时可能发生故障的网络/云环境中，非常有用，强烈建议您使用故障转移机制，以防碎片/节点因某种原因脱机或消失。为此，Elasticsearch允许您将索引分片的一个或多个副本制作为所谓的副本分片（简称副本）。

4、副本（replica）

分片处理允许用户推送超过单机容量的数据至Elasticsearch集群。副本则解决了访问压力过大时单机无法处理所有请求的问题。

分片可以是主分片，也可以是副本分片，其中副本分片是主分片的完整副本。副本分片用于搜索，或者是在原有的主分片丢失后成为新的主分片。

注意：可以在任何时候改变每个分片的副本分片的数量，因为副本分片总是可以被创建和移除的。这并不适用于索引划分为主分片的数量，在创建索引之前，必须决定主分片的数量。过少的分片将限制可扩展性，但是过多的分片会影响性能。默认设置的5份是一个不错的开始。

5.1.2 文档、类型、索引及映射

1、文档（document）

Elasticsearch是面向文档的，这意味着索引和搜索数据的最小单位是文档。

在Elasticsearch中文档有几个重要的属性。

它是自我包含的。一篇文档同时包含字段和它们的取值。

它可以是层次的。文档中还包含新的文档，字段还可以包含其他字段和取值。例如，“location”字段可以同时包含“city”和“street“两个字段。

它拥有灵活的结构。文档不依赖于预先定义的模式。并非所有的文档都需要拥有相同的字段，它们不受限于同一个模式。

2、类型（type）

类型是文档的逻辑容器，类似于表格是行的容器。在不同的类型中，最好放入不同结构的文档。例如，可以用一个类型定义聚会时的分组，而另一个类型定义人们参加的活动。

3、索引（index）

索引是映射类型的容器。一个Elasticsearch索引是独立的大量的文档集合。每个索引存储在磁盘上的同组文件中，索引存储了所有映射类型的字段，还有一些设置。

4、映射（mapping）

所有文档在写入索引前都将被分析，用户可以设置一些参数，决定如何将输入文本分割为词条，哪些词条应该被过滤掉，或哪些附加处理有必要被调用（比如移除HTML标签）。这就是映射扮演的角色：存储分析链所需的所有信息。

Elasticsearch也是基于Lucene的全文检索库，本质也是存储数据，很多概念与MySQL类似的。

对比关系：

索引库（indices）---------------------------------Database 数据库类型（type）----------------------------------Table 数据表文档（Document）---------------------------------Row 行域字段（Field）--------------------------------Columns 列映射配置（mappings）-------------------------------每个列的约束（类型、长度）

详细说明：

索引库（indices)	indices是index的复数，代表许多的索引
概念	说明
类型（type）	类型是模拟mysql中的table概念，一个索引库下可以有不同类型的索引（目前6.X以后的版本只能有一个类型），类似数据库中的表概念。数据库表中有表结构，也就是表中每个字段的约束信息；索引库的类型中对应表结构的叫做映射(mapping) ，用来定义每个字段的约束。
文档（document）	存入索引库原始的数据。比如每一条商品信息，就是一个文档
字段（field）	文档中的属性
映射配置（mappings）	字段的数据类型、属性、是否索引、是否存储等特性

5.2.创建索引库

5.2.1.语法

Elasticsearch采用Rest风格API，因此其API就是一次http请求，你可以用任何工具发起http请求

创建索引的请求格式：

请求方式：PUT

请求路径：/索引库名

请求参数：json格式：

{"settings": {"属性名": "属性值"}
}

settings：就是索引库设置，其中可以定义索引库的各种属性，目前我们可以不设置，都走默认。

5.2.3.使用kibana创建

kibana的控制台，可以对http请求进行简化，示例：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4fWTihEC-1637500558788)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对索引库操作–创建索引库–使用kibana创建)]

相当于是省去了elasticsearch的服务器地址

而且还有语法提示，非常舒服。

5.3.查看索引库

Get请求可以帮我们查看索引信息，格式：

GET /索引库名

5.4.删除索引库

删除索引使用DELETE请求

DELETE /索引库名

示例

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FHr2NMXM-1637500558791)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对索引库操作–删除索引库01)]

再次查看lagou：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-48tUVuO1-1637500558793)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对索引库操作–删除索引库02)]

当然，我们也可以用HEAD请求，查看索引是否存在：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HuwNDsNn-1637500558797)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对索引库操作–删除索引库03)]

6 使用kibana对类型及映射操作

有了索引库，等于有了数据库中的database 。接下来就需要索引库中的类型了，也就是数据库中的表。创建数据库表需要设置字段约束，索引库也一样，在创建索引库的类型时，需要知道这个类型下有哪些字段，每个字段有哪些约束信息，这就叫做字段映射(mapping)

注意：Elasticsearch7.x取消了索引type类型的设置，不允许指定类型，默认为_doc，但字段仍然是有的，我们需要设置字段的约束信息，叫做字段映射（mapping）

字段的约束我们在学习Lucene中我们都见到过，包括到不限于：

字段的数据类型

是否要存储

是否要索引

是否分词

分词器是什么

6.1.创建字段映射

请求方式依然是PUT

PUT /索引库名/_mapping/typeName
{"properties": {"字段名": {"type": "类型","index": true，"store": true，"analyzer": "分词器"}}
}

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-d5m9sjiM-1637500558798)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对类型及映射操作–创建字段映射)]

类型名称：就是前面将的type的概念，类似于数据库中的表

字段名：任意填写，下面指定许多属性，例如：

type：类型，可以是text、keyword、long、short、date、integer、object等

index：是否索引，默认为true

store：是否存储，默认为false

analyzer：分词器，这里的ik_max_word即使用ik分词器

示例

发起请求：

PUT /lagou/_mapping/good
{"properties": {"title":{"type": "text","store": true,"analyzer": "ik_max_word"},"images":{"type": "keyword","store": true,"index": false},"price":{"type": "float"}}
}

响应结果：

{"acknowledged": true
}

上述案例中，就给lagou这个索引库添加了一个名为 goods 的类型，并且在类型中设置了3个字段：

title：商品标题

images：商品图片

price：商品价格

并且给这些字段设置了一些属性，至于这些属性对应的含义，我们在后续会详细介绍。

6.2.查看映射关系

GET /索引库名/_mapping

查看某个索引库中的所有类型的映射。如果要查看某个类型映射，可以再路径后面跟上类型名称。即：

GET /索引库名/_mapping/类型名

示例：

GET /lagou/_mapping/good

响应：

{"lagou": {"mappings": {"good": {"properties": {"images": {"type": "keyword","index": false,"store": true},"price": {"type": "float"},"title": {"type": "text","store": true,"analyzer": "ik_max_word"}}}}}
}

6.3.映射属性详解

1）type

Elasticsearch中支持的数据类型非常丰富：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zbLkgQGZ-1637500558801)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对类型及映射操作–映射属性详解)]


一级分类	二级分类	具体类型	使用
核心类型	字符串类型	text,keyword	结构化搜索，全文文本搜索、聚合、排序等
	整数类型	integer,long,short,byte	字段的长度越短，索引和搜索的效率越高
	浮点类型	double,float,half_float,scaled_float
	逻辑类型	boolean
	日期类型	date
	范围类型	range
	二进制类型	binary	该binary类型接受二进制值作为Base64编码的字符串。该字段默认情况下不存储(store)，并且不可搜索
复合类型	数组类型	array
	对象类型	object	用于单个JSON对象
	嵌套类型	nested	用于JSON对象数组
地理类型	地理坐标类型	geo_point	纬度/经度积分
	地理地图	geo_shape	用于多边形等复杂形状
特殊类型	IP类型	ip	用于IPv4和IPv6地址
	范围类型	completion	提供自动完成建议
	令牌计数类型	token_count	计算字符串中令牌的数量

我们说几个关键的：

String类型，又分两种：

text：使用文本数据类型的字段，它们会被分词，文本字段不用于排序，很少用于聚合，如文章标题、正文。

keyword：关键字数据类型，用于索引结构化内容的字段，不会被分词，必须完整匹配的内容，如邮箱，身份证号。支持聚合

这两种类型都是比较常用的，但有的时候，对于一个字符串字段，我们可能希望他两种都支持，此时，可以利用其多字段特性

"properties": {"title":{"type": "text","analyzer": "ik_max_word","fields": {"sort":{"type": "keyword"}},"index": true}
}

Numerical：数值类型，分两类

基本数据类型：long、interger、short、byte、double、float、half_float

double 双精度64位

float 单精度32位

half_float 半精度16位

浮点数的高精度类型：scaled_float

带有缩放因子的缩放类型浮点数，依靠一个long数字类型通过一个固定的( double 类型)缩放因数进行缩放

需要指定一个精度因子，比如10或100。elasticsearch会把真实值乘以这个因子后存储，取出时再还原

Date：日期类型

elasticsearch可以对日期格式化为字符串存储，但是建议我们存储为毫秒值，存储为long，节省空间

Array：数组类型

进行匹配时，任意一个元素满足，都认为满足

排序时，如果升序则用数组中的最小值来排序，如果降序则用数组中的最大值来排序

字符串数组：["one", "two"]
整数数组：[1,2]
数组的数组：[1, [2, 3]]，等价于[1,2,3]
对象数组：[ { "name": "Mary", "age": 12 }, { "name": "John", "age": 10 }]

Object：对象

JSON文档本质上是分层的：文档包含内部对象，内部对象本身还包含内部对象

{"region": "US","manager.age": 30,"manager.name ": "John Smith"
}
索引方法如下：
{"mappings": {"properties": {"region": { "type": "keyword" },"manager": {"properties": {"age": { "type": "integer" },"name": { "type": "text" }}}}}
}

如果存储到索引库的是对象类型，例如上面的girl，会把girl编程两个字段：girl.name和girl.age

ip地址

PUT my_index
{"mappings": {"_doc": {"properties": {"ip_addr": {"type": "ip"}}}}
}PUT my_index/_doc/1
{"ip_addr": "192.168.1.1"
}GET my_index/_search
{"query": {"term": {"ip_addr": "192.168.0.0/16"}}
}

2）index

index影响字段的索引情况。

true：字段会被索引，则可以用来进行搜索过滤。默认值就是true，只有当某一个字段的index值设置为true时，检索ES才可以作为条件去检索。

false：字段不会被索引，不能用来搜索

index的默认值就是true，也就是说你不进行任何配置，所有字段都会被索引。

但是有些字段是我们不希望被索引的，比如商品的图片信息（URL），就需要手动设置index为false。

3）store

是否将数据进行额外存储。

在学习lucene时，我们知道如果一个字段的store设置为false，那么在文档列表中就不会有这个字段的值，用户的搜索结果中不会显示出来。

但是在Elasticsearch中，即便store设置为false，也可以搜索到结果。

原因是Elasticsearch在创建文档索引时，会将文档中的原始数据备份，保存到一个叫做 _source 的属性中。而且我们可以通过过滤 _source 来选择哪些要显示，哪些不显示。

而如果设置store为true，就会在 _source 以外额外存储一份数据，多余，因此一般我们都会将store设置为false，事实上，store的默认值就是false。

在某些情况下，这对 store 某个领域可能是有意义的。例如，如果您的文档包含一个 title ，一个 date 和一个非常大的 content 字段，则可能只想检索the title 和the date 而不必从一个大 _source 字段中提取这些字段：

PUT my_index
{"mappings": {"_doc": {"properties": {"title": {"type": "text","store": true},"date": {"type": "date","store": true},"content": {}}}}
}

4）boost

网站权重：网站权重是指搜索引擎给网站（包括网页）赋予一定的权威值，对网站（含网页）权威的评估评价。一个网站权重越高，在搜索引擎所占的份量越大，在搜索引擎排名就越好。提高网站权重，不但利于网站（包括网页）在搜索引擎的排名更靠前，还能提高整站的流量，提高网站信任度。所以提高网站的权重具有相当重要的意义。权重即网站在SEO中的重要性，权威性。英文：Page Strength。1、权重不等于排名 2、权重对排名有着非常大的影响 3、整站权重的提高有利于内页的排名。

权重，新增数据时，可以指定该数据的权重，权重越高，得分越高，排名越靠前。

PUT my_index
{"mappings": {"_doc": {"properties": {"title": {"type": "text","boost": 2},"content": {"type": "text"}}}}
}

title 字段上的匹配项的权重是字段上的匹配项的权重的两倍 content ，默认 boost 值为1.0 。提升仅适用于Term查询（不提升prefix，range和模糊查询）。

6.4.一次创建索引库和类型

第一步：
PUT /lagou第二步：
PUT lagou/_mapping/goods
{"properties": {"title": {"type": "text","analyzer": "ik_max_word"},"images": {"type": "keyword","index": "false"},"price": {"type": "float"}}
}

刚才的案例中我们是把创建索引库和类型分开来做，其实也可以在创建索引库的同时，直接制定索引库中的类型，基本语法：

put /索引库名
{"settings":{"索引库属性名":"索引库属性值"},"mappings":{"类型名":{"properties":{"字段名":{"映射属性名":"映射属性值"}}}}
}

来试一下吧：

PUT /lagou2
{"settings": {},"mappings": {"goods":{"properties": {"title":{"index": true,"store": true,"analyzer": "ik_max_word","type": "text"},"price":{"index": true,"store": true,"type": "float"}}}}
}

结果：

{"acknowledged": true,"shards_acknowledged": true,"index": "lagou2"
}

7 使用kibana对文档操作

文档，即索引库中某个类型下的数据，会根据规则创建索引，将来用来搜索。可以类比做数据库中的每一行数据。

7.1.新增文档

7.1.1.新增并随机生成id

通过POST请求，可以向一个已经存在的索引库中添加文档数据。

POST /索引库名/类型名
{"key":"value"
}

示例：

POST /lagou/good
{"title":"HuaWei P30Pro","image":"http://image.lagou.com/12479122.jpg","price":9999.9
}

响应：

{"_index": "lagou","_type": "good","_id": "KSBaKH0BeslWetzn6_70","_version": 1,"result": "created","_shards": {"total": 2,"successful": 1,"failed": 0},"_seq_no": 0,"_primary_term": 2
}

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Digcnvb0-1637500558805)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–新增并随机生成id)]

可以看到结果显示为： created ，应该是创建成功了。

另外，需要注意的是，在响应结果中有个_id字段，这个就是这条文档数据的唯一标示，以后的增删改查都依赖这个id作为唯一标示。

可以看到id的值为： tURGznQB29tVfg_iWHfl ,这里我们新增时没有指定id，所以是ES帮我们随机生成的id。

7.2.查看文档

根据rest风格，新增是post，查询应该是get，不过查询一般都需要条件，这里我们把刚刚生成数据的id带上。

通过kibana查看数据：

GET /lagou/good/KSBaKH0BeslWetzn6_70

查看结果：

{"_index": "lagou","_type": "good","_id": "KSBaKH0BeslWetzn6_70","_version": 1,"found": true,"_source": {"title": "HuaWei P30Pro","image": "http://image.lagou.com/12479122.jpg","price": 9999.9}
}

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-y4qYR7tK-1637500558807)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查看文档)]

_source：源文档信息，所有的数据都在里面。

_id：这条文档的唯一标示

自动生成的id，长度为20个字符，URL安全，base64编码，GUID（全局唯一标识符），分布式系统并行生成时不可能会发生冲突

在实际开发中不建议使用ES生成的ID，太长且为字符串类型，检索时效率低。建议：将数据表中唯一的ID，作为ES的文档ID

7.3.新增文档并自定义id

如果我们想要自己新增的时候指定id，可以这么做：

POST /索引库名/类型/id值
{...
}

示例：

POST /lagou/good/10
{"title":"大米手机","images":"http://image.lagou.com/12479122.jpg","price":2899.00
}

得到的数据：

{"_index": "lagou","_type": "good","_id": "10","_score": 1,"_source": {"title": "大米手机","images": "http://image.lagou.com/12479122.jpg","price": 2899}
}

7.4.修改数据

PUT：修改文档

POST：新增文档

把刚才新增的请求方式改为PUT，就是修改了。不过修改必须指定id

id对应文档存在，则修改

id对应文档不存在，则新增

比如，我们把使用id为11，不存在，则应该是新增：

PUT /lagou/good/11
{"title":"lagou手机Pro","images":"http://image.lagou.com/12479122.jpg","price":3899.00
}

结果：

{"_index": "lagou","_type": "good","_id": "11","_version": 1,"result": "created","_shards": {"total": 2,"successful": 1,"failed": 0},"_seq_no": 0,"_primary_term": 2
}

可以看到是created，是新增。

我们再次执行刚才的请求，不过把数据改一下：

PUT /lagou/good/11
{"title":"lagou手机拉勾Pro","images":"http://image.lagou.com/12479122.jpg","price":3899.00
}

查看结果：

{"_index": "lagou","_type": "good","_id": "11","_version": 2,"result": "updated","_shards": {"total": 2,"successful": 1,"failed": 0},"_seq_no": 1,"_primary_term": 2
}

可以看到结果是：updated，显然是更新数据

7.5.删除数据

删除使用DELETE请求，同样，需要根据id进行删除：

DELETE /索引库名/类型名/id值

示例：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ySoLWsyq-1637500558809)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–.删除数据)]

7.6.智能判断

刚刚我们在新增数据时，添加的字段都是提前在类型中定义过的，如果我们添加的字段并没有提前定义过，能够成功吗？

事实上Elasticsearch非常智能，你不需要给索引库设置任何mapping映射，它也可以根据你输入的数据来判断类型，动态添加数据映射。

测试一下：

POST /lagou/good/3
{"title":"超大米手机","images":"http://image.lagou.com/12479122.jpg","price":3299.00,"stock": 200,"saleable":true,"subTitle":"大米"
}

我们额外添加了stock库存，saleable是否上架，subtitle副标题、3个字段。

来看结果：

{"_index": "lagou","_type": "good","_id": "3","_version": 1,"result": "created","_shards": {"total": 2,"successful": 1,"failed": 0},"_seq_no": 2,"_primary_term": 2
}

成功了！在看下索引库的映射关系:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-G9bVmJLJ-1637500558812)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–智能判断)]

{"lagou": {"aliases": {},"mappings": {"good": {"properties": {"images": {"type": "keyword","index": false,"store": true},"price": {"type": "float"},"saleable": {"type": "boolean"},"stock": {"type": "long"},"subTitle": {"type": "text","fields": {"keyword": {"type": "keyword","ignore_above": 256}}},"title": {"type": "text","store": true,"analyzer": "ik_max_word"}}}}}
}

stock、saleable、subtitle都被成功映射了。

subtitle是String类型数据，ES无法智能判断，它就会存入两个字段。例如：

subtitle：text类型

subtitle.keyword：keyword类型

这种智能映射，底层原理是动态模板映射，如果我们想修改这种智能映射的规则，其实只要修改动态模板即可！

7.7.动态映射模板

动态模板的语法：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Pu7K3azT-1637500558814)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–动态映射模板)]

1）模板名称，随便起

2）匹配条件，凡是符合条件的未定义字段，都会按照这个规则来映射

3）映射规则，匹配成功后的映射规则

举例，我们可以把所有未映射的string类型数据自动映射为keyword类型：

PUT /lagou2
{"mappings": {"goods":{"properties": {"title":{"type": "text","index": true,"store": true}},"dynamic_templates": [{"string_template":{"match_mapping_type":"string","mapping":{"type":"keyword"}}}]}}
}

在这个案例中，我们把做了两个映射配置：

title字段：统一映射为text类型，并制定分词器

其它字段：只要是string类型，统一都处理为keyword类型

这样，未知的string类型数据就不会被映射为text和keyword并存，而是统一以keyword来处理！

我们试试看新增一个数据：

POST /lagou2/goods/1
{"title":"超大米手机","images":"http://image.lagou.com/12479122.jpg","price":3299.00
}

我们只对title做了配置，现在来看看images和price会被映射为什么类型呢：

GET /lagou2/_mapping

结果：

{"lagou2": {"mappings": {"goods": {"dynamic_templates": [{"string_template": {"match_mapping_type": "string","mapping": {"type": "keyword"}}}],"properties": {"images": {"type": "keyword"},"price": {"type": "float"},"title": {"type": "text","store": true}}}}}
}

可以看到images被映射成了keyword，而非之前的text和keyword并存，说明我们的动态模板生效了！

8 查询（重点）

8.1.基本查询

基本语法

GET /索引库名/_search
{"query":{"查询类型":{"查询条件":"查询条件值"}}
}

这里的query代表一个查询对象，里面可以有不同的查询属性

查询类型：

例如：match_all，match，term，range等等

查询条件：查询条件会根据类型的不同，写法也有差异，后面详细讲解

8.1.1 查询所有（match_all)

示例：

GET /lagou/_search
{"query": {"match_all": {}}
}

query：代表查询对象

match_all：代表查询所有

结果：

{"took": 3,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 3,"max_score": 1,"hits": [{"_index": "lagou","_type": "good","_id": "10","_score": 1,"_source": {"title": "lagou手机Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899}},{"_index": "lagou","_type": "good","_id": "11","_score": 1,"_source": {"title": "lagou手机拉勾Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899}},{"_index": "lagou","_type": "good","_id": "3","_score": 1,"_source": {"title": "超大米手机","images": "http://image.lagou.com/12479122.jpg","price": 3299,"stock": 200,"saleable": true,"subTitle": "大米"}}]}
}

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ha6qM65b-1637500558817)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查询–查询所有01)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mI8n66Jj-1637500558820)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查询–查询所有02)]

took：查询花费时间，单位是毫秒

time_out：是否超时

_shards：分片信息

hits：搜索结果总览对象

total：搜索到的总条数

max_score：所有结果中文档得分的最高分

hits：搜索结果的文档对象数组，每个元素是一条搜索到的文档信息

index：索引库

type：文档类型

_id：文档id

score：文档得分

source：文档的源数据

文档得分：使用ES时，对于查询出的文档无疑会有文档相似度之别。而理想的排序是和查询条件相关性越高排序越靠前，而这个排序的依据就是_score

8.1.2 匹配查询（match）

我们先加入一条数据，便于测试：

PUT /lagou/goods/3
{"title":"小米电视4A","images":"http://image.lagou.com/12479122.jpg","price":3899.00
}

现在，索引库中有2部手机，1台电视：

or关系

match类型查询，会把查询条件进行分词，然后进行查询，多个词条之间是or的关系

GET /lagou/_search
{"query":{"match":{"title":"小米电视"}}
}

在上面的案例中，不仅会查询到电视，而且与小米相关的都会查询到，多个词之间是or的关系。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZhGYLFsA-1637500558823)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查询–匹配查询)]

and关系

某些情况下，我们需要更精确查找：比如在电商平台精确搜索商品时，我们希望这个关系（查询条件切分词之后的关系）变成and（既要满足你，又要满足我），可以这样做：

GET /lagou/_search
{"query": {"match": {"title": {"query": "小米电视","operator": "and"}}}
}

本例中，只有同时包含小米和电视的词条才会被搜索到。

8.1.3 词条匹配(term)

term查询被用于精确值匹配，这些精确值可能是数字、时间、布尔或者那些未分词的字符串，keyword类型的字符串

效果类似于：select * from tableName where colName=‘value’;

GET /lagou/_search
{"query": {"term": {"price": 3899}}
}

结果：

{"took": 2,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 3,"max_score": 1,"hits": [{"_index": "lagou","_type": "good","_id": "10","_score": 1,"_source": {"title": "lagou手机Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899}},{"_index": "lagou","_type": "good","_id": "11","_score": 1,"_source": {"title": "lagou手机拉勾Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899}},{"_index": "lagou","_type": "good","_id": "3","_score": 1,"_source": {"title": "小米电视4A","images": "http://image.lagou.com/12479122.jpg","price": 3899}}]}
}

8.1.4 布尔组合（bool)

bool把各种其它查询通过must（与）、must_not（非）、should（或）的方式进行组合

GET /lagou/_search
{"query": {"bool": {"must": {"match":{"title":"大米"}},"must_not": {"match":{"title":"电视"}},"should": {"match":{"title":"手机"}}}}
}

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vLTi7hPi-1637500558828)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查询–布尔组合)]

结果：

{"took": 10,"timed_out": false,"_shards": {"total": 3,"successful": 3,"skipped": 0,"failed": 0},"hits": {"total": 1,"max_score": 0.5753642,"hits": [{"_index": "lagou","_type": "goods","_id": "2","_score": 0.5753642,"_source": {"title": "大米手机","images": "http://image.lagou.com/12479122.jpg","price": 2899}}]}
}

8.1.5 范围查询(range)

range查询找出那些落在指定区间内的数字或者时间

GET /lagou/_search
{"query": {"range": {"price": {"gte": 1000,"lt": 3990}}}
}

range查询允许以下字符：

操作符	说明
gt	大于
gte	大于等于
lt	小于
lte	小于等于

8.1.6 模糊查询(fuzzy)

fuzzy 查询是 term 查询的模糊等价，很少直接使用它。

我们新增一个商品：

POST /lagou/good/5
{"title":"apple手机","images":"http://image.lagou.com/12479122.jpg","price":6899.00
}

fuzzy查询是term查询的模糊等价。它允许用户搜索词条与实际词条的拼写出现偏差，但是偏差的编辑距离不得超过2：

GET /lagou/_search
{"query": {"fuzzy": {"title": "appla"}}
}

上面的查询，也能查询到apple手机

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ABS1ltlH-1637500558831)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–使用kibana对文档操作–查询–模糊查询)]

8.2.结果过滤

默认情况下，elasticsearch在搜索的结果中，会把文档中保存在_source的所有字段都返回。

如果我们只想获取其中的部分字段，我们可以添加_sourc的过滤

8.2.1.直接指定字段

示例：

GET /lagou/_search
{"_source": ["title","price"], "query": {"match": {"title": "小米手机"}}
}

返回的结果：

{"took": 75,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 4,"max_score": 0.6931472,"hits": [{"_index": "lagou","_type": "good","_id": "11","_score": 0.6931472,"_source": {"price": 3899,"title": "lagou手机拉勾Pro"}}]}
}

8.2.2.指定includes和excludes

我们也可以通过：

includes：来指定想要显示的字段

excludes：来指定不想要显示的字段

二者都是可选的。

示例：

GET /lagou/_search
{"_source": {"includes": ["price","", "title"]},"query": {"match": {"title": "小米手机"}}
}

与下面的结果将是一样的：

GET /lagou/_search
{"_source": {"excludes": ["images"]},"query": {"match": {"title": "小米手机"}}
}

8.3 过滤(filter)

Elasticsearch 使用的查询语言（DSL）拥有一套查询组件，这些组件可以以无限组合的方式进行搭配。这套组件可以在以下两种情况下使用：过滤情况（filtering context）和查询情况（query context）。

如何选择查询与过滤：

通常的规则是，使用查询（query）语句来进行全文搜索或者其它任何需要影响相关性得分的搜索。除此以外的情况都使用过滤（filters)。

条件查询中进行过滤

所有的查询都会影响到文档的评分及排名。如果我们需要在查询结果中进行过滤，并且不希望过滤条件影响评分，那么就不要把过滤条件作为查询条件来用。而是使用filter方式：

GET /lagou/_search
{"query": {"bool": {"must": {"match":{"title":"手机"}},"filter": {"range": {"price": {"gte": 2000,"lte": 4000}}}}}
}

无查询条件，直接过滤

如果一次查询只有过滤，没有查询条件，不希望进行评分，我们可以使用constant_score取代只有 filter 语句的 bool 查询。在性能上是完全相同的，但对于提高查询简洁性和清晰度有很大帮助。

GET /lagou/_search
{"query": {"constant_score": {"filter": {"range": {"price": {"gte": 2000,"lte": 4000}}}}}
}

8.4 排序

8.4.1 单字段排序

sort可以让我们按照不同的字段进行排序，并且通过order指定排序的方式

GET /lagou/_search
{"query": {"match": {"title": "小米手机"}},"sort": [{"price": {"order": "desc"}}]
}

8.4.2 多字段排序

假定我们想要结合使用 price和 _score（得分）进行查询，并且匹配的结果首先按照价格排序，然后按照相关性得分排序：

GET /lagou/_search
{"query": {"match": {"title": "小米手机"}},"sort": [{"price": {"order": "desc"}},{"_score":{"order": "desc"}}]
}

8.5.分页

Elasticsearch中数据都存储在分片中，当执行搜索时每个分片独立搜索后，数据再经过整合返回。那么，如果要实现分页查询该怎么办呢？

elasticsearch的分页与mysql数据库非常相似，都是指定两个值：

from：目标数据的偏移值（开始位置），默认from为0

size：每页大小

GET /lagou/_search
{"query": {"match": {"title": "小米手机"}},"sort": [{"price": {"order": "desc"}}],"from": 2,"size": 2
}

结果：

{"took": 3,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 6,"max_score": null,"hits": [{"_index": "lagou","_type": "good","_id": "11","_score": null,"_source": {"title": "lagou手机拉勾Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899},"sort": [3899]},{"_index": "lagou","_type": "good","_id": "3","_score": null,"_source": {"title": "小米电视4A","images": "http://image.lagou.com/12479122.jpg","price": 3899},"sort": [3899]}]}
}

8.6.高亮

高亮原理：

服务端搜索数据，得到搜索结果

把搜索结果中，搜索关键字都加上约定好的标签

前端页面提前写好标签的CSS样式，即可高亮

elasticsearch中实现高亮的语法比较简单：

GET /lagou/_search
{"query": {"match": {"title": "手机"}},"highlight": {"pre_tags": "<em>","post_tags": "</em>","fields": {"title": {}}}
}

在使用match查询的同时，加上一个highlight属性：

pre_tags：前置标签

post_tags：后置标签

fields：需要高亮的字段

title：这里声明title字段需要高亮

结果：

{"took": 88,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 5,"max_score": 0.6931472,"hits": [{"_index": "lagou","_type": "good","_id": "11","_score": 0.6931472,"_source": {"title": "lagou手机拉勾Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899},"highlight": {"title": ["lagou<em>手机</em>拉勾Pro"]}},{"_index": "lagou","_type": "good","_id": "4","_score": 0.2876821,"_source": {"title": "小米手机","images": "http://image.lagou.com/12479122.jpg","price": 2699},"highlight": {"title": ["小米<em>手机</em>"]}},{"_index": "lagou","_type": "good","_id": "5","_score": 0.14181954,"_source": {"title": "apple手机","images": "http://image.lagou.com/12479122.jpg","price": 6899},"highlight": {"title": ["apple<em>手机</em>"]}},{"_index": "lagou","_type": "good","_id": "12","_score": 0.14181954,"_source": {"title": "小米手机","images": "http://image.lagou.com/12479122.jpg","price": 2699},"highlight": {"title": ["小米<em>手机</em>"]}},{"_index": "lagou","_type": "good","_id": "10","_score": 0.11955718,"_source": {"title": "lagou手机Pro","images": "http://image.lagou.com/12479122.jpg","price": 3899},"highlight": {"title": ["lagou<em>手机</em>Pro"]}}]}
}

9 聚合aggregations

聚合可以让我们极其方便的实现对数据的统计、分析。例如：

什么品牌的手机最受欢迎？

这些手机的平均价格、最高价格、最低价格？

这些手机每月的销售情况如何？

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

9.1 基本概念

Elasticsearch中的聚合，包含多种类型，最常用的两种，一个叫桶，一个叫度量：

桶（bucket）类似于 group by

桶的作用，是按照某种方式对数据进行分组，每一组数据在ES中称为一个桶，例如我们根据国籍对人划分，可以得到中国桶、英国桶，日本桶 ……或者我们按照年龄段对人进行划分： 0_10,1020,20_30,3040等。

Elasticsearch中提供的划分桶的方式有很多：

Date Histogram Aggregation：根据日期阶梯分组，例如给定阶梯为周，会自动每周分为一组

Histogram Aggregation：根据数值阶梯分组，与日期类似，需要知道分组的间隔（interval）

Terms Aggregation：根据词条内容分组，词条内容完全匹配的为一组

Range Aggregation：数值和日期的范围分组，指定开始和结束，然后按段分组

…

综上所述，我们发现bucket aggregations 只负责对数据进行分组，并不进行计算，因此往往bucket中往往会嵌套另一种聚合：metrics aggregations即度量

度量（metrics）相当于聚合的结果

分组完成以后，我们一般会对组中的数据进行聚合运算，例如求平均值、最大、最小、求和等，这些在ES中称为 度量

比较常用的一些度量聚合方式：

Avg Aggregation：求平均值

Max Aggregation：求最大值

Min Aggregation：求最小值

Percentiles Aggregation：求百分比

Stats Aggregation：同时返回avg、max、min、sum、count等

Sum Aggregation：求和

Top hits Aggregation：求前几

Value Count Aggregation：求总数

……

为了测试聚合，我们先批量导入一些数据

创建索引：

PUT /car
{"mappings": {"orders":{"properties": {"color":{"type": "keyword"},"make":{"type": "keyword"}}}}
}

注意：在ES中，需要进行聚合、排序、过滤的字段其处理方式比较特殊，因此不能被分词，必须使用keyword或 数值类型。这里我们将color和make这两个文字类型的字段设置为keyword类型，这个类型不会被分词，将来就可以参与聚合

导入数据，这里是采用批处理的API，大家直接复制到kibana运行即可：

POST /car/orders/_bulk
{"index":{}}
{"price":10000,"color":"红","make":"本田","sold":"2020-10-28"}
{"index":{}}
{"price":20000,"color":"红","make":"本田","sold":"2020-11-05"}
{"index":{}}
{"price":30000,"color":"绿","make":"福特","sold":"2020-05-18"}
{"index":{}}
{"price":15000,"color":"蓝","make":"丰田","sold":"2020-07-02"}
{"index":{}}
{"price":12000,"color":"绿","make":"丰田","sold":"2020-08-19"}
{"index":{}}
{"price":20000,"color":"红","make":"本田","sold":"2020-11-05"}
{"index":{}}
{"price":80000,"color":"红","make":"宝马","sold":"2020-01-01"}
{"index":{}}
{"price":25000,"color":"蓝","make":"福特","sold":"2020-02-12"}

9.2 聚合为桶

首先，我们按照汽车的颜色color来划分桶，按照颜色分桶，最好是使用TermAggregation类型，按照颜色的名称来分桶。

GET /car/_search
{"size": 0,"aggs": {"popular_color": {"terms": {"field": "color"}}}
}

size：查询条数，这里设置为0，因为我们不关心搜索到的数据，只关心聚合结果，提高效率

aggs：声明这是一个聚合查询，是aggregations的缩写

popular_colors：给这次聚合起一个名字，可任意指定。

terms：聚合的类型，这里选择terms，是根据词条内容（这里是颜色）划分

field：划分桶时依赖的字段

结果：

{"took": 4,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 8,"max_score": 0,"hits": []},"aggregations": {"popular_color": {"doc_count_error_upper_bound": 0,"sum_other_doc_count": 0,"buckets": [{"key": "红","doc_count": 4},{"key": "绿","doc_count": 2},{"key": "蓝","doc_count": 2}]}}
}

hits：查询结果为空，因为我们设置了size为0

aggregations：聚合的结果

popular_colors：我们定义的聚合名称

buckets：查找到的桶，每个不同的color字段值都会形成一个桶

key：这个桶对应的color字段的值

doc_count：这个桶中的文档数量

通过聚合的结果我们发现，目前红色的小车比较畅销！

9.3 桶内度量

前面的例子告诉我们每个桶里面的文档数量，这很有用。但通常，我们的应用需要提供更复杂的文档度量。例如，每种颜色汽车的平均价格是多少？

因此，我们需要告诉Elasticsearch使用哪个字段， 使用何种度量方式进行运算，这些信息要嵌套在桶内，度量的运算会基于桶内的文档进行

现在，我们为刚刚的聚合结果添加求价格平均值的度量：

GET /car/_search
{"size": 0,"aggs": {"popular_color": {"terms": {"field": "color"},"aggs": {"avg_price": {"avg": {"field": "price"}}}}}
}

aggs：我们在上一个aggs(popular_colors)中添加新的aggs。可见度量也是一个聚合

avg_price：聚合的名称

avg：度量的类型，这里是求平均值

field：度量运算的字段

结果：

{"took": 2,"timed_out": false,"_shards": {"total": 5,"successful": 5,"skipped": 0,"failed": 0},"hits": {"total": 8,"max_score": 0,"hits": []},"aggregations": {"popular_color": {"doc_count_error_upper_bound": 0,"sum_other_doc_count": 0,"buckets": [{"key": "红","doc_count": 4,"avg_price": {"value": 32500}},{"key": "绿","doc_count": 2,"avg_price": {"value": 21000}},{"key": "蓝","doc_count": 2,"avg_price": {"value": 20000}}]}}
}

可以看到每个桶中都有自己的avg_price字段，这是度量聚合的结果

10 Elasticsearch集群

在之前的课程中，我们都是使用单点的elasticsearch，接下来我们会学习如何搭建Elasticsearch的集群。

10.1.单点的问题

单点的elasticsearch存在哪些可能出现的问题呢？

单台机器存储容量有限，无法实现高存储

单服务器容易出现单点故障，无法实现高可用

单服务的并发处理能力有限，无法实现高并发

所以，为了应对这些问题，我们需要对elasticsearch搭建集群

10.2.集群的结构

10.2.1.数据分片

首先，我们面临的第一个问题就是数据量太大，单点存储量有限的问题。

大家觉得应该如何解决？

没错，我们可以把数据拆分成多份，每一份存储到不同机器节点（node），从而实现减少每个节点数据量的目的。这就是数据的分布式存储，也叫做：数据分片（Shard）。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eEqQAMk1-1637500558835)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch集群–集群的结构–数据分片)]

10.2.2.数据备份

数据分片解决了海量数据存储的问题，但是如果出现单点故障，那么分片数据就不再完整，这又该如何解决呢？

没错，就像大家为了备份手机数据，会额外存储一份到移动硬盘一样。我们可以给每个分片数据进行备份，存储到其它节点，防止数据丢失，这就是数据备份，也叫数据副本（replica）。

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

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

首先对数据分片，存储到不同节点

然后对每个分片进行备份，放到对方节点，完成互相备份

这样可以大大减少所需要的服务节点数量，如图，我们以3分片，每个分片备份一份为例：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6ZbVz5ro-1637500558837)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch集群–集群的结构–数据备份)]

在这个集群中，如果出现单节点故障，并不会导致数据缺失，所以保证了集群的高可用，同时也减少了节点中数据存储量。并且因为是多个节点存储数据，因此用户请求也会分发到不同服务器，并发能力也得到了一定的提升。

10.3.搭建集群

集群需要多台机器，我们这里用一台机器来模拟，因此我们需要在一台虚拟机中部署多个elasticsearch节点，每个elasticsearch的端口都必须不一样。

一台机器进行模拟：将我们的ES的安装包复制三份，修改端口号，data和log存放位置的不同。

实际开发中：将每个ES节点放在不同的服务器上。

我们计划集群名称为：lagou-elastic，部署3个elasticsearch节点，分别是：

node-01：http端口9201，TCP端口9301

node-02：http端口9202，TCP端口9302

node-03：http端口9203，TCP端口9303

http：表示使用http协议进行访问时使用端口，elasticsearch-head、kibana、postman，默认端口号是9200。

tcp：集群间的各个节点进行通讯的端口，默认9300

第一步：复制es软件粘贴3次，分别改名

第二步：修改每一个节点的配置文件 config下的elasticsearch.yml，下面已第一份配置文件为例

三个节点的配置文件几乎一致，除了：node.name、path.data、path.log、http.port、transport.tcp.port

node-01：

#允许跨域名访问
http.cors.enabled: true
#当设置允许跨域，默认为*,表示支持所有域名
http.cors.allow-origin: "*"
#允许所有节点访问
network.host: 0.0.0.0
# 集群的名称，同一个集群下所有节点的集群名称应该一致
cluster.name: lagou-elastic
#当前节点名称 每个节点不一样
node.name: node-01
#数据的存放路径 每个节点不一样，不同es服务器对应的data和log存储的路径不能一样
path.data: e:\class\es-9201\data
#日志的存放路径 每个节点不一样
path.logs: e:\class\es-9201\logs
# http协议的对外端口 每个节点不一样，默认：9200
http.port: 9201
# TCP协议对外端口 每个节点不一样，默认：9300
transport.tcp.port: 9301
#三个节点相互发现，包含自己，使用tcp协议的端口号
discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"]
#声明大于几个的投票主节点有效，请设置为（nodes / 2） + 1
discovery.zen.minimum_master_nodes: 2
# 是否为主节点
node.master: true

node-02：

#允许跨域名访问
http.cors.enabled: true
http.cors.allow-origin: "*"
network.host: 0.0.0.0
# 集群的名称
cluster.name: lagou-elastic
#当前节点名称 每个节点不一样
node.name: node-02
#数据的存放路径 每个节点不一样
path.data: e:\class\es-9202\data
#日志的存放路径 每个节点不一样
path.logs: e:\class\es-9202\logs
# http协议的对外端口 每个节点不一样
http.port: 9202
# TCP协议对外端口 每个节点不一样
transport.tcp.port: 9302
#三个节点相互发现
discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"]
#声明大于几个的投票主节点有效，请设置为（nodes / 2） + 1
discovery.zen.minimum_master_nodes: 2
# 是否为主节点
node.master: true

node-03：

#允许跨域名访问
http.cors.enabled: true
http.cors.allow-origin: "*"
network.host: 0.0.0.0
# 集群的名称
cluster.name: lagou-elastic
#当前节点名称 每个节点不一样
node.name: node-03
#数据的存放路径 每个节点不一样
path.data: e:\class\es-9203\data
#日志的存放路径 每个节点不一样
path.logs: e:\class\es-9203\logs
# http协议的对外端口 每个节点不一样
http.port: 9203
# TCP协议对外端口 每个节点不一样
transport.tcp.port: 9303
#三个节点相互发现
discovery.zen.ping.unicast.hosts: ["127.0.0.1:9301","127.0.0.1:9302","127.0.0.1:9303"]
#声明大于几个的投票主节点有效，请设置为（nodes / 2） + 1
discovery.zen.minimum_master_nodes: 2
# 是否为主节点
node.master: true

第三步：启动集群

把三个节点分别启动，启动时不要着急，要一个一个地启动

使用head插件查看：

10.4.测试集群中创建索引库

配置kibana，再重启

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gYsv9NPX-1637500558839)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch集群–测试集群中创建索引库01)]

搭建集群以后就要创建索引库了，那么问题来了，当我们创建一个索引库后，数据会保存到哪个服务节点上呢？如果我们对索引库分片，那么每个片会在哪个节点呢？

这个要亲自尝试才知道。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cB73FSQg-1637500558840)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch集群–测试集群中创建索引库02)]

还记得创建索引库的API吗？

请求方式：PUT

请求路径：/索引库名

请求参数：json格式：

{"settings": {"属性名": "属性值"}
}

settings：就是索引库设置，其中可以定义索引库的各种属性，目前我们可以不设置，都走默认。

这里给搭建看看集群中分片和备份的设置方式，示例：

PUT /lagou
{"settings": {"number_of_shards": 3,"number_of_replicas": 1}
}

这里有两个配置：

number_of_shards：分片数量，这里设置为3

number_of_replicas：副本数量，这里设置为1，每个分片一个备份，一个原始数据，共2份。

通过chrome浏览器的head查看，我们可以查看到分片的存储结构：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7DQjWL0n-1637500558842)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch集群–测试集群中创建索引库03)]

可以看到，lagou这个索引库，有三个分片，分别是0、1、2，每个分片有1个副本，共6份。

node-01上保存了1号分片和2号分片的副本

node-02上保存了0号分片和2号分片的副本

node-03上保存了0号分片和1号分片的副本

10.5.集群工作原理

10.5.1.shad与replica机制

（1）一个index包含多个shard，也就是一个index存在多个服务器上

（2）每个shard都是一个最小工作单元，承载部分数据，比如有三台服务器，现在有三条数据，这三条数据在三台服务器上各方一条

（3）增减节点时，shard会自动在nodes中负载均衡

（4）primary shard（主分片）和replica shard（副本分片），每个document肯定只存在于某一个 primary shard以及其对应的replica shard中，不可能存在于多个primary shard

（5）replica shard是primary shard的副本，负责容错，以及承担读请求负载

（6）primary shard的数量在创建索引的时候就固定了，replica shard的数量可以随时修改

（7）primary shard的默认数量是5，replica默认是1（每个主分片一个副本分片），默认有10个 shard，5个primary shard，5个replica shard

（8）primary shard不能和自己的replica shard放在同一个节点上（否则节点宕机，primary shard和副本都丢失，起不到容错的作用），但是可以和其他primary shard的replica shard放在同一个节点上

10.5.2.集群写入数据

客户端选择一个node发送请求过去，这个node就是coordinating node (协调节点)
coordinating node，对document进行路由，将请求转发给对应的node。（根据一定的算法选择对应的节点进行存储）
实际上的node上的primary shard处理请求，将数据保存在本地，然后将数据同步到replica node
coordinating node，如果发现primary node和所有的replica node都搞定之后，就会返回请求到客户端

这个路由简单的说就是取模算法，比如说现在有3太服务器，这个时候传过来的id是5，那么5%3=2，就放在第2台服务器

10.5.3.ES查询数据

倒排序算法

查询有个算法叫倒排序：简单的说就是：通过分词把词语出现的id进行记录下来，再查询的时候先去查到哪些id包含这个数据，然后再根据id把数据查出来

查询过程

客户端发送一个请求给coordinate node
协调节点将搜索的请求转发给所有的shard对应的primary shard 或replica shard
query phase（查询阶段）：每一个shard 将自己搜索的结果（其实也就是一些唯一标识），返回给协调节点，有协调节点进行数据的合并，排序，分页等操作，产出最后的结果
fetch phase（获取阶段），接着由协调节点，根据唯一标识去各个节点进行拉取数据，最终返回给客户端

11 Elasticsearch客户端

11.1.客户端介绍

在elasticsearch官网中提供了各种语言的客户端：https://www.elastic.co/guide/en/elasticsearch/client/index.html

注意点击进入后，选择版本到6.2.4，因为我们之前按照的都是6.2.4版本：

11.2.创建Demo工程

11.2.1.初始化项目

11.2.2.pom文件

注意，这里我们直接导入了SpringBoot的启动器，方便后续讲解。不过还需要手动引入elasticsearch的High-level-Rest-Client的依赖：

<properties><java.version>11</java.version>
</properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-logging</artifactId></dependency><dependency><groupId>com.google.code.gson</groupId><artifactId>gson</artifactId><version>2.8.5</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.8.1</version></dependency><!--Apache开源组织提供的用于操作JAVA BEAN的工具包--><dependency><groupId>commons-beanutils</groupId><artifactId>commons-beanutils</artifactId><version>1.9.1</version></dependency><!--ES高级Rest Client--><dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId><version>6.4.3</version></dependency>
</dependencies>
<build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins>
</build>

11.2.3.配置文件

我们在resource下创建application.yml

11.3.索引库及映射

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jUtQhg7d-1637500558846)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–索引库及映射)]

创建索引库的同时，我们也会创建type及其映射关系，但是这些操作不建议使用java客户端完成，原因如下：

索引库和映射往往是初始化时完成，不需要频繁操作，不如提前配置好

官方提供的创建索引库及映射API非常繁琐，需要通过字符串拼接json结构：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nspEW0A3-1637500558850)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–索引库及映射02)]

因此，这些操作建议还是使用我们昨天学习的Rest风格API去实现。

我们接下来以这样一个商品数据为例来创建索引库：

package com.lagou.pojo;import lombok.Data;@Data
public class Product {private Long id;private String title; //标题private String category;// 分类private String brand; // 品牌private Double price; // 价格private String images; // 图片地址
}

分析一下数据结构：

id：可以认为是主键，将来判断数据是否重复的标示，不分词，可以使用keyword类型

title：搜索字段，需要分词，可以用text类型

category：商品分类，这个是整体，不分词，可以使用keyword类型

brand：品牌，与分类类似，不分词，可以使用keyword类型

price：价格，这个是double类型

images：图片，用来展示的字段，不搜索，index为false，不分词，可以使用keyword类型

我们可以编写这样的映射配置：

PUT /lagou
{"settings": {"number_of_shards": 3,"number_of_replicas": 1},"mappings": {"item": {"properties": {"id": {"type": "keyword"},"title": {"type": "text","analyzer": "ik_max_word"},"category": {"type": "keyword"},"brand": {"type": "keyword"},"images": {"type": "keyword","index": false},"price": {"type": "double"}}}}
}

11.4.索引数据操作

有了索引库，我们接下来看看如何新增索引数据

操作MYSQL数据库：

1.获取数据库连接

2.完成数据的增删改查

3.释放资源

11.4.1.初始化客户端

完成任何操作都需要通过HighLevelRestClient客户端，看下如何创建。

我们先编写一个测试类：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jh6Yfa1L-1637500558854)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–.索引数据操作–.初始化客户端)]

然后再@Before的方法中编写client初始化：

public class ElasticSearchTest {private RestHighLevelClient client;// Json工具private Gson gson = new Gson();@Beforepublic void init(){// 初始化HighLevel客户端client = new RestHighLevelClient(RestClient.builder(new HttpHost("127.0.0.1", 9201, "http"),new HttpHost("127.0.0.1", 9202, "http"),new HttpHost("127.0.0.1", 9203, "http")));}@Afterpublic void close() throws IOException {// 关闭客户端client.close();}
}

11.4.2.新增文档

示例：

/*** 插入文档* @throws IOException*/
@Test
public void testInsert() throws IOException {//1.文档数据Product product = new Product();product.setBrand("华为");product.setCategory("手机");product.setId(1L);product.setImages("http://image.huawei.com/1.jpg");product.setPrice(5999.99);product.setTitle("华为P50就是棒");//2.将文档数据转换为json格式String source = gson.toJson(product);//3.创建索引请求对象 访问哪个索引库、哪个type、指定文档ID//public IndexRequest(String index, String type, String id)IndexRequest request = new IndexRequest("lagou","item",product.getId().toString());request.source(source, XContentType.JSON);//4.发出请求IndexResponse response = restHighLevelClient.index(request,RequestOptions.DEFAULT);System.out.println(response);
}

看下响应：

response = IndexResponse[index=item,type=docs,id=1,version=2,result=created,seqNo=1,primaryTerm=1,shards={"total":2,"successful":2,"failed":0}
]

11.4.3.查看文档

根据rest风格，查看应该是根据id进行get查询，难点是对结果的解析：

/*** 查看文档* @throws IOException*/
@Test
public void testQuery() throws IOException {// 1.初始化GetRequest对象，并指定idGetRequest getRequest = new GetRequest("lagou", "item", "1");// 2.执行查询GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);// 3.取出数据String source = getResponse.getSourceAsString();Product product = gson.fromJson(source, Product.class);System.out.println(product);
}

结果：

Product(id=1, title=华为P50就是棒, category=手机, brand=华为, price=5999.99, images=http://image.huawei.com/1.jpg)

11.4.4.修改文档

新增时，如果传递的id是已经存在的，则会完成修改操作，如果不存在，则是新增。

11.4.5.删除文档

根据id删除：

/*** 删除文档* @throws IOException*/
@Test
public void testDelete() throws IOException {// 1.初始化DeleteRequest对象DeleteRequest deleteRequest = new DeleteRequest("lagou", "item", "id");// 2.执行删除DeleteResponse delete = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);System.out.println(delete);
}

11.5.搜索数据

11.5.1.查询所有match_all

@Test
public void testMatchAll() throws IOException {// 创建搜索对象SearchRequest request = new SearchRequest();// 查询构建工具SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();// 添加查询条件，通过QueryBuilders获取各种查询sourceBuilder.query(QueryBuilders.matchAllQuery());request.source(sourceBuilder);// 搜索SearchResponse response = client.search(request, RequestOptions.DEFAULT);// 解析SearchHits hits = response.getHits();SearchHit[] searchHits = hits.getHits();for (SearchHit hit : searchHits) {// 取出source数据String json = hit.getSourceAsString();// 反序列化Product item = gson.fromJson(json, Item.class);System.out.println("item = " + item);}
}

结果：

Product(id=5, title=荣耀V10, category=手机, brand=华为, price=2799.0, images=http://image.lagou.com/13123.jpg)
Product(id=2, title=坚果手机R1, category=手机, brand=锤子, price=3699.0, images=http://image.lagou.com/13123.jpg)
Product(id=4, title=小米Mix2S, category=手机, brand=小米, price=4299.0, images=http://image.lagou.com/13123.jpg)
Product(id=1, title=小米手机7, category=手机, brand=小米, price=2299.0, images=http://image.lagou.com/13123.jpg)
Product(id=3, title=华为META10, category=手机, brand=华为, price=4499.0, images=http://image.lagou.com/13123.jpg)

注意，上面的代码中，搜索条件是通过sourceBuilder.query(QueryBuilders.matchAllQuery())来添加的。这个query()方法接受的参数是：QueryBuilder接口类型

这个接口提供了很多实现类，分别对应我们在之前中学习的不同类型的查询，例如：term查询、match 查询、range查询、boolean查询等，如下图

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-yIGLRyVK-1637500558858)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–搜索数据–查询所有)]

因此，我们如果要使用各种不同查询，其实仅仅是传递给sourceBuilder.query()方法的参数不同而已。而这些实现类不需要我们去new，官方提供了QueryBuilders工厂帮我们构建各种实现类：

11.5.2.关键字搜索match

其实搜索类型的变化，仅仅是利用QueryBuilders构建的查询对象不同而已，其他代码基本一致：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-IH5s0a0E-1637500558860)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–搜索数据–关键字搜索)]

结果：

item = Item{id=2, title='坚果手机R1', category='手机', brand='锤子', price=3699.0,
images='http://image.lagou.com/13123.jpg'}
item = Item{id=1, title='小米手机7', category='手机', brand='小米', price=3299.0,
images='http://image.lagou.com/13123.jpg'}

因此，我们可以把这段代码封装，然后把查询条件作为参数传递：

private void basicQuery(SearchSourceBuilder sourceBuilder) throws IOException {// 创建搜索对象SearchRequest request = new SearchRequest();request.source(sourceBuilder);// 搜索SearchResponse response = client.search(request,RequestOptions.DEFAULT);// 解析SearchHits hits = response.getHits();SearchHit[] searchHits = hits.getHits();for (SearchHit hit : searchHits) {// 取出source数据String json = hit.getSourceAsString();// 反序列化Product item = gson.fromJson(json, Item.class);System.out.println("item = " + item);}
}

调用封装的方法，并传递查询条件：

/*** 关键字搜索match*/
@Test
public void matchQuery() throws IOException {SearchSourceBuilder builder = new SearchSourceBuilder();// 设置查询类型和查询条件builder.query(QueryBuilders.matchQuery("title","手机"));// 调用基础查询方法baseQuery(builder);
}

11.5.3.范围查询range

RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("price");

与页面上一样，支持下面的范围关键字：

方法	说明
gt(Object from)	大于
gte(Object from)	大于等于
lt(Object from)	小于
lte(Object from)	小于等于

示例：

/*** 范围查询range* price：3600 - 4300*/
@Test
public void rangeQuery() throws IOException {SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();// 指定查询条件和查询类型RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("price");rangeQueryBuilder.gte(3600);rangeQueryBuilder.lte(4300);sourceBuilder.query(rangeQueryBuilder);baseQuery(sourceBuilder);
}

结果：

Product(id=2, title=坚果手机R1, category=手机, brand=锤子, price=3699.0, images=http://image.lagou.com/13123.jpg)
Product(id=4, title=小米Mix2S, category=手机, brand=小米, price=4299.0, images=http://image.lagou.com/13123.jpg)

11.5.4.source过滤

_source：存储原始文档

默认情况下，索引库中所有数据都会返回，如果我们想只返回部分字段，可以通过source filter来控制。

/*** source过滤* @throws IOException*/
@Test
public void sourceFilter() throws IOException {SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();// 指定查询条件和查询类型RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("price");rangeQueryBuilder.gte(3600);rangeQueryBuilder.lte(4300);sourceBuilder.query(rangeQueryBuilder);// source过滤，只保留id、title、pricesourceBuilder.fetchSource(new String[]{"id","title","price"},null);baseQuery(sourceBuilder);
}

关键代码：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-maV9QDFP-1637500558864)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Elasticsearch客户端–搜索数据–source过滤)]

结果：

Product(id=2, title=坚果手机R1, category=null, brand=null, price=3699.0, images=null)
Product(id=4, title=小米Mix2S, category=null, brand=null, price=4299.0, images=null)

11.6.排序

依然是通过sourceBuilder来配置：

/*** 排序*/
@Test
public void sort() throws IOException {// 创建搜索请求对象SearchRequest request = new SearchRequest();// 查询构建工具SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();// 添加查询条件，执行查询类型sourceBuilder.query(QueryBuilders.matchAllQuery());// 执行排序 价格降序排序sourceBuilder.sort("price", SortOrder.DESC);baseQuery(sourceBuilder);
}

结果：

Product(id=3, title=华为META10, category=手机, brand=华为, price=4499.0, images=http://image.lagou.com/13123.jpg)
Product(id=4, title=小米Mix2S, category=手机, brand=小米, price=4299.0, images=http://image.lagou.com/13123.jpg)
Product(id=2, title=坚果手机R1, category=手机, brand=锤子, price=3699.0, images=http://image.lagou.com/13123.jpg)
Product(id=5, title=荣耀V10, category=手机, brand=华为, price=2799.0, images=http://image.lagou.com/13123.jpg)
Product(id=1, title=小米手机7, category=手机, brand=小米, price=2299.0, images=http://image.lagou.com/13123.jpg)

11.7.分页

分页需要视图层传递两个参数给我们：

当前页：page

每页大小：size

而elasticsearch中需要的不是当前页，而是起始位置，还好有公式可以计算出：

from–>起始位置，0表示第一条

起始位置：start = (page - 1) * size

第一页：（1-1）5 = 0

第二页：（2-1）*5 = 5

代码：

/*** 分页* @throws IOException*/
@Test
public void sortAndPage() throws IOException {// 创建搜索请求对象SearchRequest request = new SearchRequest();// 查询构建工具SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();// 添加查询条件，执行查询类型sourceBuilder.query(QueryBuilders.matchAllQuery());// 执行排序 价格降序排序sourceBuilder.sort("price", SortOrder.DESC);// 分页信息int pageNum = 1;int pageSize = 3;int from = (pageNum-1)*pageSize;// 设置分页sourceBuilder.from(from);sourceBuilder.size(3);baseQuery(sourceBuilder);
}

结果：

Product(id=3, title=华为META10, category=手机, brand=华为, price=4499.0, images=http://image.lagou.com/13123.jpg)
Product(id=4, title=小米Mix2S, category=手机, brand=小米, price=4299.0, images=http://image.lagou.com/13123.jpg)
Product(id=2, title=坚果手机R1, category=手机, brand=锤子, price=3699.0, images=http://image.lagou.com/13123.jpg)

当我们传page为2的时候，结果是：

Product(id=5, title=荣耀V10, category=手机, brand=华为, price=2799.0, images=http://image.lagou.com/13123.jpg)
Product(id=1, title=小米手机7, category=手机, brand=小米, price=2299.0, images=http://image.lagou.com/13123.jpg)

12 Spring Data Elasticsearch

接下来我们学习Spring提供的elasticsearch组件：Spring Data Elasticsearch

12.1.什么是SpringDataElasticsearch

Spring Data Elasticsearch（以后简称SDE）是Spring Data项目下的一个子模块。

Spring Data 的使命是给各种数据访问提供统一的编程接口，不管是关系型数据库（如MySQL），还是非关系数据库（如Redis），或者类似Elasticsearch这样的索引数据库。从而简化开发人员的代码，提高开发效率。

Spring Data Elasticsearch的页面：https://projects.spring.io/spring-data-elasticsearch/

特征：

支持Spring的基于 @Configuration 的java配置方式，或者XML配置方式

提供了用于操作ES的便捷工具类 ElasticsearchTemplate 。包括实现文档到POJO之间的自动智能映射。

利用Spring的数据转换服务实现的功能丰富的对象映射

基于注解的元数据映射方式，而且可扩展以支持更多不同的数据格式，可以定义JavaBean：类名、属性

根据持久层接口自动生成对应实现方法，无需人工编写基本操作代码（类似mybatis，根据接口自动得到实现）。当然，也支持人工定制查询

12.2.配置SpringDataElasticsearch

我们在pom文件中，引入SpringDataElasticsearch的启动器：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

然后，只需要在resources下新建application.yml文件，引入elasticsearch的host和port即可：

spring:data:elasticsearch:cluster-name: lagou-elasticcluster-nodes: 127.0.0.1:9301,127.0.0.1:9302,127.0.0.1:9303

需要注意的是，SpringDataElasticsearch底层使用的不是Elasticsearch提供的RestHighLevelClient，而是TransportClient，并不采用Http协议通信，而是访问elasticsearch对外开放的tcp端口，我们之前集群配置中，设置的分别是：9301,9302,9303

添加引导类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class EsApplication {public static void main(String[] args) {SpringApplication.run(EsApplication.class,args);}
}

另外，SpringBoot已经帮我们配置好了各种SDE配置，并且注册了一个ElasticsearchTemplate供我们使用。接下来一起来试试吧。

12.3.索引库操作

12.3.1.创建索引库

准备一个pojo对象

然后准备一个新的实体类，作为下面与索引库对应的文档：

package com.lagou.pojo;import lombok.Data;@Data
public class Product {private Long id;private String title; //标题private String category;// 分类private String brand; // 品牌private Double price; // 价格private String images; // 图片地址
}

我们先创建一个测试类，然后注入ElasticsearchTemplate：

@RunWith(SpringRunner.class)
@SpringBootTest
public class EsclientApplicationTests2 {@Autowiredprivate ElasticsearchTemplate esTemplate;
}

下面是创建索引库的API示例：

/*** 创建索引库*/
@Test
public void createIndex(){// 创建索引的方法template.createIndex(Product.class);
}

发现没有，创建索引库需要指定的信息，比如：索引库名、类型名、分片、副本数量、还有映射信息都没有填写，这是怎么回事呢？

实际上，与我们自定义工具类类似，SDE也是通过实体类上的注解来配置索引库信息的，我们需要在Goods上添加下面的一些注解：

@Data
@Document(indexName = "lagou",type = "product",shards = 3,replicas = 1)
public class Product {@Id // 作为document的idprivate Long id;@Field(type = FieldType.Text,analyzer = "ik_max_word")private String title; // 标题@Field(type = FieldType.Keyword)private String category;// 分类@Field(type = FieldType.Keyword)private String brand; // 品牌@Field(type = FieldType.Double)private Double price; // 价格@Field(type = FieldType.Keyword,index = false)private String images; // 图片地址
}

几个用到的注解：

@Document：声明索引库配置

indexName：索引库名称

type：类型名称，默认是“docs”

shards：分片数量，默认5

replicas：副本数量，默认1

@Id：声明实体类的id

@Field：声明字段属性

type：字段的数据类型

analyzer：指定分词器类型

index：是否创建索引

12.3.2.创建映射

刚才的注解已经把映射关系也配置上了，所以创建映射只需要这样：

/*** 创建映射*/
@Test
public void createType(){// 创建类型映射template.putMapping(Product.class);
}

查看索引库：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cr44gw1U-1637500558866)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Spring Data Elasticsearch–索引库操作–创建映射)]

12.4.索引数据CRUD

SDE的索引数据CRUD并没有封装在ElasticsearchTemplate中，而是有一个叫做ElasticsearchRepository的接口：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bEkiGuvm-1637500558868)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Spring Data Elasticsearch–索引数据CRUD)]

我们需要自定义接口，继承ElasticsearchRespository：

package com.lagou.repository;import com.lagou.pojo.Product;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;/*** 当SDE访问索引库时，需要定义一个持久层的接口去继承ElasticsearchRepository接口即可，无需实现*/
public interface ProductRepository extends ElasticsearchRepository<Product,Long> {
}

12.4.1.创建索引数据

创建索引有单个创建和批量创建之分，先来看单个创建

@Autowired
private ProductRepository productRepository;/*** 创建索引数据*/
@Test
public void insertDocument(){Product product = new Product(1L,"锤子手机","手机","锤子",3299.99,"http://image.zhuizi.com/1.jpg");// 新增productRepository.save(product);System.out.println("save success");
}

再来看批量创建：

/*** 批量创建索引数据*/
@Test
public void insertDocuments(){Product product1 = new Product(2L,"坚果手机","手机","phone",4299.99,"http://image.phone.com/1.jpg");Product product2 = new Product(3L,"华为手机","手机","phone",5299.99,"http://image.phone.com/1.jpg");Product product3 = new Product(4L,"苹果手机","手机","phone",6299.99,"http://image.phone.com/1.jpg");Product product4 = new Product(5L,"索尼手机","手机","phone",7299.99,"http://image.phone.com/1.jpg");List<Product> list = new ArrayList<>();list.add(product1);list.add(product2);list.add(product3);list.add(product4);// 新增productRepository.saveAll(list);System.out.println("save success");
}

通过elasticsearch-head查看：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SaTpBDGi-1637500558870)(E:\MarkDown\拉勾笔记\Lucene&ElasticSeach–Spring Data Elasticsearch–索引数据CRUD–创建索引数据)]

12.4.2.查询索引数据

默认提供了根据id查询，查询所有两个功能：

根据id查询

/*** 根据id查询索引数据*/
@Test
public void findById(){Optional<Product> optional = productRepository.findById(3L);// 取出数据// orElse方法的作用：如果optional中封装的实体对象为空也就是没有从索引库差存储匹配的文档，返回orElse方法的参数Product defaultResult = new Product();defaultResult.setTitle("default data");Product product = optional.orElse(defaultResult);System.out.println(product);
}

结果：

Product(id=3, title=华为手机, category=手机, brand=phone, price=5299.99, images=http://image.phone.com/1.jpg)

查询所有：

/*** 查询所有*/
@Test
public void findAll(){Iterable<Product> iterable = productRepository.findAll();iterable.forEach(System.out::println);
}

结果：

Product(id=2, title=坚果手机, category=手机, brand=phone, price=4299.99, images=http://image.phone.com/1.jpg)
Product(id=4, title=苹果手机, category=手机, brand=phone, price=6299.99, images=http://image.phone.com/1.jpg)
Product(id=5, title=索尼手机, category=手机, brand=phone, price=7299.99, images=http://image.phone.com/1.jpg)
Product(id=1, title=锤子手机, category=手机, brand=锤子, price=3299.99, images=http://image.zhuizi.com/1.jpg)
Product(id=3, title=华为手机, category=手机, brand=phone, price=5299.99, images=http://image.phone.com/1.jpg)

12.4.3.自定义方法查询

ProductRepository提供的查询方法有限，但是它却提供了非常强大的自定义查询功能：

只要遵循SpringData提供的语法，我们可以任意定义方法声明：

public interface ProductRepository extends ElasticsearchRepository<Product,Long> {/*** 根据价格区间查询* @param from 开始价格* @param to 结束价格* @return 符合条件的products*/List<Product> findByPriceBetween(Double from,Double to);
}

无需写实现，SDE会自动帮我们实现该方法，我们只需要用即可：

/*** 自定义方法查询*/
@Test
public void findByPrice(){List<Product> list = productRepository.findByPriceBetween(4000.00, 7000.00);//System.out.println(list.size());list.forEach(System.out::println);
}

结果：

Product(id=2, title=坚果手机, category=手机, brand=phone, price=4299.99, images=http://image.phone.com/1.jpg)
Product(id=4, title=苹果手机, category=手机, brand=phone, price=6299.99, images=http://image.phone.com/1.jpg)
Product(id=3, title=华为手机, category=手机, brand=phone, price=5299.99, images=http://image.phone.com/1.jpg)

支持的一些语法示例：

Keyword	Sample	Elasticsearch Query String
And	findByNameAndPrice	{“bool” : {“must” : [ {“field” : {“name” : “?”}}, {“field” : {“price” : “?”}} ]}}
Or	findByNameOrPrice	{“bool” : {“should” : [ {“field” : {“name” : “?”}}, {“field” : {“price” : “?”}} ]}}
Is	findByName	{“bool” : {“must” : {“field” : {“name” : “?”}}}}
Not	findByNameNot	{“bool” : {“must_not” : {“field” : {“name” : “?”}}}}
Between	findByPriceBetween	{“bool” : {“must” : {“range” : {“price” : {“from” : ?,“to” : ?,“include_lower” : true,“include_upper” : true}}}}}
LessThanEqual	findByPriceLessThan	{“bool” : {“must” : {“range” : {“price” : {“from” : null,“to” : ?,“include_lower” : true,“include_upper” : true}}}}}
GreaterThanEqual	findByPriceGreaterThan	{“bool” : {“must” : {“range” : {“price” : {“from” : ?,“to” : null,“include_lower” : true,“include_upper” : true}}}}}
Before	findByPriceBefore	{“bool” : {“must” : {“range” : {“price” : {“from” : null,“to” : ?,“include_lower” : true,“include_upper” : true}}}}}
After	findByPriceAfter	{“bool” : {“must” : {“range” : {“price” : {“from” : ?,“to” : null,“include_lower” : true,“include_upper” : true}}}}}
Like	findByNameLike	{“bool” : {“must” : {“field” : {“name” : {“query” : “? *”,“analyze_wildcard” : true}}}}}
StartingWith	findByNameStartingWith	{“bool” : {“must” : {“field” : {“name” : {“query” : “? *”,“analyze_wildcard” : true}}}}}
EndingWith	findByNameEndingWith	{“bool” : {“must” : {“field” : {“name” : {“query” : “*?”,“analyze_wildcard” : true}}}}}
Contains/Containing	findByNameContaining	{“bool” : {“must” : {“field” : {“name” : {“query” : “? ”,“analyze_wildcard” : true}}}}}
In	findByNameIn(Collectionnames)	{“bool” : {“must” : {“bool” : {“should” : [ {“field” : {“name” : “?”}}, {“field” : {“name” : “?”}} ]}}}}
NotIn	findByNameNotIn(Collectionnames)	{“bool” : {“must_not” : {“bool” : {“should” : {“field” : {“name” : “?”}}}}}}
Near	findByStoreNear	Not Supported Yet !
True	findByAvailableTrue	{“bool” : {“must” : {“field” : {“available” : true}}}}
False	findByAvailableFalse	{“bool” : {“must” : {“field” : {“available” : false}}}}
OrderBy	findByAvailableTrueOrderByNameDesc	{“sort” : [{ “name” : {“order” : “desc”} }],“bool” : {“must” : {“field” : {“available” : true}}}}

12.5.原生查询

如果觉得上述接口依然不符合你的需求，SDE也支持原生查询，这个时候还是使用 ElasticsearchTemplate

而查询条件的构建是通过一个名为 NativeSearchQueryBuilder 的类来完成的，不过这个类的底层还是使用的原生API中的 QueryBuilders 、 AggregationBuilders 、 HighlightBuilders等工具。

需求：

查询title中包含小米手机的商品，以价格升序排序，分页查询：每页展示2条，查询第1页。

对查询结果进行聚合分析：获取品牌及个数

示例：

/*** 原生查询* 需求：* 查询title中包含小米手机的商品，以价格升序排序，分页查询：每页展示2条，查询第1页。* 对查询结果进行聚合分析：获取品牌及个数*/
@Test
public void nativeQuery(){// 1.构建一个原生查询器NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();// 2.source过滤// 2.1 参数：final String[] includes, final String[] excludes// 如果不想执行source过滤可以将改行注释queryBuilder.withSourceFilter(new FetchSourceFilter(new String[0],new String[0]));// 3.查询条件queryBuilder.withQuery(QueryBuilders.matchQuery("title","小米手机"));// 4.设置分页并排序规则queryBuilder.withPageable(PageRequest.of(0,3, Sort.by(Sort.Direction.DESC,"price")));// 5.高亮// 6.聚合queryBuilder.addAggregation(AggregationBuilders.terms("brandAgg").field("brand"));// 7.查询AggregatedPage<Product> result = template.queryForPage(queryBuilder.build(), Product.class);// 获取结果long total = result.getTotalElements();// 获得页码int totalPages = result.getTotalPages();// 获得本页的数据集合List<Product> content = result.getContent();System.out.println(total+" "+totalPages);content.stream().forEach(product -> System.out.println(product));// 获得聚合结果Aggregations aggregations = result.getAggregations();Terms terms = aggregations.get("brandAgg");// 获取桶并且遍历桶中的内容terms.getBuckets().forEach(b->{System.out.println("品牌："+b.getKeyAsString());System.out.println("个数："+b.getDocCount());});
}

注：上述查询不支持高亮结果。

高亮展示：

1、自定义搜索结果映射

/*** 自定义结果映射，处理高亮*/
public class ESSearchResultMapper implements SearchResultMapper {/*** 完成结果映射* 操作的重点应该是将原有的结果：_source取出来，放入高亮的数据* @param response* @param clazz* @param pageable* @param <T>* @return AggregatedPage需要三个参数进行构建：pageable，List<product>，总记录数*/@Overridepublic <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {// 获得总记录数long totalHits = response.getHits().getTotalHits();// 记录列表List<T> list = new ArrayList<>();// 获取原始的搜索结果SearchHits hits = response.getHits();for (SearchHit hit : hits) {if (hits.getHits().length <= 0){return null;}// 获得_source属性中的数据Map<String, Object> map = hit.getSourceAsMap();// 获得高亮的字段Map<String, HighlightField> highlightFields = hit.getHighlightFields();// 每一个高亮字段都需要进行设置for (Map.Entry<String,HighlightField> highlightField : highlightFields.entrySet()){// 获得高亮的key：高亮字段String key = highlightField.getKey();// 获得value：高亮之后的效果HighlightField value = highlightField.getValue();// 将高亮字段和文本效果放入到map中map.put(key,value.getFragments()[0].toString());}// 将map转换为对象Gson gson = new Gson();// map-->jsonString-->对象T t = gson.fromJson(gson.toJson(map), clazz);list.add(t);}// 返回的是带分页的结果return new AggregatedPageImpl<>(list,pageable,totalHits);}
}

2、高亮实现：

@Test
public void nativeQuery(){// 1.构建一个原生查询器NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();// 2.source过滤// 2.1 参数：final String[] includes, final String[] excludes// 如果不想执行source过滤可以将改行注释queryBuilder.withSourceFilter(new FetchSourceFilter(new String[0],new String[0]));// 3.查询条件queryBuilder.withQuery(QueryBuilders.matchQuery("title","小米手机"));// 4.设置分页并排序规则queryBuilder.withPageable(PageRequest.of(0,3, Sort.by(Sort.Direction.DESC,"price")));// 5.高亮HighlightBuilder.Field field = new HighlightBuilder.Field("title");field.preTags("<font style='color:red'>");field.postTags("</fond>");queryBuilder.withHighlightFields(field);// 6.聚合queryBuilder.addAggregation(AggregationBuilders.terms("brandAgg").field("brand"));// 7.查询AggregatedPage<Product> result = template.queryForPage(queryBuilder.build(), Product.class,new ESSearchResultMapper());// 获取结果long total = result.getTotalElements();// 获得页码int totalPages = result.getTotalPages();// 获得本页的数据集合List<Product> content = result.getContent();System.out.println(total+" "+totalPages);content.stream().forEach(product -> System.out.println(product));
}

kafka

1. 消息队列(MQ)

1.1 什么是消息队列

消息队列不知道大家看到这个词的时候，会不会觉得它是一个比较高端的技术。消息队列，一般我们会简称它为MQ（Message Queue）。

消息队列是一种帮助开发人员解决系统间异步通信的中间件，常用于解决系统解耦和请求的削峰平谷的问题。

队列（Queue）：

Queue 是一种先进先出的数据结构，容器

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hNt9Dd4g-1637500558872)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–队列（Queue）)

消息（Message）：

不同应用之间传送的数据

消息队列：

我们可以把消息队列比作是一个存放消息的容器，当我们需要使用消息的时候可以取出消息供自己使用。消息队列是分布式系统中重要的组件，使用消息队列主要是为了通过异步处理提高系统性能和削峰、降低系统耦合性。队列 Queue 是一种先进先出的数据结构，所以消费消息时也是按照顺序来消费的。比如生产者发送消息1,2,3…对于消费者就会按照1,2,3…的顺序来消费。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mwzG1rXK-1637500558874)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–传统方式)

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hY8S9xVC-1637500558877)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–MQ方式)

1.2 消息队列的应用场景

消息队列在实际应用中包括如下四个场景：

应用耦合：多应用间通过消息队列对同一消息进行处理，避免调用接口失败导致整个过程失败；
异步处理：多应用对消息队列中同一消息进行处理，应用间并发处理消息，相比串行处理，减少处理时间；
限流削峰：广泛应用于秒杀或抢购活动中，避免流量过大导致应用系统挂掉的情况；
消息驱动的系统：系统分为消息队列、消息生产者、消息消费者，生产者负责产生消息，消费者(可能有多个)负责对消息进行处理

1.2.1 异步处理

具体场景：

用户为了使用某个应用，进行注册，系统需要发送注册邮件并验证短信。

对这两个子系统操作的处理方式有两种：串行及并行。

涉及到三个子系统：注册系统、邮件系统、短信系统

1）串行方式：新注册信息生成后，先发送注册邮件，再发送验证短信；

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ThoVDjQA-1637500558879)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–异步处理–串行方式)

在这种方式下，需要最终发送验证短信后再返回给客户端。

并行处理：新注册信息写入后，由发短信和发邮件并行处理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DB1bBCs5-1637500558880)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–异步处理–并行处理)

在这种方式下，发短信和发邮件需处理完成后再返回给客户端。

假设以上三个子系统处理的时间均为50ms，且不考虑网络延迟，则总的处理时间：

串行：50+50+50=150ms

并行：50+50 = 100ms

如果引入消息队列, 在来看整体的执行效率：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7KoUWge7-1637500558882)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–异步处理–消息队列)

在写入消息队列后立即返回成功给客户端，则总的响应时间依赖于写入消息队列的时间，而写入消息队列的时间本身是可以很快的，基本可以忽略不计，因此总的处理时间为50ms，相比串行提高了2倍，相比并行提高了一倍；

1.2.2 应用耦合

具体场景：

用户使用QQ相册上传一张图片，人脸识别系统会对该图片进行人脸识别。

一般的做法是，服务器接收到图片后，图片上传系统立即调用人脸识别系统，调用完成后再返回成功，如下图所示：

调用方式：webService、Http协议（HttpClient、RestTemplate)、Tcp协议（Dubbo）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SjktkcKH-1637500558884)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–应用耦合01)

该方法有如下缺点：

1) 人脸识别系统被调失败，导致图片上传失败；
2) 延迟高，需要人脸识别系统处理完成后，再返回给客户端，即使用户并不需要立即知道结果；
3) 图片上传系统与人脸识别系统之间互相调用，需要做耦合；

若使用消息队列：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ni8gBADj-1637500558886)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–应用耦合02)

客户端上传图片后，图片上传系统将图片信息批次写入消息队列，直接返回成功；

人脸识别系统则定时从消息队列中取数据，完成对新增图片的识别。

图片上传系统并不需要关心人脸识别系统是否对这些图片信息的处理、以及何时对这些图片信息进行处理。

事实上，由于用户并不需要立即知道人脸识别结果，人脸识别系统可以选择不同的调度策略，按照闲时、忙时、正常时间，对队列中的图片信息进行处理。

1.2.3 限流削峰

具体场景：

购物网站开展秒杀活动，一般由于瞬时访问量过大，服务器接收过大，会导致流量暴增，相关系统无法处理请求甚至崩溃。

而加入消息队列后，系统可以从消息队列中取数据，相当于消息队列做了一次缓冲。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZQ4ahAs1-1637500558888)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]-- 限流削峰)

该方法有如下优点：

请求先入消息队列，而不是由业务处理系统直接处理，做了一次缓冲,极大地减少了业务处理系统的压力；

队列长度可以做限制，事实上，秒杀时，后入队列的用户无法秒杀到商品，这些请求可以直接被抛弃，返回活动已结束或商品已售完信息；

1.2.4 消息事件驱动的系统

具体场景：

用户新上传了一批照片 ---->人脸识别系统需要对这个用户的所有照片进行聚类 -------> 由对账系统重新生成用户的人脸索引（加快查询）。

这三个子系统间由消息队列连接起来，前一个阶段的处理结果放入队列中，后一个阶段从队列中获取消息继续处理。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-skfjEqHm-1637500558889)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–消息事件驱动的系统)

该方法有如下优点：

避免了直接调用下一个系统导致当前系统失败；

每个子系统对于消息的处理方式可以更为灵活，可以选择收到消息时就处理，可以选择定时处理，也可以划分时间段按不同处理速度处理；

1.3 消息队列的两种模式

消息队列包括两种模式，点对点模式（point to point， queue）和发布/订阅模式（publish/subscribe，topic）

1.3.1 点对点模式

点对点模式下包括三个角色

消息队列

发送者 (生产者)

接收者（消费者）

每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，可以放在内存中也可以持久化，直到他们被消费或超时。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GzmtEL2A-1637500558891)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–消息队列的两种模式–点对点模式)

点对点模式特点：

每个消息只有一个消费者，一旦被消费，消息就不再在消息队列中；

发送者和接收者间没有依赖性，发送者发送消息之后，不管有没有接收者在运行，都不会影响到发送者下次发送消息；

接收者在成功接收消息之后需向队列应答成功，以便消息队列删除当前接收的消息；

1.3.2 发布/订阅模式

发布/订阅模式下包括三个角色：

角色主题（Topic）

发布者(Publisher)

订阅者(Subscriber)

消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被多个订阅者消费。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-a3SQ7Ttu-1637500558893)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–消息队列的两种模式–发布-订阅模式)

发布/订阅模式特点：

每个消息可以有多个订阅者；

发布者和订阅者之间有时间上的依赖性

为了消费消息，订阅者必须保持在线运行。

1.4 消息队列实现机制

1.4.1 JMS

JMS（JAVA Message Service，Java消息服务）是一个Java平台中关于面向消息中间件的API

允许应用程序组件基于JavaEE平台创建、发送、接收和读取消息

是一个消息服务的标准或者说是规范，是 Java 平台上有关面向消息中间件的技术规范

便于消息系统中的 Java 应用程序进行消息交换，并且通过提供标准的产生、发送、接收消息的接口，简化企业应用的开发。

JMS 消息机制主要分为两种模型：PTP 模型和 Pub/Sub 模型。

实现产品：Apache ActiveMQ

1.4.2 AMQP

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aFuWFGeR-1637500558895)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–消息队列实现机制–AMQP)

1.4.2 JMS VS AMQP

	JMS	AMQP
定义	Java api	Wire-protocol
跨语言	否	是
跨平台	否	是
Model	提供两种消息模型/模式：（1）、Peer-2- Peer（2）、Pub/sub	提供了五种消息模型：（1）、direct exchange（2）、fanout exchange（3）、topic change（4）、headers exchange（5）、system exchange本质来讲，后四种和JMS的pub/sub模型没有太大差别，仅是在路由机制上做了更详细的划分；
支持消息类型	多种消息类型：TextMessage、 MapMessage、BytesMessage、 StreamMessage、ObjectMessage、 Message （只有消息头和属性）	byte[]当实际应用时，有复杂的消息，可以将消息序列化后发送。
综合评价	MS 定义了JAVA API层面的标准；在java 体系中，多个client均可以通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差；	AMQP定义了wire-level层的协议标准；天然具有跨平台、跨语言特性。

1.5常见的消息队列产品

RabbitMQ

RabbitMQ 2007年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。

ActiveMQ

ActiveMQ是由Apache出品，ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。它非常快速，支持多种语言的客户端和协议，而且可以非常容易的嵌入到企业的应用环境中，并有许多高级功能

RocketMQ

RocketMQ RocketMQ出自阿里公司的开源产品，用 Java 语言实现，在设计时参考了 Kafka，并做出了自己的一些改进，消息可靠性上比 Kafka 更好。RocketMQ在阿里集团被广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理等

Kafka

Apache Kafka是一个分布式消息发布订阅系统。它最初由LinkedIn公司基于独特的设计实现为一个分布式的提交日志系统( a distributed commit log)，之后成为Apache项目的一部分。Kafka系统快速、可扩展并且可持久化。它的分区特性，可复制和可容错都是其不错的特性。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SgwIARz1-1637500558897)(E:\MarkDown\拉勾笔记\kafka–消息队列(MQ)]–常见的消息队列产品)

综合上面的材料得出以下两点：

（1）中小型软件公司：

建议选RabbitMQ。一方面，erlang语言天生具备高并发的特性，而且他的管理界面用起来十分方便。正所谓，成也萧何，败也萧何！他的弊端也在这里，虽然RabbitMQ是开源的，然而国内有几个能定制化开发erlang的程序员呢？所幸，RabbitMQ的社区十分活跃，可以解决开发过程中遇到的bug，这点对于中小型公司来说十分重要。不考虑rocketmq和kafka的原因是，一方面中小型软件公司不如互联网公司，数据量没那么大，选消息中间件，应首选功能比较完备的，所以kafka排除。不考虑rocketmq的原因是，rocketmq是阿里出品，如果阿里放弃维护rocketmq，中小型公司一般抽不出人来进行rocketmq的定制化开发，因此不推荐。

（2）大型软件公司：

根据具体使用在rocketMq和kafka之间二选一。一方面，大型软件公司，具备足够的资金搭建分布式环境，也具备足够大的数据量。针对rocketMQ，大型软件公司也可以抽出人手对rocketMQ进行定制化开发，毕竟国内有能力改JAVA源码的人，还是相当多的。至于kafka，根据业务场景选择，如果有日志采集功能，肯定是首选kafka了。

2. kafka的基本介绍

2.1 什么是Kafka

官网：http://kafka.apache.org/

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做 MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

主要应用场景是：日志收集系统和消息系统。

Kafka主要设计目标如下：

以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。

算法复杂度：时间复杂度和空间复杂度

以时间复杂度为O(1)的方式：常数时间运行和数据量的增长无关，假如操作一个链表，那么无论链表的大还是小，操作时间是一样的

高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。

支持普通服务器每秒百万级写入请求

Memory mapped Files

支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。

同时支持离线数据处理和实时数据处理。

Scale out：支持在线水平扩展

2.2 kafka的特点

（1）解耦。Kafka具备消息系统的优点，只要生产者和消费者数据两端遵循接口约束，就可以自行扩展或修改数据处理的业务过程。

（2）高吞吐量、低延迟。即使在非常廉价的机器上，Kafka也能做到每秒处理几十万条消息，而它的延迟最低只有几毫秒。

（3）持久性。Kafka可以将消息直接持久化在普通磁盘上，且磁盘读写性能优异。

（4）扩展性。Kafka集群支持热扩展，Kafka集群启动运行后，用户可以直接向集群添。

（5）容错性。Kafka会将数据备份到多台服务器节点中，即使Kafka集群中的某一台加新的Kafka服务节点宕机，也不会影响整个系统的功能。

（6）支持多种客户端语言。Kafka支持Java、.NET、PHP、Python等多种语言。

（7）支持多生产者和多消费者。

2.3 kafka的主要应用场景

消息处理（MQ）

KafKa可以代替传统的消息队列软件，使用KafKa来实现队列有如下优点

KafKa的append来实现消息的追加，保证消息都是有序的有先来后到的顺序

稳定性强队列在使用中最怕丢失数据，KafKa能做到理论上的写成功不丢失

分布式容灾好

容量大相对于内存队列，KafKa的容量受硬盘影响

数据量不会影响到KafKa的速度

分布式日志系统(Log)

在很多时候我们需要对一些庞大的数据进行存留，日志存储这块会遇到巨大的问题，日志不能丢，日志存文件不好找，定位一条消息成本高（遍历当天日志文件），实时显示给用户难，这几类问题KafKa都能游刃有余

KafKa的集群备份机制能做到n/2的可用，当n/2以下的机器宕机时存储的日志不会丢失

KafKa可以对消息进行分组分片

KafKa非常容易做到实时日志查询

流式处理

流式处理就是指实时地处理一个或多个事件流。

流式的处理框架(spark，storm，flink) 从主题中读取数据，对其进行处理，并将处理后的结果数据写入新的主题，供用户和应用程序使用，kafka的强耐久性在流处理的上下文中也非常的有用

3. kafka的架构

3.1 架构案例

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Sc3XwR57-1637500558900)(E:\MarkDown\拉勾笔记\kafka–kafka的架构–架构案例)]

**Kafka Cluster：**由多个服务器组成。每个服务器单独的名字broker（掮客）。

**kafka broker：**kafka集群中包含的服务器

**Kafka Producer：**消息生产者、发布消息到 kafka 集群的终端或服务。

**Kafka consumer：**消息消费者、负责消费数据。

Kafka Topic: 主题，一类消息的名称。存储数据时将一类数据存放在某个topci下，消费数据也是消费一类数据。

订单系统：创建一个topic，叫做order。

用户系统：创建一个topic，叫做user。

商品系统：创建一个topic，叫做product。

注意：Kafka的元数据都是存放在zookeeper中。

3.2 架构剖析

kafka架构的内部细节剖析:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-25zIK0OQ-1637500558902)(E:\MarkDown\拉勾笔记\kafka–kafka的架构–架构剖析01)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3DEgLSk2-1637500558903)(E:\MarkDown\拉勾笔记\kafka–kafka的架构–架构剖析02)]

说明：kafka支持消息持久化，消费端为拉模型来拉取数据，消费状态和订阅关系有客户端负责维护，消息消费完后，不会立即删除，会保留历史消息。因此支持多订阅时，消息只会存储一份就可以了。

Broker：kafka集群中包含一个或者多个服务实例，这种服务实例被称为Broker

Topic：每条发布到kafka集群的消息都有一个类别，这个类别就叫做Topic

Partition：分区，物理上的概念，每个topic包含一个或多个partition，一个partition对应一个文件夹，这个文件夹下存储partition的数据和索引文件，每个partition内部是有序的

3.3 关系解释

Topic & Partition

Topic 就是数据主题，是数据记录发布的地方，可以用来区分业务系统。

Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。

一个topic为一类消息，每条消息必须指定一个topic。

对于每一个topic， Kafka集群都会维持一个分区日志。如下图

每个分区都是有序且顺序不可变的记录集，并且不断地追加到结构化的commit log文件。

分区中的每一个记录都会分配一个id号来表示顺序，称之为offset，offset用来唯一的标识分区中每一条记录。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tM8h3u04-1637500558905)(E:\MarkDown\拉勾笔记\kafka–kafka的架构–关系解释)]

在每一个消费者中唯一保存的元数据是offset（偏移量）即消费在log中的位置，偏移量由消费者所控制：通常在读取记录后，消费者会以线性的方式增加偏移量，但是实际上，由于这个位置由消费者控制，所以消费者可以采用任何顺序来消费记录。例如，一个消费者可以重置到一个旧的偏移量，从而重新处理过去的数据；也可以跳过最近的记录，从"现在"开始消费。

这些细节说明Kafka消费者是非常廉价的—消费者的增加和减少，对集群或者其他消费者没有多大的影响。

4. kafka集群环境搭建

ZooKeeper 作为给分布式系统提供协调服务的工具被 kafka 所依赖。在分布式系统中，消费者需要知道有哪些生产者是可用的，而如果每次消费者都需要和生产者建立连接并测试是否成功连接，那效率也太低了，显然是不可取的。而通过使用 ZooKeeper 协调服务，Kafka 就能将 Producer，Consumer，Broker 等结合在一起，同时借助 ZooKeeper，Kafka就能够将所有组件在无状态的条件下建立起生产者和消费者的订阅关系，实现负载均衡。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-pE9G7GCa-1637500558909)(E:\MarkDown\拉勾笔记\kafka–. kafka集群环境搭建)]

4.1 准备工作

环境准备：

准备三台服务器, 安装jdk1.8 ,其中每一台虚拟机的hosts文件中都需要配置如下的内容

192.168.44.134 node1
192.168.44.135 node2
192.168.44.136 node3

实现方式：

1.将原有学习Docker时的Centos7克隆，克隆出一台虚拟机

2、修改IP地址

修改网卡配置文件 vi /etc/sysconfig/network-scripts/ifcfg-ens33

（1）bootproto=static，表示使用静态IP

（2）onboot=yes，表示将网卡设置为开机启用

（3）将原有的原有IP修改为192.168.44.134

TYPE=Ethernet
PROXY_METHOD=none
BROWSER_ONLY=no
BOOTPROTO=static
DEFROUTE=yes
IPV4_FAILURE_FATAL=no
IPV6INIT=yes
IPV6_AUTOCONF=yes
IPV6_DEFROUTE=yes
IPV6_FAILURE_FATAL=no
IPV6_ADDR_GEN_MODE=stable-privacy
NAME=ens33
UUID=b8fd5718-51f5-48f8-979b-b9f1f7a5ebf2
DEVICE=ens33
ONBOOT=yes
IPADDR=192.168.44.134
GATEWAY=192.168.44.2
NETMASK=255.255.255.0
NM_CONTROLLED=no
DNS1=8.8.8.8
DNS2=8.8.4.4

（4）重启网络服务

service network restart

（5）安装目录创建

安装包存放的目录：/export/software
安装程序存放的目录：/export/servers
数据目录：/export/data
日志目录：/export/logs
创建各级目录命令:mkdir -p /export/servers/
mkdir -p /export/software/
mkdir -p /export/data/
mkdir -p /export/logs/

（6）修改host

执行命令“cd /etc/”进入服务器etc目录；

执行命令“vi hosts”编辑hosts文件；

输入你要修改的内容：

127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6192.168.44.134 node1
192.168.44.135 node2
192.168.44.136 node3

执行命令”/etc/init.d/network restart“ 重启hosts；

执行命令”cat /etc/hosts“可以查看到hosts文件修改成功。

4.2 Zookeeper集群搭建

1.Linux安装JDK，三台Linux都安装。

上传JDK到linux

上传文件两种方式：使用SSH方式、使用CRT方式

使用CRT需要先在Linux虚拟机上安装lrzsz上传工具，安装方式： yum install -y lrzsz

安装lrzsz之后，是需要在在Linux命令行中输入：rz ，就可以弹出一个文件上传窗口

安装并配置JDK

# 使用rpm安装JDK
rpm -ivh jdk-8u261-linux-x64.rpm# 默认的安装路径是/usr/java/jdk1.8.0_261-amd64
# 配置JAVA_HOME
vi /etc/profile# 文件最后添加两行
export JAVA_HOME=/usr/java/jdk1.8.0_261-amd64
export PATH=$PATH:$JAVA_HOME/bin# 退出vi，使配置生效
source /etc/profile

查看JDK是否正确安装

java -version

2.Linux 安装Zookeeper，三台Linux都安装，以搭建Zookeeper集群

上传zookeeper-3.4.14.tar.gz

解压并配置zookeeper（配置data目录，集群节点）

# node1操作
# 解压到/opt目录
tar -zxf zookeeper-3.4.14.tar.gz -C /opt# 配置
cd /opt/zookeeper-3.4.14/conf# 配置文件重命名后生效
cp zoo_sample.cfg zoo.cfg#编辑
vi zoo.cfg
# 设置数据目录
dataDir=/var/lagou/zookeeper/data
# 添加 配置zookeeper集群节点
server.1=node1:2881:3881
server.2=node2:2881:3881
server.3=node3:2881:3881# 退出vim
mkdir -p /var/lagou/zookeeper/data
echo 1 > /var/lagou/zookeeper/data/myid# 配置环境变量
vi /etc/profile
# 添加
export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14
export PATH=$PATH:$ZOOKEEPER_PREFIX/bin
export ZOO_LOG_DIR=/var/lagou/zookeeper/log# 退出vim，让配置生效
source /etc/profile

node2配置

# 配置环境变量
vim /etc/profile
# 在配置JDK环境变量基础上，添加内容
export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14
export PATH=$PATH:$ZOOKEEPER_PREFIX/bin
export ZOO_LOG_DIR=/var/lagou/zookeeper/log# 退出vim，让配置生效
source /etc/profilemkdir -p 192.168.44.136
echo 2 > /var/lagou/zookeeper/data/myid

node3配置

# 配置环境变量
vim /etc/profile
# 在配置JDK环境变量基础上，添加内容
export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14
export PATH=$PATH:$ZOOKEEPER_PREFIX/bin
export ZOO_LOG_DIR=/var/lagou/zookeeper/log# 退出vim，让配置生效
source /etc/profilemkdir -p /var/lagou/zookeeper/data
echo 3 > /var/lagou/zookeeper/data/myid

启动zookeeper

# 在三台Linux上启动Zookeeper
[root@node1 ~]# zkServer.sh start
[root@node2 ~]# zkServer.sh start
[root@node3 ~]# zkServer.sh start# 在三台Linux上查看Zookeeper的状态
[root@node1 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.14/bin/../conf/zoo.cfg
Mode: follower[root@node2 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.14/bin/../conf/zoo.cfg
Mode: leader[root@node3 ~]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.14/bin/../conf/zoo.cfg
Mode: follower

4.3 下载安装包

中文网站: http://kafka.apachecn.org/

英文网站: http://kafka.apache.org/

由于kafka是scala语言编写的，基于scala的多个版本，kafka发布了多个版本。

其中2.11是推荐版本

推荐直接使用资料中的版本即可

4.4 上传安装包并解压

使用 rz 命令将安装包上传至 /export/software
1) 切换目录上传安装包
cd /export/software
rz # 选择对应安装包上传即可2) 解压安装包到指定目录下
tar -zxvf kafka_2.11-1.0.0.tgz -C /export/servers/
cd /export/servers/3) 重命名(由于名称太长)
mv kafka_2.11-1.0.0 kafka

4.5 修改kafka的核心配置文件

cd /export/servers/kafka/config/
vi server.properties主要修改一下6个地方:1) broker.id 需要保证每一台kafka都有一个独立的broker2) log.dirs 数据存放的目录3) zookeeper.connect zookeeper的连接地址信息4) delete.topic.enable 是否直接删除topic5) host.name 主机的名称6) 修改: listeners=PLAINTEXT://node1:9092
#broker.id 标识了kafka集群中一个唯一broker。
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600# 存放生产者生产的数据 数据一般以topic的方式存放
log.dirs=/export/data/kafka
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000# zk的信息
zookeeper.connect=node1:2181,node2:2181,node3:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0delete.topic.enable=true
host.name=node1

4.6 将配置好的kafka分发到其他二台主机

cd /export/servers
scp -r kafka/ node2:$PWD
scp -r kafka/ node3:$PWD

Linux scp 命令用于 Linux 之间复制文件和目录。

scp 是 secure copy 的缩写，scp 是 linux 系统下基于 ssh 登陆进行安全的远程文件拷贝命令。

拷贝后，需要修改每一台的broker.id 和 host.name和listeners

ip为134的服务器: broker.id=0 , host.name=node1 listeners=PLAINTEXT://node1:9092
ip为135的服务器: broker.id=1 , host.name=node2 listeners=PLAINTEXT://node2:9092
ip为136的服务器: broker.id=2 , host.name=node3 listeners=PLAINTEXT://node3:9092

在每一台的服务器执行创建数据文件的命令

mkdir -p /export/data/kafka

4.7 启动集群

注意事项：在kafka启动前，一定要让zookeeper启动起来

cd /export/servers/kafka/bin
#前台启动
./kafka-server-start.sh /export/servers/kafka/config/server.properties#后台启动
nohup ./kafka-server-start.sh /export/servers/kafka/config/server.properties 2>&1 &注意：可以启动一台broker，单机版。也可以同时启动三台broker，组成一个kafka集群版#kafka停止
./kafka-server-stop.sh

可以通过 jps 查看 kafka进程是否已经启动了

登录的前提是, 通过jps是可以看到kafka的进程
登录zookeeper: zkCli.sh
执行: ls /brokers/ids

4.8 Docker环境下的Kafka集群搭建

hostname	ip addr	port	listener
zoo1	192.168.0.11	2184:2181
zoo2	192.168.0.12	2185:2181
zoo3	192.168.0.13	2186:2181
kafka1	192.168.0.14	9092:9092	kafka1
kafka2	192.168.0.15	9093:9092	kafka1
kafka3	192.168.0.16	9094:9092	kafka1
kafka-manager	192.168.0.17	9000:9000
宿主机	192.168.44.20

4.8.1 准备工作

1) 克隆VM，修改IP地址为192.168.44.20

修改网络配置： vi /etc/sysconfig/network-scrpits/ifcfg-ens33

TYPE=Ethernet
PROXY_METHOD=none
BROWSER_ONLY=no
BOOTPROTO=static
DEFROUTE=yes
IPV4_FAILURE_FATAL=no
IPV6INIT=yes
IPV6_AUTOCONF=yes
IPV6_DEFROUTE=yes
IPV6_FAILURE_FATAL=no
IPV6_ADDR_GEN_MODE=stable-privacy
NAME=ens33
UUID=b8fd5718-51f5-48f8-979b-b9f1f7a5ebf2
DEVICE=ens33
ONBOOT=yes
IPADDR=192.168.44.20
GATEWAY=192.168.44.2
NETMASK=255.255.255.0
NM_CONTROLLED=no
DNS1=8.8.8.8
DNS2=8.8.4.4

2) 安装docker - compose

Compose 是用于定义和运行多容器 Docker 应用程序的工具。

如果我们还是使用原来的方式操作docker，那么就需要下载三个镜像：Zookeeper、Kafka、Kafka-Manager，需要对Zookeeper安装三次并配置集群、需要对Kafka安装三次，修改配置文件，Kafka-Manager安装一次，但是需要配置端口映射机器Zookeeper、Kafka容器的信息。但是引入Compose之后可以使用yaml格式的配置文件配置好这些信息，每个image只需要编写一个yaml文件，可以在文件中定义集群信息、端口映射等信息，运行该文件即可创建完成集群。

通过Compose，您可以使用 YML 文件来配置应用程序需要的所有服务。然后，使用一个命令，就可以从 YML 文件配置中创建并启动所有服务。

Compose 使用的两个步骤：

使用 docker-compose.yml 定义构成应用程序的服务，这样它们可以在隔离环境中一起运行。

执行 docker-compose up 命令来启动并运行整个应用程序。

#curl 是一种命令行工具，作用是发出网络请求，然后获取数据
curl -L https://github.com/docker/compose/releases/download/1.8.0/run.sh > /usr/local/bin/docker-compose#chmod（change mode）命令是控制用户对文件的权限的命令
chmod +x /usr/local/bin/docker-compose#查看版本
docker-compose --version

3）拉取镜像

#拉取Zookeeper镜像
docker pull zookeeper:3.4
#拉取kafka镜像
docker pull wurstmeister/kafka
#拉取kafka-manager镜像
docker pull sheepkiller/kafka-manager:latest

4) 创建集群网络

基于Linux宿主机而工作的，也是在Linux宿主机创建，创建之后Docker容器中的各个应用程序可以使用该网络。

#创建
docker network create --driver bridge --subnet 192.168.0.0/24 --gateway 192.168.0.1 kafka
#查看
docker network ls

5）网络设置

新建网段之后可能会出现：WARNING: IPv4 forwarding is disabled. Networking will not work.

解决方式：

第一步：在宿主机上执行： echo “net.ipv4.ip_forward=1” >>/usr/lib/sysctl.d/00-system.conf

第二步：重启network和docker服务

[root@localhost /]# systemctl restart network && systemctl restart docker

4.8.2 搭建过程

每个镜像一个Yml文件，Zookeeper、Kafka、Kafka-Manager一个

编写yml文件

（1）docker-compose-zookeeper.yml

Zookeeper各个节点的信息，端口映射，集群信息，网络配置

version: '2'  #指定 compose 文件的版本services:  #通过镜像安装容器的配置zoo1:image: zookeeper:3.4  #使用的镜像restart: always		  #当Docker重启时，该容器重启hostname: zoo1		  #类似于在基于Linux虚拟机Kafka集群中hosts文件的值container_name: zoo1ports:- 2184:2181   #端口映射environment:  #集群环境ZOO_MY_ID: 1 #当前Zookeeper实例的idZOO_SERVERS: server.1=zoo1:2888:3888 server.2=zoo2:2888:3888 server.3=zoo3:2888:3888 #集群节点networks:     #使用的网络配置kafka:ipv4_address: 192.168.0.11zoo2:image: zookeeper:3.4restart: alwayshostname: zoo2container_name: zoo2ports:- 2185:2181environment:ZOO_MY_ID: 2ZOO_SERVERS: server.1=zoo1:2888:3888 server.2=0.0.0.0:2888:3888 server.3=zoo3:2888:3888networks:kafka:ipv4_address: 192.168.0.12zoo3:image: zookeeper:3.4restart: alwayshostname: zoo3container_name: zoo3ports:- 2186:2181environment:ZOO_MY_ID: 3ZOO_SERVERS: server.1=zoo1:2888:3888 server.2=zoo2:2888:3888 server.3=0.0.0.0:2888:3888networks:kafka:ipv4_address: 192.168.0.13networks:kafka:external:name: kafka

（2）docker-compose-kafka.yml

version: '2'services:kafka1:image: wurstmeister/kafka #imagerestart: alwayshostname: kafka1container_name: kafka1privileged: trueports:- 9092:9092environment: #集群环境配置KAFKA_ADVERTISED_HOST_NAME: kafka1KAFKA_LISTENERS: PLAINTEXT://kafka1:9092KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092KAFKA_ADVERTISED_PORT: 9092KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181external_links:  # 配置Zookeeper集群的地址- zoo1- zoo2- zoo3networks:kafka:ipv4_address: 192.168.0.14kafka2:image: wurstmeister/kafkarestart: alwayshostname: kafka2container_name: kafka2privileged: trueports:- 9093:9093environment:KAFKA_ADVERTISED_HOST_NAME: kafka2KAFKA_LISTENERS: PLAINTEXT://kafka2:9093KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9093KAFKA_ADVERTISED_PORT: 9093KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181external_links:- zoo1- zoo2- zoo3networks:kafka:ipv4_address: 192.168.0.15kafka3:image: wurstmeister/kafkarestart: alwayshostname: kafka3container_name: kafka3privileged: trueports:- 9094:9094environment:KAFKA_ADVERTISED_HOST_NAME: kafka3KAFKA_LISTENERS: PLAINTEXT://kafka3:9094KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka3:9094KAFKA_ADVERTISED_PORT: 9094KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181external_links:- zoo1- zoo2- zoo3networks:kafka:ipv4_address: 192.168.0.16networks:kafka:external:name: kafka

（3）docker-compose-manager.yml

version: '2'services:kafka-manager:image: sheepkiller/kafka-manager:latestrestart: alwayscontainer_name: kafka-managerhostname: kafka-managerports:- 9000:9000environment:  #可以管理zoo集群和kafka集群ZK_HOSTS: zoo1:2181,zoo2:2181,zoo3:2181KAFKA_BROKERS: kafka1:9092,kafka2:9092,kafka3:9092APPLICATION_SECRET: letmeinKM_ARGS: -Djava.net.preferIPv4Stack=truenetworks:kafka:ipv4_address: 192.168.0.17networks:kafka:external:name: kafka

2）将yaml文件上传到Docker宿主机中

安装：yum install -y lrzsz

上传到指定目录

3）开始部署

使用命令：docker-compose up -d

参数说明： up表示启动， -d表示后台运行

docker-compose -f /home/docker-compose-zookeeper.yml up -d

参数说明： -f:表示加载指定位置的yaml文件

docker-compose -f /home/docker-compose-kafka.yml up -d

docker-compose -f /home/docker-compose-manager.yml up -d

4）测试

浏览器访问宿主机：http://192.168.44.20:9000/

测试：

5. kafka的基本操作

在docker环境中演示

(1) 创建topic

创建一个名字为test的主题，有一个分区，有三个副本。一个主题下可以有多个分区，每个分区可以用对应的副本。

Docker：

#登录到Kafka容器
docker exec -it 9218e985e160 /bin/bash#切换到bin目录
cd opt/kafka/bin/#执行创建
kafka-topics.sh --create --zookeeper zoo1:2181 --replication-factor 3 --partitions 1 --topic test

–create：新建命令

–zookeeper：Zookeeper节点，一个或多个

–replication-factor：指定副本，每个分区有三个副本。

–partitions：1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6qjSXuPX-1637500558911)(E:\MarkDown\拉勾笔记\kafka–kafka的基本操作–创建topic)]

(2) 查看主题命令

查看kafka当中存在的主题

kafka-topics.sh --list --zookeeper zoo1:2181,zoo2:2181,zoo3:2181

__consumer_offsets 这个topic是由kafka自动创建的，默认50个分区，存储消费位移信息（offset），老版本架构中是存储在Zookeeper中。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qLjWTkvE-1637500558913)(E:\MarkDown\拉勾笔记\kafka–kafka的基本操作–查看主题命令)]

(3) 生产者生产数据

模拟生产者来生产数据：

Kafka自带一个命令行客户端，它从文件或标准输入中获取输入，并将其作为message（消息）发送到Kafka集群。

默认情况下，每行将作为单独的message发送。

运行 producer，然后在控制台输入一些消息以发送到服务器。

kafka-console-producer.sh --broker-list kafka1:9092,kafka2:9093,kafka3:9094 --topic test
This is a message
This is another message

(4) 消费者消费数据

kafka-console-consumer.sh --bootstrap-server kafka1:9092, kafka2:9093, kafka3:9094 --topic test --from-beginning

在使用的时候会用到bootstrap与broker.list其实是实现一个功能，broker.list是旧版本命令。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-oksVWzBJ-1637500558914)(E:\MarkDown\拉勾笔记\kafka–kafka的基本操作-- 消费者消费数据)]

一、确保消费者消费的消息是顺序的，需要把消息存放在同一个topic的同一个分区

二、一个主题多个分区，分区内消息有序。

(5) 运行describe的命令

运行describe查看topic的相关详细信息

#查看topic主题详情，Zookeeper节点写一个和全部写，效果一致
kafka-topics.sh --describe --zookeeper zoo1:2181,zoo2:2181,zoo3:2181 --topic test#结果列表
Topic: test1 PartitionCount: 3 ReplicationFactor: 3 Configs:
Topic: test1 Partition: 0 Leader: 1001 Replicas: 1001,1003,1002 Isr: 1001,1003,1002
Topic: test1 Partition: 1 Leader: 1002 Replicas: 1002,1001,1003 Isr: 1002,1001,1003
Topic: test1 Partition: 2 Leader: 1003 Replicas: 1003,1002,1001 Isr: 1003,1002,1001

结果说明：

这是输出的解释。第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。有几个分区，下面就显示几行

leader：是负责给定分区的所有读取和写入的节点。每个节点将成为随机选择的分区部分的领导者。

replicas：显示给定partiton所有副本所存储节点的节点列表，不管该节点是否是leader或者是否存活。

isr：副本都已同步的的节点集合，这个集合中的所有节点都是存活状态，并且跟leader同步

(6) 增加topic分区数

任意kafka服务器执行以下命令可以增加topic分区数

kafka-topics.sh --zookeeper zkhost:port --alter --topic test --partitions 8

(7) 增加配置

flush.messages：此项配置指定时间间隔：强制进行fsync日志，默认值为None。

例如，如果这个选项设置为1，那么每条消息之后都需要进行fsync，如果设置为5，则每5条消息就需要进行一次fsync。

一般来说，建议你不要设置这个值。此参数的设置，需要在"数据可靠性"与"性能"之间做必要的权衡。

如果此值过大，将会导致每次"fsync"的时间较长(IO阻塞)。

如果此值过小，将会导致"fsync"的次数较多，这也意味着整体的client请求有一定的延迟，物理server故障，将会导致没有fsync的消息丢失。

动态修改kakfa的配置

kafka-topics.sh --zookeeper zoo1:2181 --alter --topic test --config flush.messages=1

(8) 删除配置

动态删除kafka集群配置

kafka-topics.sh --zookeeper zoo1:2181 --alter --topic test --delete-config flush.messages

(9) 删除topic

目前删除topic在默认情况只是打上一个删除的标记，在重新启动kafka后才删除。如果需要立即删除，则需要在 server.properties中配置：

delete.topic.enable=true（集群中的所有实例节点），一个主题会在不同的kafka节点中分配分组信息和副本信息

然后执行以下命令进行删除topic

kafka-topics.sh --zookeeper zoo1:2181 --delete --topic test

6. Java API操作kafka

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Kzabx0Kh-1637500558917)(E:\MarkDown\拉勾笔记\kafka–Java API操作kafka)]

修改Windows的Host文件：

目录：C:\Windows\System32\drivers\etc (win10)

内容：

192.168.44.20 kafka1
192.168.44.20 kafka2
192.168.44.20 kafka3

创建maven的工程，导入kafka相关的依赖

<dependencies><!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>1.0.0</version></dependency>
</dependencies>
<build><plugins><!-- java编译插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.2</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin></plugins>
</build>

6.1 生产者代码

/*** 消息生产者*/
public class PeoducerDemo {public static void main(String[] args) {// 要构造一个消息生产者对象，关于kafka集群等相关配置，可以从Properties文件中加载也可以凶一个Properties对象中// KafkaProducer按照固定的key取出对应的valueProperties properties = new Properties();// 指定集群节点properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.44.20:9092,192.168.44.20:9093,192.168.44.20:9094");// 发送消息，网络传输，需要对key和value指定对应的序列化类properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);// 创建消息生产者对象KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);// 主题String topic = "lagou";// 发送除100条消息for (int i=1;i<=100;i++){// 设置消息的内容String msg = "hello,"+i;// 构建一个消息对象：主题（如果不存在，kafka会帮我们创建一个一个分区一个副本的主题），消息ProducerRecord<String,String> record = new ProducerRecord<>(topic,msg);// 发送producer.send(record);System.out.println("消息发送成功,msg:"+msg);try {Thread.sleep(500);} catch (InterruptedException e) {e.printStackTrace();}}// 关闭消息生产者对象producer.close();}
}

6.2 消费者代码

/*** 消息消费者*/
public class ConsumerDemo {public static void main(String[] args) {// 属性对象Properties properties = new Properties();// 指定集群节点properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.44.20:9092,192.168.44.20:9093,192.168.44.20:9094");// 反序列化类properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);// 指定分组名称properties.put(ConsumerConfig.GROUP_ID_CONFIG,"lagou_grop1");// 消息消费者KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String, String>(properties);// 主题String topic = "lagou";// 订阅消息kafkaConsumer.subscribe(Collections.singletonList(topic));while (true){// 获取消息的方法时一个阻断式方法ConsumerRecords<String, String> records = kafkaConsumer.poll(500);for (ConsumerRecord<String,String> record : records) {System.out.println("主题："+record.topic()+",偏移量:"+record.offset()+",msg:"+record.value());}}}
}

7. Apache kafka原理

7.1 分区副本机制

kafka有三层结构：kafka有多个主题，每个主题有多个分区，每个分区又有多条消息。

分区机制：主要解决了单台服务器存储容量有限和单台服务器并发数限制的问题一个分片的不同副本不能放到同一个broker上。

当主题数据量非常大的时候，一个服务器存放不了，就将数据分成两个或者多个部分，存放在多台服务器上。每个服务器上的数据，叫做一个分片

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-saxtYc2f-1637500558919)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–分区副本机制)]

分区对于 Kafka 集群的好处是：实现负载均衡，高存储能力、高伸缩性。分区对于消费者来说，可以提高并发度，提高效率。

副本：副本备份机制解决了数据存储的高可用问题

当数据只保存一份的时候，有丢失的风险。为了更好的容错和容灾，将数据拷贝几份，保存到不同的机器上。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Abq1k0rF-1637500558922)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–分区副本机制02)]

多个follower副本通常存放在和leader副本不同的broker中。通过这样的机制实现了高可用，当某台机器挂掉后，其他follower副本也能迅速”转正“，开始对外提供服务。

kafka的副本都有哪些作用？

在kafka中，实现副本的目的就是冗余备份，且仅仅是冗余备份，所有的读写请求都是由leader副本进行处理的。follower副本仅有一个功能，那就是从leader副本拉取消息，尽量让自己跟leader副本的内容一致。

说说follower副本为什么不对外提供服务？

这个问题本质上是对性能和一致性的取舍。试想一下，如果follower副本也对外提供服务那会怎么样呢？首先，性能是肯定会有所提升的。但同时，会出现一系列问题。类似数据库事务中的幻读，脏读。

比如你现在写入一条数据到kafka主题a，消费者b从主题a消费数据，却发现消费不到，因为消费者b去读取的那个分区副本中，最新消息还没写入。而这个时候，另一个消费者c却可以消费到最新那条数据，因为它消费了leader副本。

为了提高那么些性能而导致出现数据不一致问题，那显然是不值得的。

7.2 kafka保证数据不丢失机制

从Kafka的大体角度上可以分为数据生产者，Kafka集群，还有就是消费者，而要保证数据的不丢失也要从这三个角度去考虑。

7.2.1. 消息生产者

消息生产者保证数据不丢失：消息确认机制（ACK机制），参考值有三个：0，1，-1

//producer无需等待来自broker的确认而继续发送下一批消息。
//这种情况下数据传输效率最高，但是数据可靠性确是最低的。
properties.put(ProducerConfig.ACKS_CONFIG,"0");//producer只要收到一个分区副本成功写入的通知就认为推送消息成功了。
//这里有一个地方需要注意，这个副本必须是leader副本。
//只有leader副本成功写入了，producer才会认为消息发送成功。
properties.put(ProducerConfig.ACKS_CONFIG,"1");//ack=-1，简单来说就是，producer只有收到分区内所有副本的成功写入的通知才认为推送消息成功了。
properties.put(ProducerConfig.ACKS_CONFIG,"-1");

7.2.2 消息消费者

kafka消费消息的模型：

即消费消息，设置好offset，类比一下：

Kafka动作	看书动作
消费消息	看书
offset位移	书签

什么时候消费者丢失数据呢？

由于Kafka consumer默认是自动提交位移的（先更新位移，再消费消息），如果消费程序出现故障，没消费完毕，则丢失了消息，此时，broker并不知道。

解决方案：

enable.auto.commit=false 关闭自动提交位移

在消息被完整处理之后再手动提交位移

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

7.3 消息存储及查询机制

kafka 使用日志文件的方式来保存生产者消息，每条消息都有一个 offset 值来表示它在分区中的偏移量

Kafka 中存储的一般都是海量的消息数据，为了避免日志文件过大，一个分片并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是<topic_name>_<partition_id>。

kafka容器数据目录：

/kafka/kafka-logs-kafka1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qGUWPKQV-1637500558924)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–消息存储及查询机制)]

7.3.1 消息存储机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zsRZkMTa-1637500558926)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–消息存储及查询机制–消息存储机制)]

log分段：

每个分片目录中，kafka 通过分段的方式将数据分为多个 LogSegment。

一个 LogSegment 对应磁盘上的一个日志文件（00000000000000000000.log）和一个索引文件(如上： 00000000000000000000.index)。

其中日志文件是用来记录消息的。索引文件是用来保存消息的索引。

每个LogSegment 的大小可以在server.properties 中log.segment.bytes=107370 （设置分段大小,默认是1gb）选项进行设置。

当log文件等于1G时，新的会写入到下一个segment中。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SFmi68U1-1637500558928)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–消息存储及查询机制–消息存储机制02)]

timeindex文件，是kafka的具体时间日志

7.3.2 通过 offset 查找 message

存储的结构：一个主题 --> 多个分区 ----> 多个日志段（多个文件）

第一步：查询segment file：

segment file命名规则跟offset有关，根据segment file可以知道它的起始偏移量，因为Segment file的命名规则是上一个segment文件最后一条消息的offset值。所以只要根据offset 二分查找文件列表，就可以快速定位到具体文件。

比如

第一个segment file是00000000000000000000.index表示最开始的文件，起始偏移量(offset)为0。

第二个是00000000000000091932.index：代表消息量起始偏移量为91933 = 91932 + 1。那么offset=5000时应该定位 00000000000000000000.index

第二步通过segment file查找message：

通过第一步定位到segment file，当offset=5000时，依次定位到00000000000000000000.index的元数据物理位置和 00000000000000000000.log的物理偏移地址，然后再通过00000000000000000000.log顺序查找直到offset=5000为止。

7.4 生产者消息分发策略

kafka在数据生产的时候，有一个数据分发策略。默认的情况使用DefaultPartitioner.class类

这个类中就定义数据分发的策略。

public interface Partitioner extends Configurable, Closeable {/*** Compute the partition for the given record.** @param topic The topic name* @param key The key to partition on (or null if no key)* @param keyBytes The serialized key to partition on( or null if no key)* @param value The value to partition on or null* @param valueBytes The serialized value to partition on or null* @param cluster The current cluster metadata*/public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);/*** This is called when partitioner is closed.*/public void close();
}

默认实现类：org.apache.kafka.clients.producer.internals.DefaultPartitioner

1）如果是用户指定了partition，生产就不会调用DefaultPartitioner.partition()方法

数据分发策略的时候，可以指定数据发往哪个partition。

当ProducerRecord 的构造参数中有partition的时候，就可以发送到对应partition上

/*** Creates a record to be sent to a specified topic and partition** @param topic The topic the record will be appended to* @param partition The partition to which the record should be sent* @param key The key that will be included in the record* @param value The record contents*/
public ProducerRecord(String topic, Integer partition, K key, V value) {this(topic, partition, null, key, value, null);
}

2） DefaultPartitioner源码

如果指定key，是取决于key的hash值

如果不指定key，轮询分发

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {//获取该topic的分区列表List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);//获得分区的个数int numPartitions = partitions.size();//如果key值为nullif (keyBytes == null) {//如果没有指定key，那么就是轮询//维护一个key为topic的ConcurrentHashMap，并通过CAS操作的方式对value值执行递增+1操作int nextValue = nextValue(topic);//获取该topic的可用分区列表List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);if (availablePartitions.size() > 0) {//如果可用分区大于0//执行求余操作，保证消息落在可用分区上int part = Utils.toPositive(nextValue) % availablePartitions.size();return availablePartitions.get(part).partition();} else {// 没有可用分区的话，就给出一个不可用分区return Utils.toPositive(nextValue) % numPartitions;}} else {//不过指定了key，key肯定就不为null// 通过计算key的hash，确定消息分区return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;}
}

7.5 消费者负载均衡机制

同一个分区中的数据，只能被一个消费者组中的一个消费者所消费。例如 P0分区中的数据不能被Consumer Group A中C1与C2同时消费。

消费组：一个消费组中可以包含多个消费者，properties.put(ConsumerConfig.GROUP_ID_CONFIG,“groupName”)；如果该消费组有四个消费者，主题有四个分区，那么每人一个。多个消费组可以重复消费消息。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZkXciyuQ-1637500558931)(E:\MarkDown\拉勾笔记\kafka–Apache kafka原理–消费者负载均衡机制)]

如果有3个Partition, p0/p1/p2，同一个消费组有3个消费者，c0/c1/c2，则为一一对应关系；

如果有3个Partition, p0/p1/p2，同一个消费组有2个消费者，c0/c1，则其中一个消费者消费2个分区的数据，另一个消费者消费一个分区的数据；

如果有2个Partition, p0/p1，同一个消费组有3个消费者，c0/c1/c3，则其中有一个消费者空闲，另外2个消费者消费分别各自消费一个分区的数据；

8. kakfa配置文件说明

server.properties

1、broker.id=0：

kafka集群是由多个节点组成的，每个节点称为一个broker，中文翻译是代理。每个broker都有一个不同的brokerId，由broker.id指定，是一个不小于0的整数，各brokerId必须不同，但不必连续。如果我们想扩展kafka集群，只需引入新节点，分配一个不同的broker.id即可。

启动kafka集群时，每一个broker都会实例化并启动一个kafkaController，并将该broker的brokerId注册到zooKeeper的相应节点中。集群各broker会根据选举机制选出其中一个broker作为leader，即leader kafkaController。leader kafkaController负责主题的创建与删除、分区和副本的管理等。当leader kafkaController宕机后，其他broker会再次选举出新的leader kafkaController。

2、log.dir = /export/data/kafka/

broker持久化消息到哪里，数据目录

3、log.retention.hours = 168

log文件最小存活时间，默认是168h，即7天。相同作用的还有log.retention.minutes、log.retention.ms。retention是保存的意思。

数据存储的最大时间超过这个时间会根据log.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据。

log.retention.bytes和log.retention.hours任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖。

4、log.retention.check.interval.ms

多长时间检查一次是否有log文件要删除。默认是300000ms，即5分钟。

5、log.retention.bytes

限制单个分区的log文件的最大值，超过这个值，将删除旧的log，以满足log文件不超过这个值。默认是-1，即不限制。

6、log.roll.hours

多少时间会生成一个新的log segment，默认是168h，即7天。相同作用的还有log.roll.ms、segment.ms。

7、log.segment.bytes

log segment多大之后会生成一个新的log segment，默认是1073741824，即1G。

8、log.flush.interval.messages

指定broker每收到几个消息就把消息从内存刷到硬盘（刷盘）。默认是9223372036854775807好大。

kafka官方不建议使用这个配置，建议使用副本机制和操作系统的后台刷新功能，因为这更高效。这个配置可以根据不同的topic设置不同的值，即在创建topic的时候设置值。

补充说明：

在Linux操作系统中，当我们把数据写入到文件系统之后，数据其实在操作系统的page cache里面，并没有刷到磁盘上去。如果此时操作系统挂了，其实数据就丢了。1、kafka是多副本的，当你配置了同步复制之后。多个副本的数据都在page cache里面，出现多个副本同时挂掉的概率比1个副本挂掉，概率就小很多了2、操作系统有后台线程，定期刷盘。如果应用程序每写入1次数据，都调用一次fsync，那性能损耗就很大，所以一般都会在性能和可靠性之间进行权衡。因为对应一个应用来说，虽然应用挂了，只要操作系统不挂，数据就不会丢。

9、log.flush.interval.ms

指定broker每隔多少毫秒就把消息从内存刷到硬盘。默认值同log.flush.interval.messages一样， 9223372036854775807。

同log.flush.interval.messages一样，kafka官方不建议使用这个配置。

10、delete.topic.enable=true

是否允许从物理上删除topic

9. kafka监控与运维

9.1 kafka-eagle概述

在生产环境下，在Kafka集群中，消息数据变化是我们关注的问题，当业务前提不复杂时，我们可以使用Kafka 命令提供带有Zookeeper 客户端工具的工具，可以轻松完成我们的工作。随着业务的复杂性，增加Group和 Topic，那么我们使用Kafka提供命令工具，已经感到无能为力，那么Kafka监控系统目前尤为重要，我们需要观察消费者应用的细节。

为了简化开发者和服务工程师维护Kafka集群的工作有一个监控管理工具，叫做 Kafka-eagle。这个管理工具可以很容易地发现分布在集群中的哪些topic分布不均匀，或者是分区在整个集群分布不均匀的的情况。它支持管理多个集群、选择副本、副本重新分配以及创建 Topic。同时，这个管理工具也是一个非常好的可以快速浏览这个集群的工具

9.2 搭建安装 kafka-eagle

环境要求:需要安装jdk，启动zk以及kafka的服务

# 启动Zookeeper
zkServer.sh start#启动Kafka
nohup ./kafka-server-start.sh /export/servers/kafka/config/server.properties 2>&1 &

修改windows host文件

192.168.44.20 kafka1
192.168.44.20 kafka2
192.168.44.20 kafka3
192.168.44.134 node1
192.168.44.135 node2
192.168.44.136 node3

搭建步骤：

下载kafka-eagle的源码包

kafka-eagle官网：http://download.kafka-eagle.org/

我们可以从官网上面直接下载最新的安装包即可kafka-eagle-bin-1.3.2.tar.gz这个版本即可

代码托管地址：https://github.com/smartloli/kafka-eagle/releases

上传安装包并解压：

这里我们选择将kafak-eagle安装在第三台

如果要解压的是zip格式，需要先安装命令支持。

yum install unzip

unzip xxxx.zip

#将安装包上传至 node01服务器的/export/softwares路径下, 然后解压
cd /export/softwares/
unzip kafka-eagle.zip
cd cd kafka-eagle-web/target/
tar -zxf kafka-eagle-web-2.0.1-bin.tar.gz -C /export/servers

准备数据库：

kafka-eagle需要使用一个数据库来保存一些元数据信息，我们这里直接使用msyql数据库来保存即可，在node01服务器执行以下命令创建一个mysql数据库即可

SQLite、MySQL

--进入mysql客户端:
create database eagle;

修改kafka-eagle配置文件

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZgCL96Z3-1637500558933)(E:\MarkDown\拉勾笔记\kafka–kafka监控与运维–搭建安装 kafka-eagle)]

cd /export/servers/kafka-eagle-bin-1.3.2/kafka-eagle-web-1.3.2/conf
vim system-config.properties#内容如下:
kafka.eagle.zk.cluster.alias=cluster1
cluster1.zk.list=node1:2181,node2:2181,node3:2181kafka.eagle.driver=com.mysql.jdbc.Driver
kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/eagle
kafka.eagle.username=root
kafka.eagle.password=123456

默认情况下MySQL只允许本机连接到MYSQL实例中，所以如果要远程访问，必须开放权限：

update user set host = ‘%’ where user =‘root’; //修改权限

flush privileges; //刷新配置

配置环境变量

kafka-eagle必须配置环境变量，node03服务器执行以下命令来进行配置环境变量

vi /etc/profile
#内容如下:
export KE_HOME=/export/servers/kafka-eagle-bin-1.3.2/
export PATH=:$KE_HOME/bin:$PATH#让修改立即生效，执行
source /etc/profile

启动kakfa-eagle

cd kafka-eagle-web-1.3.2/bin
chmod u+x ke.sh
./ke.sh start

访问主界面：

http://node1:8048/ke/account/signin?/ke/

用户名：admin

密码：123456