玩轉Elasticsearch源碼-ActionModule啓動分析

ActionModule的用途

org.elasticsearch.action.ActionModule主要維護了請求和響應相關組件，它們可能來自ES自己或者來自plugin。

• transportClient：標識是不是客戶端
• settings： 配置信息
• indexNameExpressionResolver：索引名稱表達式處理器
• indexScopedSettings：index級別的配置信息
• clusterSettings：cluster級別的配置信息
• settingsFilter：容許經過簡單正則表達式模式或完整設置鍵篩選設置對象的類。它用於rest層上的響應過濾，例如過濾出access keys等敏感信息
• actionPlugins：實現ActionPlugin接口的插件
• actions：請求和響應的處理邏輯
• actionFilters：從ActionPlugin中獲取，在TransportAction裏面requestFilterChain會去調用
• autoCreateIndex：封裝是否建立新索引的邏輯
• destructiveOperations：幫助處理破壞性操做和通配符使用
• restController：rest接口處理器

源碼分析

構造方法

先看看構造方法，每一步都簡單加了解釋

public ActionModule(boolean transportClient, Settings settings, IndexNameExpressionResolver indexNameExpressionResolver,
                        IndexScopedSettings indexScopedSettings, ClusterSettings clusterSettings, SettingsFilter settingsFilter,
                        ThreadPool threadPool, List<ActionPlugin> actionPlugins, NodeClient nodeClient,
            CircuitBreakerService circuitBreakerService, UsageService usageService) {
        this.transportClient = transportClient;//服務端爲false，客戶端爲true
        this.settings = settings;
        this.indexNameExpressionResolver = indexNameExpressionResolver;//將提供的索引表達式轉換爲實際的具體索引
        this.indexScopedSettings = indexScopedSettings;//index級別的配置
        this.clusterSettings = clusterSettings;//集羣級別的配置
        this.settingsFilter = settingsFilter;//容許經過簡單正則表達式模式或完整設置鍵篩選設置對象的類。它用於rest層上的響應過濾，例如過濾出access keys等敏感信息。
        this.actionPlugins = actionPlugins;//actionPlugins主要實現了 ActionPlugin 的 getActions()
        actions = setupActions(actionPlugins);//安裝actions
        actionFilters = setupActionFilters(actionPlugins);//從ActionPlugin中獲取，後面在TransportAction裏面requestFilterChain會去調用，之後再細講
        autoCreateIndex = transportClient ? null : new AutoCreateIndex(settings, clusterSettings, indexNameExpressionResolver);//封裝是否建立新索引的邏輯
        destructiveOperations = new DestructiveOperations(settings, clusterSettings);//幫助處理破壞性操做和通配符使用。
        Set<String> headers = actionPlugins.stream().flatMap(p -> p.getRestHeaders().stream()).collect(Collectors.toSet());
        UnaryOperator<RestHandler> restWrapper = null;//RestHandler包裝器，下面從actionPlugins裏面遍歷獲取
        for (ActionPlugin plugin : actionPlugins) {
            UnaryOperator<RestHandler> newRestWrapper = plugin.getRestHandlerWrapper(threadPool.getThreadContext());
            if (newRestWrapper != null) {
                logger.debug("Using REST wrapper from plugin " + plugin.getClass().getName());
                if (restWrapper != null) {
                    throw new IllegalArgumentException("Cannot have more than one plugin implementing a REST wrapper");
                }
                restWrapper = newRestWrapper;
            }
        }
        if (transportClient) {//客戶端不須要暴露http服務
            restController = null;
        } else {//服務端建立一個RestController,實現了Dispatcher接口，系統接受到請求後從netty的handler一路調用到 dispatchRequest 方法
            restController = new RestController(settings, headers, restWrapper, nodeClient, circuitBreakerService, usageService);
        }
    }

首先設置了settings等參數，而後進入

setupActions代碼比較好理解，直接貼源碼：

