ElasticSearch分布式架构原理

1.相关概念

• Near Realtime（NRT）：近实时，两个意思，从写入数据到数据可以被搜索到有一个小延迟（大概1秒）；基于es执行搜索和分析可以达到秒级

• Cluster：集群，包含多个节点，每个节点属于哪个集群是通过一个配置（集群名称，默认是elasticsearch）来决定的，对于中小型应用来说，刚开始一个集群就一个节点很正常

• Node：节点(简单理解为集群中的一个服务器)，集群中的一个节点，节点也有一个名称（默认是随机分配的），节点名称很重要（在执行运维管理操作的时候），默认节点会去加入一个名称为“elasticsearch”的集群，如果直接启动一堆节点，那么它们会自动组成一个elasticsearch集群，当然一个节点也可以组成一个elasticsearch集群

• Index：索引(简单理解就是一个数据库)，包含一堆有相似结构的文档数据，比如可以有一个客户索引，商品分类索引，订单索引，索引有一个名称。一个index包含很多document，一个index就代表了一类类似的或者相同的document。比如说建立一个product index，商品索引，里面可能就存放了所有的商品数据，所有的商品document。

• Type：类型(简单理解就是一张表)，每个索引里都可以有一个或多个type，type是index中的一个逻辑数据分类，一个type下的document，都有相同的field，比如博客系统，有一个索引，可以定义用户数据type，博客数据type，评论数据type。

• Document&field：文档(就是一行数据)，es中的最小数据单元，一个document可以是一条客户数据，一条商品分类数据，一条订单数据，通常用JSON数据结构表示，每个index下的type中，都可以去存储多个document。一个document里面有多个field，每个field就是一个数据字段。

• shard：分片是ES中最小的工作单元，是一个Lucence的Index，单台机器无法存储大量数据，es可以将一个索引中的数据切分为多个shard，分布在多台服务器上存储。有了shard就可以横向扩展，存储更多数据，让搜索和分析等操作分布到多台服务器上去执行，提升吞吐量和性能。每个shard都是一个lucene index。

• replica：任何一个服务器随时可能故障或宕机，此时shard可能就会丢失，因此可以为每个shard创建多个replica副本。replica可以在shard故障时提供备用服务，保证数据不丢失，多个replica还可以提升搜索操作的吞吐量和性能。primary shard（建立索引时一次设置，不能修改，默认5个），replica shard（随时修改数量，默认1个），默认每个索引10个shard，5个primary shard，5个replica shard，最小的高可用配置，是2台服务器。

2、ElasticSearch分布式架构原理

2.1、shad与replica机制

• 一个index包含多个shard,也就是一个index存在多个服务器上

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

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

• primary shard和replica shard，每个document肯定只存在于某一个primary shard以及其对应的replica shard中，不可能存在于多个primary shard

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

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

• primary shard的默认数量是5，replica默认是1，默认有10个shard，5个primary shard，5个replica shard

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

2.2、分布式架构图

2.3、容错机制

在集群中会有一个master负责当leader进行协调,比如上图的Node2为master, 那么当它挂了的时候会重现选举一个新的master,比如新选举的是Node3,这个时候replica 2这时候会变成primary.

当Node2恢复了的时候,这个时候node2的prinary会变成replica

2.4、ES写入数据的过程

2.4.1简单流程:

1、客户端选择一个node发送请求过去，这个node就是coordinating node (协调节点)

2、coordinating node，对document进行路由，将请求转发给对应的node

3、实际上的node上的primary shard处理请求，然后将数据同步到replica node

4、coordinating node，如果发现primary node和所有的replica node都搞定之后，就会返回请求到客户端

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

2.4.2写入数据底层原理:

1、数据先写入到buffer里面，在buffer里面的数据时搜索不到的，同时将数据写入到translog日志文件之中

