快捷導(dǎo)航

elasticsearch索引的創(chuàng)建過程index?create邏輯分析

更新時(shí)間：2022年04月22日 11:12:05 作者：zziawan

這篇文章主要介紹了elasticsearch索引核心index?create，索引的創(chuàng)建過程解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

索引的創(chuàng)建過程

從本篇開始，就進(jìn)入了Index的核心代碼部分。這里首先分析一下索引的創(chuàng)建過程。elasticsearch中的索引是多個(gè)分片的集合，它只是邏輯上的索引，并不具備實(shí)際的索引功能，所有對數(shù)據(jù)的操作最終還是由每個(gè)分片完成。

創(chuàng)建索引的過程，從elasticsearch集群上來說就是寫入索引元數(shù)據(jù)的過程，這一操作只能在master節(jié)點(diǎn)上完成。這是一個(gè)阻塞式動作，在加上分配在集群上均衡的過程也非常耗時(shí)，因此在一次創(chuàng)建大量索引的過程master節(jié)點(diǎn)會出現(xiàn)單點(diǎn)性能瓶頸，能夠看到響應(yīng)過程很慢。

在開始具體源碼分析之前，首先回顧一下Action部分的內(nèi)容（參考index action分析），elasticsearch的每一個(gè)功能都對應(yīng)兩個(gè)Action，*action和Transport*action。*action中定義了每個(gè)功能對應(yīng)的路徑，同時(shí)Action的instance綁定對應(yīng)的Transport*Action。所有功能請求都需要在集群上轉(zhuǎn)發(fā)，這大概也是每個(gè)功能都有Transport*Action的原因吧。對于create當(dāng)然也不例外，它的開始點(diǎn)也是TransportCreateAction。另外，在action support分析中分析過，不同的action需要經(jīng)過和需要操作的節(jié)點(diǎn)也不同。create index只能由master節(jié)點(diǎn)進(jìn)行，而且也只在master節(jié)點(diǎn)上進(jìn)行，保證集群數(shù)據(jù)的一致性。

materOperation方法實(shí)現(xiàn)

因此TransportCreateAction繼承了TransportMasterNodeOperationAction，并實(shí)現(xiàn)了materOperation方法。它的方法如下所示：

protected void masterOperation(final CreateIndexRequest request, final ClusterState state, final ActionListener&lt;CreateIndexResponse&gt; listener) throws ElasticsearchException {
        String cause = request.cause();
        if (cause.length() == 0) {
            cause = "api";
        }
        final CreateIndexClusterStateUpdateRequest updateRequest = new CreateIndexClusterStateUpdateRequest(request, cause, request.index())
                .ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout())
                .settings(request.settings()).mappings(request.mappings())
                .aliases(request.aliases()).customs(request.customs());
        createIndexService.createIndex(updateRequest, new ActionListener&lt;ClusterStateUpdateResponse&gt;() {
            @Override
            public void onResponse(ClusterStateUpdateResponse response) {
                listener.onResponse(new CreateIndexResponse(response.isAcknowledged()));
            }
            @Override
            public void onFailure(Throwable t) {
                if (t instanceof IndexAlreadyExistsException) {
                    logger.trace("[{}] failed to create", t, request.index());
                } else {
                    logger.debug("[{}] failed to create", t, request.index());
                }
                listener.onFailure(t);
            }
        });
    }

這里看上很簡單，只是調(diào)用了createIndexService（它其實(shí)是MetaDataCreateIndexService）的方法，就是修改集群matedata過程。

clusterservice處理

修改前首先獲取到index名稱對應(yīng)的lock，這樣保證操作數(shù)據(jù)一致性，然后生成updatetask，交給clusterservice處理。代碼如下所示：

public void createIndex(final CreateIndexClusterStateUpdateRequest request, final ActionListener&lt;ClusterStateUpdateResponse&gt; listener) {
        // 獲取鎖，只對該索引的操作加鎖，而不是整個(gè)cluster
        final Semaphore mdLock = metaDataService.indexMetaDataLock(request.index());
        // 如果能夠獲取鎖離開創(chuàng)建索引，否則在下面啟動新的線程進(jìn)行
        if (mdLock.tryAcquire()) {
            createIndex(request, listener, mdLock);
            return;
        }
        threadPool.executor(ThreadPool.Names.MANAGEMENT).execute(new ActionRunnable(listener) {
            @Override
            public void doRun() throws InterruptedException {
                if (!mdLock.tryAcquire(request.masterNodeTimeout().nanos(), TimeUnit.NANOSECONDS)) {
                    listener.onFailure(new ProcessClusterEventTimeoutException(request.masterNodeTimeout(), "acquire index lock"));
                    return;
                }
                createIndex(request, listener, mdLock);
            }
        });
    }

createIndex方法，會封裝create請求，然后向cluster發(fā)送一個(gè)updatetask。代碼如下所示：

private void createIndex(final CreateIndexClusterStateUpdateRequest request, final ActionListener&lt;ClusterStateUpdateResponse&gt; listener, final Semaphore mdLock) {
        ImmutableSettings.Builder updatedSettingsBuilder = ImmutableSettings.settingsBuilder();
        updatedSettingsBuilder.put(request.settings()).normalizePrefix(IndexMetaData.INDEX_SETTING_PREFIX);
        request.settings(updatedSettingsBuilder.build());
        clusterService.submitStateUpdateTask("create-index [" + request.index() + "], cause [" + request.cause() + "]", Priority.URGENT, new AckedClusterStateUpdateTask&lt;ClusterStateUpdateResponse&gt;(request, listener)

建立索引修改配置

增加或者修改mapping都是對集群狀態(tài)修改，它們的過程都很相似，都是通過clusterService提交一個(gè)更新操作，同時(shí)附帶有優(yōu)先級。clusterservice會根據(jù)優(yōu)先級和更新狀態(tài)task的類型來進(jìn)行對應(yīng)的操作。如下所示：

public void submitStateUpdateTask(final String source, Priority priority, final ClusterStateUpdateTask updateTask) {
        if (!lifecycle.started()) {
            return;
        }
        try {
            final UpdateTask task = new UpdateTask(source, priority, updateTask);//根據(jù)優(yōu)先級新建不同的task
            if (updateTask instanceof TimeoutClusterStateUpdateTask) {//超時(shí)任務(wù)，這類任務(wù)需要即時(shí)返回，因此立刻執(zhí)行。
                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 {//其它類型，可以延遲執(zhí)行，則交給線程池來執(zhí)行。
                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;
            }
        }
    }

說完它們的執(zhí)行過程，再來看一下create index的具體邏輯。這個(gè)邏輯在matedataservice所提交的AckedClusterStateUpdateTask中的execute方法中。總體來說，這一過程就是將request中關(guān)于索引的配置mapping等取出來加入到當(dāng)前的clustermatedata中，構(gòu)造一個(gè)新的matedata的過程。這一過程還是比較復(fù)雜，限于篇幅將在下次中進(jìn)行分析。