static Map<String, ActionHandler<?, ?>> setupActions(List<ActionPlugin> actionPlugins) {
        // Subclass NamedRegistry for easy registration
        class ActionRegistry extends NamedRegistry<ActionHandler<?, ?>> {
            ActionRegistry() {
                super("action");
            }

            public void register(ActionHandler<?, ?> handler) {
                register(handler.getAction().name(), handler);
            }

            public <Request extends ActionRequest, Response extends ActionResponse> void register(
                    GenericAction<Request, Response> action, Class<? extends TransportAction<Request, Response>> transportAction,
                    Class<?>... supportTransportActions) {
                register(new ActionHandler<>(action, transportAction, supportTransportActions));
            }
        }
        ActionRegistry actions = new ActionRegistry();

        actions.register(MainAction.INSTANCE, TransportMainAction.class);
        actions.register(NodesInfoAction.INSTANCE, TransportNodesInfoAction.class);
        actions.register(RemoteInfoAction.INSTANCE, TransportRemoteInfoAction.class);
        actions.register(NodesStatsAction.INSTANCE, TransportNodesStatsAction.class);
        actions.register(NodesUsageAction.INSTANCE, TransportNodesUsageAction.class);
        actions.register(NodesHotThreadsAction.INSTANCE, TransportNodesHotThreadsAction.class);
        actions.register(ListTasksAction.INSTANCE, TransportListTasksAction.class);
        actions.register(GetTaskAction.INSTANCE, TransportGetTaskAction.class);
        actions.register(CancelTasksAction.INSTANCE, TransportCancelTasksAction.class);

        actions.register(ClusterAllocationExplainAction.INSTANCE, TransportClusterAllocationExplainAction.class);
        actions.register(ClusterStatsAction.INSTANCE, TransportClusterStatsAction.class);
        actions.register(ClusterStateAction.INSTANCE, TransportClusterStateAction.class);
        actions.register(ClusterHealthAction.INSTANCE, TransportClusterHealthAction.class);
        actions.register(ClusterUpdateSettingsAction.INSTANCE, TransportClusterUpdateSettingsAction.class);
        actions.register(ClusterRerouteAction.INSTANCE, TransportClusterRerouteAction.class);
        actions.register(ClusterSearchShardsAction.INSTANCE, TransportClusterSearchShardsAction.class);
        actions.register(PendingClusterTasksAction.INSTANCE, TransportPendingClusterTasksAction.class);
        actions.register(PutRepositoryAction.INSTANCE, TransportPutRepositoryAction.class);
        actions.register(GetRepositoriesAction.INSTANCE, TransportGetRepositoriesAction.class);
        actions.register(DeleteRepositoryAction.INSTANCE, TransportDeleteRepositoryAction.class);
        actions.register(VerifyRepositoryAction.INSTANCE, TransportVerifyRepositoryAction.class);
        actions.register(GetSnapshotsAction.INSTANCE, TransportGetSnapshotsAction.class);
        actions.register(DeleteSnapshotAction.INSTANCE, TransportDeleteSnapshotAction.class);
        actions.register(CreateSnapshotAction.INSTANCE, TransportCreateSnapshotAction.class);
        actions.register(RestoreSnapshotAction.INSTANCE, TransportRestoreSnapshotAction.class);
        actions.register(SnapshotsStatusAction.INSTANCE, TransportSnapshotsStatusAction.class);

        actions.register(IndicesStatsAction.INSTANCE, TransportIndicesStatsAction.class);
        actions.register(IndicesSegmentsAction.INSTANCE, TransportIndicesSegmentsAction.class);
        actions.register(IndicesShardStoresAction.INSTANCE, TransportIndicesShardStoresAction.class);
        actions.register(CreateIndexAction.INSTANCE, TransportCreateIndexAction.class);
        actions.register(ShrinkAction.INSTANCE, TransportShrinkAction.class);
        actions.register(ResizeAction.INSTANCE, TransportResizeAction.class);
        actions.register(RolloverAction.INSTANCE, TransportRolloverAction.class);
        actions.register(DeleteIndexAction.INSTANCE, TransportDeleteIndexAction.class);
        actions.register(GetIndexAction.INSTANCE, TransportGetIndexAction.class);
        actions.register(OpenIndexAction.INSTANCE, TransportOpenIndexAction.class);
        actions.register(CloseIndexAction.INSTANCE, TransportCloseIndexAction.class);
        actions.register(IndicesExistsAction.INSTANCE, TransportIndicesExistsAction.class);
        actions.register(TypesExistsAction.INSTANCE, TransportTypesExistsAction.class);
        actions.register(GetMappingsAction.INSTANCE, TransportGetMappingsAction.class);
        actions.register(GetFieldMappingsAction.INSTANCE, TransportGetFieldMappingsAction.class,
                TransportGetFieldMappingsIndexAction.class);
        actions.register(PutMappingAction.INSTANCE, TransportPutMappingAction.class);
        actions.register(IndicesAliasesAction.INSTANCE, TransportIndicesAliasesAction.class);
        actions.register(UpdateSettingsAction.INSTANCE, TransportUpdateSettingsAction.class);
        actions.register(AnalyzeAction.INSTANCE, TransportAnalyzeAction.class);
        actions.register(PutIndexTemplateAction.INSTANCE, TransportPutIndexTemplateAction.class);
        actions.register(GetIndexTemplatesAction.INSTANCE, TransportGetIndexTemplatesAction.class);
        actions.register(DeleteIndexTemplateAction.INSTANCE, TransportDeleteIndexTemplateAction.class);
        actions.register(ValidateQueryAction.INSTANCE, TransportValidateQueryAction.class);
        actions.register(RefreshAction.INSTANCE, TransportRefreshAction.class);
        actions.register(FlushAction.INSTANCE, TransportFlushAction.class);
        actions.register(SyncedFlushAction.INSTANCE, TransportSyncedFlushAction.class);
        actions.register(ForceMergeAction.INSTANCE, TransportForceMergeAction.class);
        actions.register(UpgradeAction.INSTANCE, TransportUpgradeAction.class);
        actions.register(UpgradeStatusAction.INSTANCE, TransportUpgradeStatusAction.class);
        actions.register(UpgradeSettingsAction.INSTANCE, TransportUpgradeSettingsAction.class);
        actions.register(ClearIndicesCacheAction.INSTANCE, TransportClearIndicesCacheAction.class);
        actions.register(GetAliasesAction.INSTANCE, TransportGetAliasesAction.class);
        actions.register(AliasesExistAction.INSTANCE, TransportAliasesExistAction.class);
        actions.register(GetSettingsAction.INSTANCE, TransportGetSettingsAction.class);

        actions.register(IndexAction.INSTANCE, TransportIndexAction.class);
        actions.register(GetAction.INSTANCE, TransportGetAction.class);
        actions.register(TermVectorsAction.INSTANCE, TransportTermVectorsAction.class);
        actions.register(MultiTermVectorsAction.INSTANCE, TransportMultiTermVectorsAction.class,
                TransportShardMultiTermsVectorAction.class);
        actions.register(DeleteAction.INSTANCE, TransportDeleteAction.class);
        actions.register(UpdateAction.INSTANCE, TransportUpdateAction.class);
        actions.register(MultiGetAction.INSTANCE, TransportMultiGetAction.class,
                TransportShardMultiGetAction.class);
        actions.register(BulkAction.INSTANCE, TransportBulkAction.class,
                TransportShardBulkAction.class);
        actions.register(SearchAction.INSTANCE, TransportSearchAction.class);
        actions.register(SearchScrollAction.INSTANCE, TransportSearchScrollAction.class);
        actions.register(MultiSearchAction.INSTANCE, TransportMultiSearchAction.class);
        actions.register(ExplainAction.INSTANCE, TransportExplainAction.class);
        actions.register(ClearScrollAction.INSTANCE, TransportClearScrollAction.class);
        actions.register(RecoveryAction.INSTANCE, TransportRecoveryAction.class);

        //Indexed scripts
        actions.register(PutStoredScriptAction.INSTANCE, TransportPutStoredScriptAction.class);
        actions.register(GetStoredScriptAction.INSTANCE, TransportGetStoredScriptAction.class);
        actions.register(DeleteStoredScriptAction.INSTANCE, TransportDeleteStoredScriptAction.class);

        actions.register(FieldCapabilitiesAction.INSTANCE, TransportFieldCapabilitiesAction.class,
            TransportFieldCapabilitiesIndexAction.class);

        actions.register(PutPipelineAction.INSTANCE, PutPipelineTransportAction.class);
        actions.register(GetPipelineAction.INSTANCE, GetPipelineTransportAction.class);
        actions.register(DeletePipelineAction.INSTANCE, DeletePipelineTransportAction.class);
        actions.register(SimulatePipelineAction.INSTANCE, SimulatePipelineTransportAction.class);

        actionPlugins.stream().flatMap(p -> p.getActions().stream()).forEach(actions::register);

        return unmodifiableMap(actions.getRegistry());
    }

