Elasticsearch系列---分布式架构机制讲解

时间 2019-11-24

原文原文链接

概要

本篇主要介绍Elasticsearch的数据索引时的分片机制，集群发现机制，primary shard与replica shard是如何分工合做的，如何对集群扩容，以及集群的容错机制。java

分片机制

前面基本概念一节中，咱们有提到创建索引时，会自动将数据拆分到多个分片（shard）中，默认数量是5，这个就是索引数据分片机制。咱们在往Elasticsearch集群插入数据，并无关心过数据最终落地到哪一个shard上，这个过程对客户端来说是透明的。node

document路由原理

document要存储到Elasticsearch中，还要知足后续搜索的需求，路由到分片位置的算法确定不能是随机的，要否则搜索就无法找了，路由的过程有一个公式：算法

shard = hash(routing) % number_of_primary_shards服务器

routing值默认是document的ID值，也能够自行指定。先对routing信息求hash值，而后将hash结果对primary_shard的数量求模，好比说primary_shard是5，那么结果确定落在[0,4]区间内，这个结果值就是该document的分片位置，如示意图所示：网络

这个求模公式间接的解释了为何了索引建立时指定了primary shard的值，后续就不让改了，模数改了，以前路由的document再执行该公式时，值就可能跟改以前获得的值不一致，这样document就找不到了，如示意图所示：架构

集群发现机制

在同一个网络环境下，只要启动一个Elasticsearch实例，而且cluster.name配置得同样，这个Elasticsearch实例就会自动加入到集群当中，这个是如何实现的？并发

这个依赖于Elasticsearch的自动发现机制Zen，在elasticsearch.yml配置文件中，有一行
discovery.zen.ping.unicast.hosts: ["192.168.17.137"]
表示单播发现方式，当该Elasticsearch实例启动时，会向192.168.17.137主机发送请求，并获得整个集群里全部节点的状态，而后去联系master节点，并加入集群。负载均衡

摘抄了获取配置信息，注册discovery请求的部分源码以下：
org.elasticsearch.discovery.zen.ZenDiscovery启动时的构造器，会调用org.elasticsearch.discovery.zen.UnicastZenPing的构造器，其中UnicastZenPing的构造方式内会加载discovery.zen.ping.unicast.hosts配置项，并发送"internal:discovery/zen/unicast"请求（代码有删节）：elasticsearch

public UnicastZenPing(Settings settings, ThreadPool threadPool, TransportService transportService,
                          UnicastHostsProvider unicastHostsProvider, PingContextProvider contextProvider) {
        super(settings);

        final int concurrentConnects = DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING.get(settings);
        if (DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.exists(settings)) {
            configuredHosts = DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.get(settings);
            // we only limit to 1 addresses, makes no sense to ping 100 ports
            limitPortCounts = LIMIT_FOREIGN_PORTS_COUNT;
        } else {
            // if unicast hosts are not specified, fill with simple defaults on the local machine
            configuredHosts = transportService.getLocalAddresses();
            limitPortCounts = LIMIT_LOCAL_PORTS_COUNT;
        }
        resolveTimeout = DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT.get(settings);


        transportService.registerRequestHandler(ACTION_NAME, ThreadPool.Names.SAME, UnicastPingRequest::new,
            new UnicastPingRequestHandler());
    }

shard&replica规则

一个index的数据，是拆分存储在多个shard当中，咱们能够在Elasticsearch的数据目录里查看一下索引的存储结构（Elasticsearch服务器上导出的树状目录结构）：分布式

.
└── nodes
    └── 0
        ├── indices
        │   ├── 48G_CgE7TiWomlYsyQW1NQ #索引location的UUID
        │   │   ├── 0 #primary shard，从0-4共5个
        │   │   │   ├── index
        │   │   │   │   ├── segments_3
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-2.ckp
        │   │   │       ├── translog-2.tlog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 1
        │   │   │   ├── index
        │   │   │   │   ├── segments_3
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-2.ckp
        │   │   │       ├── translog-2.tlog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 2
        │   │   │   ├── index
        │   │   │   │   ├── _1.cfe
        │   │   │   │   ├── _1.cfs
        │   │   │   │   ├── _1.si
        │   │   │   │   ├── segments_7
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.ckp
        │   │   │       ├── translog-5.tlog
        │   │   │       ├── translog-6.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 3
        │   │   │   ├── index
        │   │   │   │   ├── _1.cfe
        │   │   │   │   ├── _1.cfs
        │   │   │   │   ├── _1.si
        │   │   │   │   ├── segments_7
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.ckp
        │   │   │       ├── translog-5.tlog
        │   │   │       ├── translog-6.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 4
        │   │   │   ├── index
        │   │   │   │   ├── _0.cfe
        │   │   │   │   ├── _0.cfs
        │   │   │   │   ├── _0.si
        │   │   │   │   ├── segments_5
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.tlog
        │   │   │       └── translog.ckp
        │   │   └── _state
        │   │       └── state-16.st
        ├── node.lock
        └── _state
            ├── global-88.st
            └── node-22.st

如上目录结构所示，展现了location索引（UUID为48G_CgE7TiWomlYsyQW1NQ）的存储信息，共5个primary shard，编号从0-4。

primary shard与replica shard，还有其余几点特性：

shard是最小的存储单元，像上面的0，1，2目录，承载部分数据。
document是最小的数据单元，只能存在一个primary shard中以及对应的replica shard中（可能有多个），不会拆分存储，也不会存在于多个primary shard里。
replica shard是primary shard的数据副本，冗余存储，负责容错，也能够承担查询请求。
primary shard不会和本身的replica shard放在一台机器上，不然容错机制就失效了，可是能够和别的replica shard混搭。
primary shard的数量在建立索引的时是多少就多少，后续不能改，但replica shard的数量能够随时修改。