2、如果buffer快满了，或是一段时间之后(定时)，就会将buffer数据refresh到一个新的OS cache之中，然后每隔1秒，就会将OS cache的数据写入到segment file之中，但是如果每一秒钟没有新的数据到buffer之中，就会创建一个新的空的segment file，只要buffer中的数据被refresh到OS cache之中，就代表这个数据可以被搜索到了。当然可以通过restful api 和Java api，手动的执行一次refresh操作，就是手动的将buffer中的数据刷入到OS cache之中，让数据立马搜索到，只要数据被输入到OS cache之中，buffer的内容就会被清空了。同时进行的是，数据到shard之后，就会将数据写入到translog之中，每隔5秒将translog之中的数据持久化到磁盘之中

3、重复以上的操作，每次一条数据写入buffer，同时会写入一条日志到translog日志文件之中去，这个translog文件会不断的变大，当达到一定的程度之后，就会触发commit操作。

4、 将一个commit point写入到磁盘文件，里面标识着这个commit point 对应的所有segment file

5、 强行将OS cache 之中的数据都fsync到磁盘文件中去。 解释：translog的作用：在执行commit之前，所有的而数据都是停留在buffer或OS cache之中，无论buffer或OS cache都是内存，一旦这台机器死了，内存的数据就会丢失，所以需要将数据对应的操作写入一个专门的日志问价之中，一旦机器出现宕机，再次重启的时候，es会主动的读取translog之中的日志文件的数据，恢复到内存buffer和OS cache之中。

6、 将现有的translog文件进行清空，然后在重新启动一个translog，此时commit就算是成功了，默认的是每隔30分钟进行一次commit，但是如果translog的文件过大，也会触发commit，整个commit过程就叫做一个flush操作，我们也可以通过ES API,手动执行flush操作，手动将OS cache 的数据fsync到磁盘上面去，记录一个commit point，清空translog文件 补充：其实translog的数据也是先写入到OS cache之中的，默认每隔5秒之中将数据刷新到硬盘中去，也就是说，可能有5秒的数据仅仅停留在buffer或者translog文件的OS cache中，如果此时机器挂了，会丢失5秒的数据，但是这样的性能比较好，我们也可以将每次的操作都必须是直接fsync到磁盘，但是性能会比较差。

7、如果时删除操作，commit的时候会产生一个.del文件，里面讲某个doc标记为delete状态，那么搜索的时候，会根据.del文件的状态，就知道那个文件被删除了。

8、如果时更新操作，就是讲原来的doc标识为delete状态，然后重新写入一条数据即可。

9、buffer每次更新一次，就会产生一个segment file 文件，所以在默认情况之下，就会产生很多的segment file 文件，将会定期执行merge操作

10、每次merge的时候，就会将多个segment file 文件进行合并为一个，同时将标记为delete的文件进行删除，然后将新的segment file 文件写入到磁盘，这里会写一个commit point，标识所有的新的segment file，然后打开新的segment file供搜索使用。

2.5、ES查询过程

2.5.1、倒排序算法

• 倒排索引采用的是Immutable Design，一旦生成不可更改。
• 不可变性，带来的好处如下：

1. 不需要考虑并发写文件的问题，避免了锁机制带来的性能问题
2. 一旦读入内核的文件系统缓存，便留在那里，只要文件系统有足够的空间，大部分请求就会直接请求内存，不会命中磁盘，极大的提高了性能
3. 缓存容易生成和为何，并且数据可以被压缩

• 不可变性，也带来了挑战： 如果需要让一个新的文档可以被搜索，需要重建整个索引

2.5.2、查询过程

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

2.5.3、查询原理

查询过程大体上分为查询和取回这两个阶段，广播查询请求到所有相关分片，并将它们的响应整合成全局排序后的结果集合，这个结果集合会返回给客户端。

• 查询阶段

1、当一个节点接收到一个搜索请求，这这个节点就会变成协调节点，第一步就是将广播请求到搜索的每一个节点的分片拷贝，查询请求可以被某一个主分片或某一个副分片处理，协调节点将在之后的请求中轮训所有的分片拷贝来分摊负载。