註冊了一大堆Action而後返回 Map<String, ActionHandler<?, ?>> 對象

能夠注意到ActionHandler包裝了action和transportAction的Class對象。action是對各個類型請求的request和response的包裝，並不是是真正的功能實現者，能夠看到它只是提供了兩個新建response和request的方法，及一個字NAME字段，這個NAME字段會用於後面action調用中。每一個action對應的功能實現是在對應的transportAction中。TransportAction通常都會有 @Inject 註解的構造方法用於依賴注入，而且實現doExecute執行具體邏輯。

configure方法

configure方法由ES封裝的注入器Injector調用，看看調用棧

能夠知道由ES啓動時候Node構造方法裏面發起的

injector = modules.createInjector();

configure裏面邏輯其實就是綁定一些對象到Injector中：

protected void configure() {
        bind(ActionFilters.class).toInstance(actionFilters);
        bind(DestructiveOperations.class).toInstance(destructiveOperations);

        if (false == transportClient) {
            // Supporting classes only used when not a transport client
            bind(AutoCreateIndex.class).toInstance(autoCreateIndex);
            bind(TransportLivenessAction.class).asEagerSingleton();

            // register GenericAction -> transportAction Map used by NodeClient
            @SuppressWarnings("rawtypes")
            MapBinder<GenericAction, TransportAction> transportActionsBinder
                    = MapBinder.newMapBinder(binder(), GenericAction.class, TransportAction.class);
            for (ActionHandler<?, ?> action : actions.values()) {
                // bind the action as eager singleton, so the map binder one will reuse it
                bind(action.getTransportAction()).asEagerSingleton();
                transportActionsBinder.addBinding(action.getAction()).to(action.getTransportAction()).asEagerSingleton();
                for (Class<?> supportAction : action.getSupportTransportActions()) {
                    bind(supportAction).asEagerSingleton();
                }
            }
        }
    }