扩容机制

扩容分为垂直扩容和水平扩容两种，垂直扩容指增长单台服务器的CPU、内存大小，磁盘容量，简单来说就是换更强大的服务器；水平扩容就是增长机器数量，经过集群化部署与分布式的技术手段，也能构建出强大的计算和存储能力。

两者简单对比：

垂直扩容：操做简单，无须要更改集群方案，缺点就是贵，成本呈指数上升，而且单台服务器瓶颈很明显。
水平扩容：业务常常采用，由于更省钱，能够多很是多的普通服务器搭建，缺点是节点数越多，集群内节点之间通讯会出现网络拥塞的问题。

Elastisearch很是适合用水平扩容方案，能胜任上百个节点，支撑PB级别的数据规模，而且扩容操做后，每增长新的节点会触发索引分片的从新分配。

举个例子，假定Elasticsearch有2个节点，primary shard设置为3，replica shard设置为1，这样1个索引就有3个primary shard，3个replica shard，P表示primary shard，R表示replica shard，分布示例图以下：

当新加入一个node-3时，触发node-1和node-2的shard进行从新分配，假定P0和R1两个shard移到node-3当中，如图所示：

重分配完成后，此时集群的示例以下：

最后补充两点：

同一个index的primay shard和replica shard不能在同一个机器上，但不一样index的primary shard和replica shard能够混搭。
负载均衡也不是彻底平均的，有的多有的少，Elasticsearch会根据当前状况自动分配shard。

容错机制

单node环境下的容错

假定Elasticsearch集群只有一个node，primary shard设置为3，replica shard设置为1，这样1个索引就应该有3个primary shard，3个replica shard，但primary shard不能与其replica shard放在一个node里，致使replica shard没法分配，这样集群的status为yellow，示例图以下：

集群能够正常工做，一旦出现node宕机，数据所有丢失，而且集群不可用。

结论：单node环境容错性为0.

2台node环境下的容错

primary shard与replica shard的设置与上文相同，此时Elasticsearch集群只有2个node，shard分布以下图所示：

若是其中一台宕机，如node-2宕机，如图所示：

此时node-1节点的R2(replica shard)会升为P2(primary shard)，此时集群还能正经常使用，数据未丢失。

结论：双node环境容错性为1。

3台node环境下的容错

咱们先按primary shard为3，replica shard为1进行容错性计算。
此时每台node存放2个shard，若是一台宕机，此时另外2台确定还有完整的数据，若是两台宕机，剩下的那台就只有2/3的数据，数据丢失1/3，容错性为1台。
若是是这样设置，那3台的容错性和2台的容错性同样，就存在资源浪费的状况。

那怎么样提高容错性呢？
把replica shard的值改为2，这样每台node存放3个shard，以下图所示：

若是有2台宕机，就剩下node-2，此时集群的数据仍是完整的，replica会升成primary shard继续提供服务，以下图所示：

结论：3台node环境容错性最大能够是2。

扩容极限与最佳实践

根据上面3个场景，咱们能够知道，若是shard总数是6个（包含primary shard 和replica shard），那么node数量上限也为6，即每台node存储1个shard，这个数据即为扩容极限，若是要突破极限，能够经过增大replica的值来实现，这样有更多的replica shard去分担查询请求，占用更多的节点，整个集群的CPU、IO、Memory资源更多，总体吞吐量也越高。

固然这个replica也不是越大越好，冗余存储占用磁盘资源，replica越大，集群内有效数据的磁盘利用率就越低。以3台node为例，想要达到容错性，磁盘利用率的最佳值，replica=2是最适宜的。

实际生产中，能够根据数据量，并发数等实际需求，在建立索引时合理设置primary shard的数量，后期优化时，再调整replica shard的值，这个须要反复验证，不断的演算调整，最终让生产Elasticsearch集群的吞吐量达到一个最佳值。

容错过程与选举机制

Elasticsearch集群中，全部的node都是对等的角色，全部的node都能接收请求，而且能自动转请求到相应的节点上（数据路由），最后能将其余节点处理的数据进行响应收集，返回给客户端。在集群中，也存在一个master节点，它的职责多一些，须要管理与维护集群的元数据，索引的建立与删除和节点的增长和删除，它都会收到相应的请求，而后进行相应的数据维护。master node在承担索引、搜索请求时，与其余node一块儿分摊，并不承担全部的请求，于是不存在单点故障这个问题。

咱们假设一下集群有3台node，其中node-1宕机的过程，若是node-1是master node，关键步骤以下：

丢失了3个shard，因为P1丢失，cluster.status瞬间状态变成red。
从新进行master选举，自动选另外一个node做为master。
新的master将丢失了P1对应的R1（在node-3上面）提高为primary shard ，现所有primary shard active，可是P1，P2的replica shard没法启动，cluster.status变成yellow。
重启故障的node-1节点，新的master会将缺失的副本都copy一份到node-1上，node-1会使用之间已有的数据，而且同步一下宕机期间的数据修改，此时全部的shard所有active状态，cluster.status从新变成green。

小结

本篇针对Elasticsearch的一些内部原理进行了简单的介绍，这些原理针对Elasticsearch的使用者是透明的，为了增长可阅读性，自行增长一些讲解的原理图，如有不详尽之处或错误之处请指正，谢谢。

专一Java高并发、分布式架构，更多技术干货分享与心得，请关注公众号：Java架构社区