亲宝软件园·资讯

展开

elasticsearch集群cluster

zziawan 人气:0

前言

上一篇通过clusterservice对cluster做了一个简单的概述, 应该能够给大家一个初步认识。本篇将对cluster的代码组成进行详细分析,力求能够对cluster做一个更清晰的描述。cluster作为多个节点的协同工作机制,它需要节点,节点间通信,各个节点的状态及各个节点上的数据(index)状态。因此这一部分代码包括了上述的几个部分。

节点DiscoveryNode

首先是节点(DiscoveryNode),这里的节点不同于之前所说的node,只是集群上一个逻辑意义上的节点,只是一个实际节点的描述信息。它实现了Streamable接口和  serializable接口,可以在物理节点上传输。下图是它的field:

可以看到它只是一个节点的信息描述。在集群中每个节点会被抽象成一个DiscoveryNode,这些DiscoveryNode被封装到DiscoveryNodes中,同时提供一下操作如查找,连接等。这样集群维护所有节点的信息,同时可以根据集群状态进行节点的操作。

集群阻塞

集群还需要有一个机制就是集群阻塞,因为处于不同状态的集群能够进行的操作不同,如没有master节点的时候,所有的master操作都要停止,当前的任务是选举master,此时一个block就会引发,通知集群所有节点。不同于同一个jvm中的同步,不同的节点处在不同的jvm中,jvm的同步机制无法使用,因此只能使用这种阻塞机制进行节点间的协调。它的部分代码如下所示:

public class ClusterBlock implements Serializable, Streamable, ToXContent {
    private int id;
    private String description;
    private EnumSet<ClusterBlockLevel> levels;
    private boolean retryable;
    private boolean disableStatePersistence = false;
    private RestStatus status;
    ClusterBlock() {
    }
    public ClusterBlock(int id, String description, boolean retryable, boolean disableStatePersistence, RestStatus status, EnumSet<ClusterBlockLevel> levels) {
        this.id = id;
        this.description = description;
        this.retryable = retryable;
        this.disableStatePersistence = disableStatePersistence;
        this.status = status;
        this.levels = levels;
    }
}

阻塞主要由三部分组成,描述(description),阻塞级别(READ(0),WRITE(1), METADATA(2)),及restful状态码RestStatus构成。阻塞级别主要用于节点间对于index的操作的阻塞,如某个index在进行恢复过程时,它的状态是MATEDATA级别,此时不能够对其进行任何读写操作。 RestStatus主要用于restful请求的阻塞。最后要说的就是ack机制,集群的很多操作都需要节点响应。因此cluster定义了ack的请求和响应接口。所有需要ack的请求通过实现此ack接口都能够实现。另外集群还涉及matedata和routing,这两部分其实都是针对数据(index),如matedata主要是mapping,index, alias的相关元数据,因此这两部分会在分析数据功能时在做分析。

clusterService接口

说完了DiscoveryNode和block,接下来通过clusterService接口,它的作用主要是提供对外调用。这个接口主要提供Listener,block的add和remove及cluster状态的更新提交。cluster只是一个理论上的实体,其实并不存在,所有功能都是由各个节点来完成的。因此clusterService接口主要方法是集群状态监听器的操作。它的类图:

这里着重说一下submitStateUpdateTask的实现。对于节点来说,如果它探测到集群状态的更新,如果它是master则它需要向其它节点发布。代码如下:

public void submitStateUpdateTask(final String source, Priority priority, final ClusterStateUpdateTask updateTask) {
        if (!lifecycle.started()) {
            return;
        }
        try {
       //封装成updateTask
            final UpdateTask task = new UpdateTask(source, priority, updateTask);
       //会超时的任务
            if (updateTask instanceof TimeoutClusterStateUpdateTask) {
                final TimeoutClusterStateUpdateTask timeoutUpdateTask = (TimeoutClusterStateUpdateTask) updateTask;
                updateTasksExecutor.execute(task, threadPool.scheduler(), timeoutUpdateTask.timeout(), new Runnable() {
                    @Override
                    public void run() {
                        threadPool.generic().execute(new Runnable() {
                            @Override
                            public void run() {
                                timeoutUpdateTask.onFailure(task.source(), new ProcessClusterEventTimeoutException(timeoutUpdateTask.timeout(), task.source()));
                            }
                        });
                    }
                });
            } else {
                updateTasksExecutor.execute(task);
            }
        } catch (EsRejectedExecutionException e) {
            // ignore cases where we are shutting down..., there is really nothing interesting
            // to be done here...
            if (!lifecycle.stoppedOrClosed()) {
                throw e;
            }
        }
    }

上面的代码逻辑很简单,对于提交的task进行封装然后运行。这里运行的是ClusterStateUpdateTask, 它的实现很多,无法一一说明。但是它的方法说明了一切,它的类图如下所示:

子类的主要逻辑实现都在execute方法中,比如ZenDiscovery中handleMasterGone中的实现,master丢失后会进行master选举或者是试图加入新组成的集群。这些在后面的分析中可以看到。

总结

cluster是由很多功能组成的,如DiscoveryNode,block等。这一部分的主要代码是提供一些集群状态更新及监听的接口。集群状态的更新发布master独有的功能,但是监听集群状态变得时每个节点都需要的。

这些功能的具体实现在后面的分析中会逐步分析,希望大家以后多多支持!

加载全部内容

相关教程
猜你喜欢
用户评论