Elasticsearch 需要了解的都在这

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

ES选主过程

其实ES的选主过程其实没有很高深的算法加持启动过程中对接点的ID进行排序取ID最大节点作为Master节点那么如果选出来的主节点中存储的元信息不是最新的怎么办

其实他是分了2个步骤做这件事先选出ID最大的主节点然后再从元信息最新的节点将元信息数据复制到选举出来的主节点

当然虽然是选出ID取值最大的节点作为主节点但是内部还是有一定限制逻辑的

参选人数需要过半达到 quorum多数后就选出了临时的主。为什么是临时的每个节点运行排序取最大值的算法结果不一定相同。举个例子集群有5台主机节点ID分别是1、2、3、4、5。当产生网络分区或节点启动速度差异较大时节点1看到的节点列表是1、2、3、4选出4节点2看到的节点列表是2、3、4、5选出5。结果就不一致了由此产生下面的第二条限制。

得票数需过半。某节点被选为主节点必须判断加入它的节点数过半才确认Master身份。解决第一个问题。

当探测到节点离开事件时必须判断当前节点数是否过半。如果达不到quorum则放弃Master身份重新加入集群。如果不这么做则设想以下情况假设5台机器组成的集群产生网络分区2台一组3台一组产生分区前Master位于2台中的一个此时3台一组的节点会重新并成功选取Master产生双主俗称脑裂。

选举主分片过程

Master节点并不知道主分片节点位于哪里它只能向所有节点上的所有Shard去询问你的website下标是不是0那么这个询问的请求量就等于节点数 * Shard数如果你的规模非常庞大那么这个将会很耗时这个时候可能会获取到多份下标为0的请求响应5.X版本之后主分片的选举是根据一个List<最新主分片>在这个集合中的分片都是元信息最新的分片数据

高并发写数会不会影响读

首先写是加锁的无论是Bulk方式还是普通方式

Bulk方式的写锁

但是这个时候的写锁并不影响读只是为了确保这一次请求被一个线程处理这个时候用的是java的可重入锁

Put方式的写锁

这里使用的是可重入读写锁然后再初始化到Es自己的ReleasableLock中执行

Get方式读取数据读锁

由于读和写使用的是ReentrantReadWriteLock进行加锁那么读和写之间就是互斥的所以主分片在写数据到Cache的时候高并发的写是会影响读的性能的但是此时副分片节点的读不受影响之后副分片节点也开始处理主分片同步过来的数据时那么副分片的读也会收到影响

数据写入顺序性怎么保证

每一个写入的文档都有一个_version每次修改会递增用来判断本次修改命令是执行还是丢弃

10亿数据导入到ES需要7个小时

我们先看下写入过程

副本节点越多写入时间越久因为写入是串行的主节点写入完成后会并发请求所有副本节点会等到每一个副本节点响应后才会返回给协调者无论副本节点是否成功都会等待每一个节点响应

如果没有任何优化的话差不多需要这个时间如果你的场景是每天都会初始化一次那么多数据到ES的话我们一定需要优化的写入优化主要是几个方面

异步刷盘

可以利用异步刷盘加速写入速度只要丢失数据的风险是可接受范围就行了

index.translog.durability:async

调整副本数

ES副本数过多的时候就会导致写入缓慢看上面的流程应该比较清楚了每次大数据量导入之前将副本节点数调整成0是加速写入最直接的办法

BULK批量写入

通过API一条条写入的话太慢了ES是原生支持BULK批量方式操作

调整FLUSH

适当调整max_index_buffer_size大小不要太频繁的flush如果你的使用场景不需要数据太实时能被读到那么适当放大这个值

调整FLUSH时间

调整REFUSH

ES在修改数据的时候并不是修改原数据而是新增了一条数据然后将原数据打上了delete标记这样做的好处就是不需要对同一条数据进行加锁但是坏处就是数据碎片过多需要不定时merge而merge过程中需要对被merge的数据块获取合并锁但是不影响读

那么如果要提高写入的性能我们通常还可以调整segment块的大小避免太多碎片出现以及调整merge的时间

public void forceMerge(final boolean flush, int maxNumSegments, boolean onlyExpungeDeletes, final String forceMergeUUID)
    throws EngineException, IOException {
    if (onlyExpungeDeletes && maxNumSegments >= 0) {
        throw new IllegalArgumentException("only_expunge_deletes and max_num_segments are mutually exclusive");}
    //获取合并锁但是没有抢占读锁
    optimizeLock.lock();try {
        ensureOpen();store.incRef(); // increment the ref just to ensure nobody closes the store while we optimize
        try {
            if (onlyExpungeDeletes) {
                indexWriter.forceMergeDeletes(true /* blocks and waits for merges*/);} else if (maxNumSegments <= 0) {
                indexWriter.maybeMerge();} else {
                indexWriter.forceMerge(maxNumSegments, true /* blocks and waits for merges*/);this.forceMergeUUID = forceMergeUUID;}
            if (flush) {
                flush(false, true);// 这里会删除已经被打上delete标记的数据
                refresh("force-merge");}
        } finally {
            store.decRef();}
    } catch (AlreadyClosedException ex) {
        //释放任何因为异常而未合并的segment
        ensureOpen(ex);failOnTragicEvent(ex);throw ex;} catch (Exception e) {
        try {
            maybeFailEngine("force merge", e);} catch (Exception inner) {
            e.addSuppressed(inner);}
        throw e;} finally {
        optimizeLock.unlock();}
}