前言

上一篇通過clusterservice對cluster做了一個(gè)簡單的概述， 應(yīng)該能夠給大家一個(gè)初步認(rèn)識(shí)。本篇將對cluster的代碼組成進(jìn)行詳細(xì)分析，力求能夠?qū)luster做一個(gè)更清晰的描述。cluster作為多個(gè)節(jié)點(diǎn)的協(xié)同工作機(jī)制，它需要節(jié)點(diǎn)，節(jié)點(diǎn)間通信，各個(gè)節(jié)點(diǎn)的狀態(tài)及各個(gè)節(jié)點(diǎn)上的數(shù)據(jù)（index）狀態(tài)。因此這一部分代碼包括了上述的幾個(gè)部分。

節(jié)點(diǎn)DiscoveryNode

首先是節(jié)點(diǎn)（DiscoveryNode），這里的節(jié)點(diǎn)不同于之前所說的node，只是集群上一個(gè)邏輯意義上的節(jié)點(diǎn)，只是一個(gè)實(shí)際節(jié)點(diǎn)的描述信息。它實(shí)現(xiàn)了Streamable接口和  serializable接口，可以在物理節(jié)點(diǎn)上傳輸。下圖是它的field：

可以看到它只是一個(gè)節(jié)點(diǎn)的信息描述。在集群中每個(gè)節(jié)點(diǎn)會(huì)被抽象成一個(gè)DiscoveryNode，這些DiscoveryNode被封裝到DiscoveryNodes中，同時(shí)提供一下操作如查找，連接等。這樣集群維護(hù)所有節(jié)點(diǎn)的信息，同時(shí)可以根據(jù)集群狀態(tài)進(jìn)行節(jié)點(diǎn)的操作。

集群阻塞

集群還需要有一個(gè)機(jī)制就是集群阻塞，因?yàn)樘幱诓煌瑺顟B(tài)的集群能夠進(jìn)行的操作不同，如沒有master節(jié)點(diǎn)的時(shí)候，所有的master操作都要停止，當(dāng)前的任務(wù)是選舉master，此時(shí)一個(gè)block就會(huì)引發(fā)，通知集群所有節(jié)點(diǎn)。不同于同一個(gè)jvm中的同步，不同的節(jié)點(diǎn)處在不同的jvm中，jvm的同步機(jī)制無法使用，因此只能使用這種阻塞機(jī)制進(jìn)行節(jié)點(diǎn)間的協(xié)調(diào)。它的部分代碼如下所示：

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狀態(tài)碼RestStatus構(gòu)成。阻塞級別主要用于節(jié)點(diǎn)間對于index的操作的阻塞，如某個(gè)index在進(jìn)行恢復(fù)過程時(shí)，它的狀態(tài)是MATEDATA級別，此時(shí)不能夠?qū)ζ溥M(jìn)行任何讀寫操作。 RestStatus主要用于restful請求的阻塞。最后要說的就是ack機(jī)制，集群的很多操作都需要節(jié)點(diǎn)響應(yīng)。因此cluster定義了ack的請求和響應(yīng)接口。所有需要ack的請求通過實(shí)現(xiàn)此ack接口都能夠?qū)崿F(xiàn)。另外集群還涉及matedata和routing,這兩部分其實(shí)都是針對數(shù)據(jù)（index），如matedata主要是mapping，index， alias的相關(guān)元數(shù)據(jù)，因此這兩部分會(huì)在分析數(shù)據(jù)功能時(shí)在做分析。

clusterService接口

說完了DiscoveryNode和block，接下來通過clusterService接口，它的作用主要是提供對外調(diào)用。這個(gè)接口主要提供Listener，block的add和remove及cluster狀態(tài)的更新提交。cluster只是一個(gè)理論上的實(shí)體，其實(shí)并不存在，所有功能都是由各個(gè)節(jié)點(diǎn)來完成的。因此clusterService接口主要方法是集群狀態(tài)監(jiān)聽器的操作。它的類圖：

這里著重說一下submitStateUpdateTask的實(shí)現(xiàn)。對于節(jié)點(diǎn)來說，如果它探測到集群狀態(tài)的更新，如果它是master則它需要向其它節(jié)點(diǎn)發(fā)布。代碼如下：

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);
　　　　　　　//會(huì)超時(shí)的任務(wù)
            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進(jìn)行封裝然后運(yùn)行。這里運(yùn)行的是ClusterStateUpdateTask， 它的實(shí)現(xiàn)很多，無法一一說明。但是它的方法說明了一切，它的類圖如下所示：

子類的主要邏輯實(shí)現(xiàn)都在execute方法中，比如ZenDiscovery中handleMasterGone中的實(shí)現(xiàn)，master丟失后會(huì)進(jìn)行master選舉或者是試圖加入新組成的集群。這些在后面的分析中可以看到。

總結(jié)

cluster是由很多功能組成的，如DiscoveryNode，block等。這一部分的主要代碼是提供一些集群狀態(tài)更新及監(jiān)聽的接口。集群狀態(tài)的更新發(fā)布master獨(dú)有的功能，但是監(jiān)聽集群狀態(tài)變得時(shí)每個(gè)節(jié)點(diǎn)都需要的。

這些功能的具體實(shí)現(xiàn)在后面的分析中會(huì)逐步分析,希望大家以后多多支持腳本之家！

最新評論