2、每一个分片将会在本地构建一个优先级队列，如果客户端要求返回结果排序中从from 名开始的数量为size的结果集，每一个节点都会产生一个from+size大小的结果集，因此优先级队列的大小也就是from+size，分片仅仅是返回一个轻量级的结果给协调节点，包括结果级中的每一个文档的ID和进行排序所需要的信息。

3、协调节点将会将所有的结果进行汇总，并进行全局排序，最总得到排序结果。

• 取值阶段

1、查询过程得到的排序结果，标记处哪些文档是符合要求的，此时仍然需要获取这些文档返回给客户端

2、协调节点会确定实际需要的返回的文档，并向含有该文档的分片发送get请求，分片获取的文档返回给协调节点，协调节点将结果返回给客户端

2.6、更新过程

2.6.1、document的全量替换

1、这个就是用新的数据全部覆盖以前的数据

2、重新创建一个document并把原来的标记为delete

3、partial update, 就是制定需要更新的字段.

全量是把数据找出来,然后再java代码中进行修改,再放回去. partial是直接提交需要修改的字段然后直接修改,在一个shard中进行,内部也是全量替换.

2.6.2、强制创建

就是不管原来的数据,直接强制创建一个新的

2.7、删除过程

当要进行删除document的时候,只是把它标记为delete,当数据到达一定的时候再进行删除, 有点像JVM中标记清除法

2.8、Lucene Index

1. 一个ES中的shard就是Lucene中的Index(An ES Shard=A Lucene Index)
2. 在Lucene中，单个倒排索引文件被称为Segment；Segment是自包含的，不可变的，多个Segment汇总在一起，称为Lucene的Index;
3. 当有新文档写入时，会生成新segment；查询是会同时查询所有的segments,并且对结果汇总。
4. Lucene中有一个文件，用来记录所有segment信息，叫做commit point.
5. 删除的文档信息，保存在.del文件中。

2.9、Refresh

1. 将Index Buffer写入Segment的过程叫做Refresh,当执行Refresh的时候不执行fsync操作
2. Refresh频率：默认1s发生一次，可通过参数index.refresh_interval配置，refresh后数据就可以被搜索到了，这也是ES被称为近实时搜索（NRT）
3. 如果有大量的数据写入，就会产生很多的segment；
4. Index Buffer被占满是，就会触发refresh，index buffer默认大小是JVM的10%；

2.10、Transaction Log

1. segment写入磁盘的过程相对耗时，借助文件系统缓存，refresh的时候，先将segment写入文件系统缓存，以开放查询。
2. 为了保证数据不会丢失，在写入文档时，同时写入Transcation Log，高版本开始，默认transaction log默认落盘；
3. 每个分片都有一个transaction log；
4. 在ES refresh的时候，index buffer被清空，transaction Log不会清空。
5. 在发生断电的时候，因为有transaction Log的落盘，数据不会丢失。

2.11、ES Flash和Lucene Commit

1. 调用Refresh,Index Buffer清空并且Refresh
2. 调用fsync,将缓存中的segments写入磁盘
3. 清空（删除）transaction log
4. 默认30分钟调用一次
5. transaction log 满（默认512M）

2.12、Merge

1. Segment很多，需要被定期合并。可以减少segments和删除已经删除的文档
2. 1ES和Lucene会自动进行merge操作。如果手动执行，可以通过使用以下API进行操作：POST my_index/_forcemerge

3、Es并发解决方案

为什么会出现并发问题,简单的说就是多个线程去操作同一个数据.

假如现在下单系统下2单,都要去减库存(100件),第一个线程进去减1件100-1=99,这时候还没更像到ES中,第二线程进去了,也要减一个库存100-1=99.现在系统卖出去2个,可是库存却还有99个,应该是98个

3.1、解决方案-悲观锁

3.2、解决方案-乐观锁

温馨提示,乐观锁会出现ABA情况

下面是2种解决方案,在网上找的图片

注意：本文归作者所有，未经作者允许，不得转载