initRestHandlers

initRestHandlers是在Node構造方法裏面直接調用的，主要是註冊一大堆RestHandler到restController用於處理http 請求

public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
        List<AbstractCatAction> catActions = new ArrayList<>();
        Consumer<RestHandler> registerHandler = a -> {
            if (a instanceof AbstractCatAction) {
                catActions.add((AbstractCatAction) a);
            }
        };
        registerHandler.accept(new RestMainAction(settings, restController));
        registerHandler.accept(new RestNodesInfoAction(settings, restController, settingsFilter));
        registerHandler.accept(new RestRemoteClusterInfoAction(settings, restController));
        registerHandler.accept(new RestNodesStatsAction(settings, restController));
        registerHandler.accept(new RestNodesUsageAction(settings, restController));
        registerHandler.accept(new RestNodesHotThreadsAction(settings, restController));
        registerHandler.accept(new RestClusterAllocationExplainAction(settings, restController));
        registerHandler.accept(new RestClusterStatsAction(settings, restController));
        registerHandler.accept(new RestClusterStateAction(settings, restController, settingsFilter));
        registerHandler.accept(new RestClusterHealthAction(settings, restController));
        registerHandler.accept(new RestClusterUpdateSettingsAction(settings, restController));
        registerHandler.accept(new RestClusterGetSettingsAction(settings, restController, clusterSettings, settingsFilter));
        registerHandler.accept(new RestClusterRerouteAction(settings, restController, settingsFilter));
        registerHandler.accept(new RestClusterSearchShardsAction(settings, restController));
        registerHandler.accept(new RestPendingClusterTasksAction(settings, restController));
        registerHandler.accept(new RestPutRepositoryAction(settings, restController));
        registerHandler.accept(new RestGetRepositoriesAction(settings, restController, settingsFilter));
        registerHandler.accept(new RestDeleteRepositoryAction(settings, restController));
        registerHandler.accept(new RestVerifyRepositoryAction(settings, restController));
        registerHandler.accept(new RestGetSnapshotsAction(settings, restController));
        registerHandler.accept(new RestCreateSnapshotAction(settings, restController));
        registerHandler.accept(new RestRestoreSnapshotAction(settings, restController));
        registerHandler.accept(new RestDeleteSnapshotAction(settings, restController));
        registerHandler.accept(new RestSnapshotsStatusAction(settings, restController));

        registerHandler.accept(new RestGetAllAliasesAction(settings, restController));
        registerHandler.accept(new RestGetAllMappingsAction(settings, restController));
        registerHandler.accept(new RestGetAllSettingsAction(settings, restController, indexScopedSettings, settingsFilter));
        registerHandler.accept(new RestGetIndicesAction(settings, restController, indexScopedSettings, settingsFilter));
        registerHandler.accept(new RestIndicesStatsAction(settings, restController));
        registerHandler.accept(new RestIndicesSegmentsAction(settings, restController));
        registerHandler.accept(new RestIndicesShardStoresAction(settings, restController));
        registerHandler.accept(new RestGetAliasesAction(settings, restController));
        registerHandler.accept(new RestIndexDeleteAliasesAction(settings, restController));
        registerHandler.accept(new RestIndexPutAliasAction(settings, restController));
        registerHandler.accept(new RestIndicesAliasesAction(settings, restController));
        registerHandler.accept(new RestCreateIndexAction(settings, restController));
        registerHandler.accept(new RestShrinkIndexAction(settings, restController));
        registerHandler.accept(new RestSplitIndexAction(settings, restController));
        registerHandler.accept(new RestRolloverIndexAction(settings, restController));
        registerHandler.accept(new RestDeleteIndexAction(settings, restController));
        registerHandler.accept(new RestCloseIndexAction(settings, restController));
        registerHandler.accept(new RestOpenIndexAction(settings, restController));

        registerHandler.accept(new RestUpdateSettingsAction(settings, restController));
        registerHandler.accept(new RestGetSettingsAction(settings, restController, indexScopedSettings, settingsFilter));

        registerHandler.accept(new RestAnalyzeAction(settings, restController));
        registerHandler.accept(new RestGetIndexTemplateAction(settings, restController));
        registerHandler.accept(new RestPutIndexTemplateAction(settings, restController));
        registerHandler.accept(new RestDeleteIndexTemplateAction(settings, restController));

        registerHandler.accept(new RestPutMappingAction(settings, restController));
        registerHandler.accept(new RestGetMappingAction(settings, restController));
        registerHandler.accept(new RestGetFieldMappingAction(settings, restController));

        registerHandler.accept(new RestRefreshAction(settings, restController));
        registerHandler.accept(new RestFlushAction(settings, restController));
        registerHandler.accept(new RestSyncedFlushAction(settings, restController));
        registerHandler.accept(new RestForceMergeAction(settings, restController));
        registerHandler.accept(new RestUpgradeAction(settings, restController));
        registerHandler.accept(new RestClearIndicesCacheAction(settings, restController));

        registerHandler.accept(new RestIndexAction(settings, restController));
        registerHandler.accept(new RestGetAction(settings, restController));
        registerHandler.accept(new RestGetSourceAction(settings, restController));
        registerHandler.accept(new RestMultiGetAction(settings, restController));
        registerHandler.accept(new RestDeleteAction(settings, restController));
        registerHandler.accept(new org.elasticsearch.rest.action.document.RestCountAction(settings, restController));
        registerHandler.accept(new RestTermVectorsAction(settings, restController));
        registerHandler.accept(new RestMultiTermVectorsAction(settings, restController));
        registerHandler.accept(new RestBulkAction(settings, restController));
        registerHandler.accept(new RestUpdateAction(settings, restController));

        registerHandler.accept(new RestSearchAction(settings, restController));
        registerHandler.accept(new RestSearchScrollAction(settings, restController));
        registerHandler.accept(new RestClearScrollAction(settings, restController));
        registerHandler.accept(new RestMultiSearchAction(settings, restController));

        registerHandler.accept(new RestValidateQueryAction(settings, restController));

        registerHandler.accept(new RestExplainAction(settings, restController));

        registerHandler.accept(new RestRecoveryAction(settings, restController));

        // Scripts API
        registerHandler.accept(new RestGetStoredScriptAction(settings, restController));
        registerHandler.accept(new RestPutStoredScriptAction(settings, restController));
        registerHandler.accept(new RestDeleteStoredScriptAction(settings, restController));

        registerHandler.accept(new RestFieldCapabilitiesAction(settings, restController));

        // Tasks API
        registerHandler.accept(new RestListTasksAction(settings, restController, nodesInCluster));
        registerHandler.accept(new RestGetTaskAction(settings, restController));
        registerHandler.accept(new RestCancelTasksAction(settings, restController, nodesInCluster));

        // Ingest API
        registerHandler.accept(new RestPutPipelineAction(settings, restController));
        registerHandler.accept(new RestGetPipelineAction(settings, restController));
        registerHandler.accept(new RestDeletePipelineAction(settings, restController));
        registerHandler.accept(new RestSimulatePipelineAction(settings, restController));

        // CAT API
        registerHandler.accept(new RestAllocationAction(settings, restController));
        registerHandler.accept(new RestShardsAction(settings, restController));
        registerHandler.accept(new RestMasterAction(settings, restController));
        registerHandler.accept(new RestNodesAction(settings, restController));
        registerHandler.accept(new RestTasksAction(settings, restController, nodesInCluster));
        registerHandler.accept(new RestIndicesAction(settings, restController, indexNameExpressionResolver));
        registerHandler.accept(new RestSegmentsAction(settings, restController));
        // Fully qualified to prevent interference with rest.action.count.RestCountAction
        registerHandler.accept(new org.elasticsearch.rest.action.cat.RestCountAction(settings, restController));
        // Fully qualified to prevent interference with rest.action.indices.RestRecoveryAction
        registerHandler.accept(new org.elasticsearch.rest.action.cat.RestRecoveryAction(settings, restController));
        registerHandler.accept(new RestHealthAction(settings, restController));
        registerHandler.accept(new org.elasticsearch.rest.action.cat.RestPendingClusterTasksAction(settings, restController));
        registerHandler.accept(new RestAliasAction(settings, restController));
        registerHandler.accept(new RestThreadPoolAction(settings, restController));
        registerHandler.accept(new RestPluginsAction(settings, restController));
        registerHandler.accept(new RestFielddataAction(settings, restController));
        registerHandler.accept(new RestNodeAttrsAction(settings, restController));
        registerHandler.accept(new RestRepositoriesAction(settings, restController));
        registerHandler.accept(new RestSnapshotAction(settings, restController));
        registerHandler.accept(new RestTemplatesAction(settings, restController));
        for (ActionPlugin plugin : actionPlugins) {
            for (RestHandler handler : plugin.getRestHandlers(settings, restController, clusterSettings, indexScopedSettings,
                    settingsFilter, indexNameExpressionResolver, nodesInCluster)) {
                registerHandler.accept(handler);
            }
        }
        registerHandler.accept(new RestCatAction(settings, restController, catActions));
    }

再深刻去看RestController裏面的註冊邏輯能夠看到是用Trie樹來作路徑匹配的，具體能夠看TrieNode類。