Elasticsearch 6.1.0 启动过程

开篇

这篇文章主要目的是想梳理下elasticserach在启动过程中的核心步骤，宏观上讲解清楚elasticsearch启动过程中都做了哪些事情。

原本想通过流程图来进行画，后来网上有人通过xmind来分析整个过程，发现也能够讲解的非常清楚，因此同样采用xmind来自上而下讲解整个过程。

启动过程图

说明：

• 1.通过XMind记录ES启动流程的整个过程。
• 2.阅读顺序从上往下，标红色旗子的部分是核心流程。
• 3.核心流程我概括为：配置加载；Bootstrap 初始化; Bootstrap setup过程；Bootstrap start过程。
• 4.每个步骤当中细分下去很多逻辑，这里只讲解能够串联整个过程的逻辑。

配置加载过程

Bootstrap 初始化

• Elasticsearch的一个重要作用是解析命令参数。
• 执行带 -h 参数的Elasticsearch启动命令。
• Elasticsearch的构造函数如下所示，跟帮助信息是一致的。

 // elasticsearch启动命令帮助 Elasticsearch() { super("starts elasticsearch", () -> {}); versionOption = parser.acceptsAll(Arrays.asList("V", "version"), "Prints elasticsearch version information and exits"); daemonizeOption = parser.acceptsAll(Arrays.asList("d", "daemonize"), "Starts Elasticsearch in the background") .availableUnless(versionOption); pidfileOption = parser.acceptsAll(Arrays.asList("p", "pidfile"), "Creates a pid file in the specified path on start") .availableUnless(versionOption) .withRequiredArg() .withValuesConvertedBy(new PathConverter()); quietOption = parser.acceptsAll(Arrays.asList("q", "quiet"), "Turns off standard output/error streams logging in console") .availableUnless(versionOption) .availableUnless(daemonizeOption); }

Elasticsearch.main过程

 // elasticsearch启动入口函数 public static void main(final String[] args) throws Exception { System.setSecurityManager(new SecurityManager() { @Override public void checkPermission(Permission perm) { } }); LogConfigurator.registerErrorListener(); final Elasticsearch elasticsearch = new Elasticsearch(); int status = main(args, elasticsearch, Terminal.DEFAULT); if (status != ExitCodes.OK) { exit(status); } } // 调用elasticsearch对象的main函数 static int main(final String[] args, final Elasticsearch elasticsearch, final Terminal terminal) throws Exception { return elasticsearch.main(args, terminal); }

• 1.创建 SecurityManager 安全管理器
• 2.LogConfigurator.registerErrorListener() 注册侦听器
• 3.创建Elasticsearch对象
• 4.进入elasticsearch.main()过程。

elasticsearch.main过程

说明：

• Elasticsearch继承关系如上图。
• elasticsearch.main()过程当中会调用EnvironmentAwareCommand、Command等类。

class Elasticsearch extends EnvironmentAwareCommand { static int main(final String[] args, final Elasticsearch elasticsearch, final Terminal terminal) throws Exception { return elasticsearch.main(args, terminal); } } public abstract class Command implements Closeable { public final int main(String[] args, Terminal terminal) throws Exception { if (addShutdownHook()) { shutdownHookThread = new Thread(() -> { try { this.close(); } catch (final IOException e) { try ( StringWriter sw = new StringWriter(); PrintWriter pw = new PrintWriter(sw)) { e.printStackTrace(pw); terminal.println(sw.toString()); } catch (final IOException impossible) { } } }); Runtime.getRuntime().addShutdownHook(shutdownHookThread); } beforeMain.run(); try { mainWithoutErrorHandling(args, terminal); } catch (OptionException e) { return ExitCodes.USAGE; } catch (UserException e) { return e.exitCode; } return ExitCodes.OK; } void mainWithoutErrorHandling(String[] args, Terminal terminal) throws Exception { final OptionSet options = parser.parse(args); if (options.has(helpOption)) { printHelp(terminal); return; } if (options.has(silentOption)) { terminal.setVerbosity(Terminal.Verbosity.SILENT); } else if (options.has(verboseOption)) { terminal.setVerbosity(Terminal.Verbosity.VERBOSE); } else { terminal.setVerbosity(Terminal.Verbosity.NORMAL); } execute(terminal, options); } }

• 5.elasticsearch.main()直接进入Command.main()方法。
• 6.Command.main()给Runtime类添加一个hook线程，该线程作用是：当Runtime异常关闭时打印异常信息。
• 7.Command.mainWithoutErrorHandling 方法，根据命令行参数打印或者设置参数，然后执行命令。
• 8.进入EnvironmentAwareCommand.execute()方法。
  
  

EnvironmentAwareCommand.execute过程

public abstract class EnvironmentAwareCommand extends Command { protected void execute(Terminal terminal, OptionSet options) throws Exception { final Map<String, String> settings = new HashMap<>(); for (final KeyValuePair kvp : settingOption.values(options)) { if (kvp.value.isEmpty()) { throw new UserException(ExitCodes.USAGE, "setting [" + kvp.key + "] must not be empty"); } if (settings.containsKey(kvp.key)) { final String message = String.format( Locale.ROOT, "setting [%s] already set, saw [%s] and [%s]", kvp.key, settings.get(kvp.key), kvp.value); throw new UserException(ExitCodes.USAGE, message); } settings.put(kvp.key, kvp.value); } putSystemPropertyIfSettingIsMissing(settings, "path.data", "es.path.data"); putSystemPropertyIfSettingIsMissing(settings, "path.home", "es.path.home"); putSystemPropertyIfSettingIsMissing(settings, "path.logs", "es.path.logs"); execute(terminal, options, createEnv(terminal, settings)); } protected Environment createEnv(final Terminal terminal, final Map<String, String> settings) throws UserException { final String esPathConf = System.getProperty("es.path.conf"); if (esPathConf == null) { throw new UserException(ExitCodes.CONFIG, "the system property [es.path.conf] must be set"); } return InternalSettingsPreparer.prepareEnvironment(Settings.EMPTY, terminal, settings, getConfigPath(esPathConf)); } protected abstract void execute(Terminal terminal, OptionSet options, Environment env) throws Exception; }

• 9.EnvironmentAwareCommand.execute，确保 es.path.data, es.path.home, es.path.logs 等参数已设置，否则从 System.properties 中读取。
• 10.createEnv最后返回一个 Environment 对象，包含plugins，bin，lib，modules等目录下的文件信息
• 11.进入elasticsearch的execute()方法。

prepareEnvironment过程

public class InternalSettingsPreparer { public static Environment prepareEnvironment(Settings input, Terminal terminal, Map<String, String> properties, Path configPath) { // just create enough settings to build the environment, to get the config dir Settings.Builder output = Settings.builder(); initializeSettings(output, input, properties); Environment environment = new Environment(output.build(), configPath); if (Files.exists(environment.configFile().resolve("elasticsearch.yaml"))) { throw new SettingsException("elasticsearch.yaml was deprecated in 5.5.0 and must be renamed to elasticsearch.yml"); } if (Files.exists(environment.configFile().resolve("elasticsearch.json"))) { throw new SettingsException("elasticsearch.json was deprecated in 5.5.0 and must be converted to elasticsearch.yml"); } output = Settings.builder(); // start with a fresh output Path path = environment.configFile().resolve("elasticsearch.yml"); if (Files.exists(path)) { try { output.loadFromPath(path); } catch (IOException e) { throw new SettingsException("Failed to load settings from " + path.toString(), e); } } // re-initialize settings now that the config file has been loaded initializeSettings(output, input, properties); finalizeSettings(output, terminal); environment = new Environment(output.build(), configPath); // we put back the path.logs so we can use it in the logging configuration file output.put(Environment.PATH_LOGS_SETTING.getKey(), environment.logsFile().toAbsolutePath().normalize().toString()); return new Environment(output.build(), configPath); } } public class Environment { public Environment(final Settings settings, final Path configPath) { final Path homeFile; if (PATH_HOME_SETTING.exists(settings)) { homeFile = PathUtils.get(PATH_HOME_SETTING.get(settings)).normalize(); } else { throw new IllegalStateException(PATH_HOME_SETTING.getKey() + " is not configured"); } if (configPath != null) { configFile = configPath.normalize(); } else { configFile = homeFile.resolve("config"); } pluginsFile = homeFile.resolve("plugins"); List<String> dataPaths = PATH_DATA_SETTING.get(settings); final ClusterName clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings); if (DiscoveryNode.nodeRequiresLocalStorage(settings)) { if (dataPaths.isEmpty() == false) { dataFiles = new Path[dataPaths.size()]; dataWithClusterFiles = new Path[dataPaths.size()]; for (int i = 0; i < dataPaths.size(); i++) { dataFiles[i] = PathUtils.get(dataPaths.get(i)); dataWithClusterFiles[i] = dataFiles[i].resolve(clusterName.value()); } } else { dataFiles = new Path[]{homeFile.resolve("data")}; dataWithClusterFiles = new Path[]{homeFile.resolve("data").resolve(clusterName.value())}; } } else { if (dataPaths.isEmpty()) { dataFiles = dataWithClusterFiles = EMPTY_PATH_ARRAY; } else { final String paths = String.join(",", dataPaths); throw new IllegalStateException("node does not require local storage yet path.data is set to [" + paths + "]"); } } if (PATH_SHARED_DATA_SETTING.exists(settings)) { sharedDataFile = PathUtils.get(PATH_SHARED_DATA_SETTING.get(settings)).normalize(); } else { sharedDataFile = null; } List<String> repoPaths = PATH_REPO_SETTING.get(settings); if (repoPaths.isEmpty()) { repoFiles = EMPTY_PATH_ARRAY; } else { repoFiles = new Path[repoPaths.size()]; for (int i = 0; i < repoPaths.size(); i++) { repoFiles[i] = PathUtils.get(repoPaths.get(i)); } } // this is trappy, Setting#get(Settings) will get a fallback setting yet return false for Settings#exists(Settings) if (PATH_LOGS_SETTING.exists(settings)) { logsFile = PathUtils.get(PATH_LOGS_SETTING.get(settings)).normalize(); } else { logsFile = homeFile.resolve("logs"); } if (PIDFILE_SETTING.exists(settings)) { pidFile = PathUtils.get(PIDFILE_SETTING.get(settings)).normalize(); } else { pidFile = null; } binFile = homeFile.resolve("bin"); libFile = homeFile.resolve("lib"); modulesFile = homeFile.resolve("modules"); Settings.Builder finalSettings = Settings.builder().put(settings); finalSettings.put(PATH_HOME_SETTING.getKey(), homeFile); if (PATH_DATA_SETTING.exists(settings)) { finalSettings.putList(PATH_DATA_SETTING.getKey(), dataPaths); } finalSettings.put(PATH_LOGS_SETTING.getKey(), logsFile.toString()); this.settings = finalSettings.build(); } }

• 12.createEnv最后返回一个 Environment 对象。

elasticsearch.execute过程

class Elasticsearch extends EnvironmentAwareCommand { protected void execute(Terminal terminal, OptionSet options, Environment env) throws UserException { if (options.nonOptionArguments().isEmpty() == false) { } if (options.has(versionOption)) { terminal.println("Version: " + Version.displayVersion(Version.CURRENT, Build.CURRENT.isSnapshot()) + ", Build: " + Build.CURRENT.shortHash() + "/" + Build.CURRENT.date() + ", JVM: " + JvmInfo.jvmInfo().version()); return; } final boolean daemonize = options.has(daemonizeOption); final Path pidFile = pidfileOption.value(options); final boolean quiet = options.has(quietOption); try { init(daemonize, pidFile, quiet, env); } catch (NodeValidationException e) { } } void init(final boolean daemonize, final Path pidFile, final boolean quiet, Environment initialEnv) throws NodeValidationException, UserException { try { Bootstrap.init(!daemonize, pidFile, quiet, initialEnv); } catch (BootstrapException | RuntimeException e) { } } }

• 13.Elasticsearch.execute ，读取daemonize， pidFile，quiet 的值，并 确保配置的临时目录(temp)是有效目录
• 14.进入Bootstrap初始化阶段

Bootstrap init过程

 static void init( final boolean foreground, final Path pidFile, final boolean quiet, final Environment initialEnv) throws BootstrapException, NodeValidationException, UserException { BootstrapInfo.init(); INSTANCE = new Bootstrap(); final SecureSettings keystore = loadSecureSettings(initialEnv); final Environment environment = createEnvironment(foreground, pidFile, keystore, initialEnv.settings(), initialEnv.configFile()); try { LogConfigurator.configure(environment); } catch (IOException e) { throw new BootstrapException(e); } if (environment.pidFile() != null) { try { PidFile.create(environment.pidFile(), true); } catch (IOException e) { throw new BootstrapException(e); } } final boolean closeStandardStreams = (foreground == false) || quiet; try { if (closeStandardStreams) { final Logger rootLogger = ESLoggerFactory.getRootLogger(); final Appender maybeConsoleAppender = Loggers.findAppender(rootLogger, ConsoleAppender.class); if (maybeConsoleAppender != null) { Loggers.removeAppender(rootLogger, maybeConsoleAppender); } closeSystOut(); } checkLucene(); Thread.setDefaultUncaughtExceptionHandler( new ElasticsearchUncaughtExceptionHandler(() -> Node.NODE_NAME_SETTING.get(environment.settings()))); INSTANCE.setup(true, environment); try { IOUtils.close(keystore); } catch (IOException e) { throw new BootstrapException(e); } INSTANCE.start(); if (closeStandardStreams) { closeSysError(); } } catch (NodeValidationException | RuntimeException e) { throw e; } }

• 15.创建Bootstrap对象， INSTANCE = new Bootstrap()。
• 16.初始化Bootstrap对象，INSTANCE.setup(true, environment)。
• 17.启动Bootstrap对象INSTANCE.start()。

Bootstrap new过程

 Bootstrap() { keepAliveThread = new Thread(new Runnable() { @Override public void run() { try { keepAliveLatch.await(); } catch (InterruptedException e) { // bail out } } }, "elasticsearch[keepAlive/" + Version.CURRENT + "]"); keepAliveThread.setDaemon(false); // keep this thread alive (non daemon thread) until we shutdown Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { keepAliveLatch.countDown(); } }); }

• 18.创建Bootstrap对象，该类构造函数会创建一个用户线程添加到Runtime Hook中，进行 countDown 操作。

Bootstrap setup过程

 private void setup(boolean addShutdownHook, Environment environment) throws BootstrapException { Settings settings = environment.settings(); try { spawner.spawnNativePluginControllers(environment); } catch (IOException e) { throw new BootstrapException(e); } initializeNatives( environment.tmpFile(), BootstrapSettings.MEMORY_LOCK_SETTING.get(settings), BootstrapSettings.SYSTEM_CALL_FILTER_SETTING.get(settings), BootstrapSettings.CTRLHANDLER_SETTING.get(settings)); // initialize probes before the security manager is installed initializeProbes(); if (addShutdownHook) { Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { try { IOUtils.close(node, spawner); LoggerContext context = (LoggerContext) LogManager.getContext(false); Configurator.shutdown(context); } catch (IOException ex) { throw new ElasticsearchException("failed to stop node", ex); } } }); } try { JarHell.checkJarHell(); } catch (IOException | URISyntaxException e) { } IfConfig.logIfNecessary(); try { Security.configure(environment, BootstrapSettings.SECURITY_FILTER_BAD_DEFAULTS_SETTING.get(settings)); } catch (IOException | NoSuchAlgorithmException e) { } node = new Node(environment) { @Override protected void validateNodeBeforeAcceptingRequests( final BootstrapContext context, final BoundTransportAddress boundTransportAddress, List<BootstrapCheck> checks) throws NodeValidationException { BootstrapChecks.check(context, boundTransportAddress, checks); } }; }

• 19.为每个插件生成生成本机控制类，spawner.spawnNativePluginControllers(environment)；尝试为给定模块生成控制器(native Controller)守护程序。 生成的进程将通过其stdin，stdout和stderr流保持与此JVM的连接，但对此包之外的代码不能使用对这些流的引用。

• 20.初始化本地资源 initializeNatives()。
• 21.使用 JarHell 检查重复的 jar 文件 JarHell.checkJarHell()。
• 22.创建 node 节点 new Node(environment)，核心！！！。

Bootstrap start过程

 private void start() throws NodeValidationException { node.start(); keepAliveThread.start(); }

• 23.启动node对象，node.start()。
• 24.启动守护进程，keepAliveThread.start()。

node创建过程

 protected Node(final Environment environment, Collection<Class<? extends Plugin>> classpathPlugins) { try { Settings tmpSettings = Settings.builder().put(environment.settings()) .put(Client.CLIENT_TYPE_SETTING_S.getKey(), CLIENT_TYPE).build(); try { nodeEnvironment = new NodeEnvironment(tmpSettings, environment); resourcesToClose.add(nodeEnvironment); } catch (IOException ex) { } final boolean hadPredefinedNodeName = NODE_NAME_SETTING.exists(tmpSettings); Logger logger = Loggers.getLogger(Node.class, tmpSettings); final String nodeId = nodeEnvironment.nodeId(); tmpSettings = addNodeNameIfNeeded(tmpSettings, nodeId); final JvmInfo jvmInfo = JvmInfo.jvmInfo(); this.pluginsService = new PluginsService(tmpSettings, environment.configFile(), environment.modulesFile(), environment.pluginsFile(), classpathPlugins); this.settings = pluginsService.updatedSettings(); localNodeFactory = new LocalNodeFactory(settings, nodeEnvironment.nodeId()); this.environment = new Environment(this.settings, environment.configFile()); final List<ExecutorBuilder<?>> executorBuilders = pluginsService.getExecutorBuilders(settings); final ThreadPool threadPool = new ThreadPool(settings, executorBuilders.toArray(new ExecutorBuilder[0])); resourcesToClose.add(() -> ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS)); DeprecationLogger.setThreadContext(threadPool.getThreadContext()); resourcesToClose.add(() -> DeprecationLogger.removeThreadContext(threadPool.getThreadContext())); final List<Setting<?>> additionalSettings = new ArrayList<>(pluginsService.getPluginSettings()); final List<String> additionalSettingsFilter = new ArrayList<>(pluginsService.getPluginSettingsFilter()); for (final ExecutorBuilder<?> builder : threadPool.builders()) { additionalSettings.addAll(builder.getRegisteredSettings()); } client = new NodeClient(settings, threadPool); .................. }

• 25.创建一个 NodeEnvironment 对象保存节点环境信息，如各种数据文件的路径
• 26.读取JVM信息
• 27.创建 PluginsService 对象，创建过程中会读取并加载所有的模块和插件
• 28.创建一个最终的 Environment 对象
• 29.创建线程池 ThreadPool 后面各类对象基本都是通过线程来提供服务，这个线程池可以管理各类线程
• 30.创建 节点客户端 NodeClient

 protected Node(final Environment environment, Collection<Class<? extends Plugin>> classpathPlugins) { .................. final ResourceWatcherService resourceWatcherService = new ResourceWatcherService(settings, threadPool); final ScriptModule scriptModule = new ScriptModule(settings, pluginsService.filterPlugins(ScriptPlugin.class)); AnalysisModule analysisModule = new AnalysisModule(this.environment, pluginsService.filterPlugins(AnalysisPlugin.class)); final SettingsModule settingsModule = new SettingsModule(this.settings, additionalSettings, additionalSettingsFilter); scriptModule.registerClusterSettingsListeners(settingsModule.getClusterSettings()); resourcesToClose.add(resourceWatcherService); final NetworkService networkService = new NetworkService( getCustomNameResolvers(pluginsService.filterPlugins(DiscoveryPlugin.class))); List<ClusterPlugin> clusterPlugins = pluginsService.filterPlugins(ClusterPlugin.class); final ClusterService clusterService = new ClusterService(settings, settingsModule.getClusterSettings(), threadPool, ClusterModule.getClusterStateCustomSuppliers(clusterPlugins)); clusterService.addListener(scriptModule.getScriptService()); resourcesToClose.add(clusterService); final IngestService ingestService = new IngestService(settings, threadPool, this.environment, scriptModule.getScriptService(), analysisModule.getAnalysisRegistry(), pluginsService.filterPlugins(IngestPlugin.class)); final DiskThresholdMonitor listener = new DiskThresholdMonitor(settings, clusterService::state, clusterService.getClusterSettings(), client); final ClusterInfoService clusterInfoService = newClusterInfoService(settings, clusterService, threadPool, client, listener::onNewInfo); final UsageService usageService = new UsageService(settings); }

• 31.创建各种服务类对象 ResourceWatcherService、NetworkService、ClusterService、IngestService、ClusterInfoService、UsageService、MonitorService、CircuitBreakerService、MetaStateService、IndicesService、MetaDataIndexUpgradeService、TemplateUpgradeService、TransportService、ResponseCollectorService、SearchTransportService、NodeService、SearchService、PersistentTasksClusterService,这些服务类是的功能可以根据名称做一个大概的判断。

 protected Node(final Environment environment, Collection<Class<? extends Plugin>> classpathPlugins) { .................. ModulesBuilder modules = new ModulesBuilder(); // plugin modules must be added here, before others or we can get crazy injection errors... for (Module pluginModule : pluginsService.createGuiceModules()) { modules.add(pluginModule); } final MonitorService monitorService = new MonitorService(settings, nodeEnvironment, threadPool, clusterInfoService); ClusterModule clusterModule = new ClusterModule(settings, clusterService, clusterPlugins, clusterInfoService); modules.add(clusterModule); IndicesModule indicesModule = new IndicesModule(pluginsService.filterPlugins(MapperPlugin.class)); modules.add(indicesModule); SearchModule searchModule = new SearchModule(settings, false, pluginsService.filterPlugins(SearchPlugin.class)); CircuitBreakerService circuitBreakerService = createCircuitBreakerService(settingsModule.getSettings(), settingsModule.getClusterSettings()); resourcesToClose.add(circuitBreakerService); modules.add(new GatewayModule()); ActionModule actionModule = new ActionModule(false, settings, clusterModule.getIndexNameExpressionResolver(), settingsModule.getIndexScopedSettings(), settingsModule.getClusterSettings(), settingsModule.getSettingsFilter(), threadPool, pluginsService.filterPlugins(ActionPlugin.class), client, circuitBreakerService, usageService); modules.add(actionModule); final RestController restController = actionModule.getRestController(); final NetworkModule networkModule = new NetworkModule(settings, false, pluginsService.filterPlugins(NetworkPlugin.class), threadPool, bigArrays, circuitBreakerService, namedWriteableRegistry, xContentRegistry, networkService, restController); Collection<UnaryOperator<Map<String, MetaData.Custom>>> customMetaDataUpgraders = pluginsService.filterPlugins(Plugin.class) .stream().map(Plugin::getCustomMetaDataUpgrader) .collect(Collectors.toList()); Collection<UnaryOperator<Map<String, IndexTemplateMetaData>>> indexTemplateMetaDataUpgraders = pluginsService.filterPlugins(Plugin.class).stream() .map(Plugin::getIndexTemplateMetaDataUpgrader) .collect(Collectors.toList()); Collection<UnaryOperator<IndexMetaData>> indexMetaDataUpgraders = pluginsService.filterPlugins(Plugin.class) .stream().map(Plugin::getIndexMetaDataUpgrader).collect(Collectors.toList()); final MetaDataUpgrader metaDataUpgrader = new MetaDataUpgrader(customMetaDataUpgraders, indexTemplateMetaDataUpgraders); final MetaDataIndexUpgradeService metaDataIndexUpgradeService = new MetaDataIndexUpgradeService(settings, xContentRegistry,indicesModule.getMapperRegistry(), settingsModule.getIndexScopedSettings(), indexMetaDataUpgraders); final GatewayMetaState gatewayMetaState = new GatewayMetaState(settings, nodeEnvironment, metaStateService, metaDataIndexUpgradeService, metaDataUpgrader); new TemplateUpgradeService(settings, client, clusterService, threadPool, indexTemplateMetaDataUpgraders); final Transport transport = networkModule.getTransportSupplier().get(); final TransportService transportService = newTransportService(settings, transport, threadPool, networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings()); final ResponseCollectorService responseCollectorService = new ResponseCollectorService(this.settings, clusterService); final SearchTransportService searchTransportService = new SearchTransportService(settings, transportService,SearchExecutionStatsCollector.makeWrapper(responseCollectorService)); final Consumer<Binder> httpBind; final HttpServerTransport httpServerTransport; if (networkModule.isHttpEnabled()) { httpServerTransport = networkModule.getHttpServerTransportSupplier().get(); httpBind = b -> { b.bind(HttpServerTransport.class).toInstance(httpServerTransport); }; } else { httpBind = b -> { b.bind(HttpServerTransport.class).toProvider(Providers.of(null)); }; httpServerTransport = null; } final DiscoveryModule discoveryModule = new DiscoveryModule(this.settings, threadPool, transportService, namedWriteableRegistry, networkService, clusterService.getMasterService(), clusterService.getClusterApplierService(), clusterService.getClusterSettings(), pluginsService.filterPlugins(DiscoveryPlugin.class), clusterModule.getAllocationService()); this.nodeService = new NodeService(settings, threadPool, monitorService, discoveryModule.getDiscovery(), transportService, indicesService, pluginsService, circuitBreakerService, scriptModule.getScriptService(), httpServerTransport, ingestService, clusterService, settingsModule.getSettingsFilter(), responseCollectorService, searchTransportService);

• 32.ModulesBuilder类加入各种模块 ScriptModule、AnalysisModule、SettingsModule、pluginModule、ClusterModule、IndicesModule、SearchModule、GatewayModule、RepositoriesModule、ActionModule、NetworkModule、DiscoveryModule

 protected Node(final Environment environment, Collection<Class<? extends Plugin>> classpathPlugins) { .................. modules.add(b -> { b.bind(Node.class).toInstance(this); b.bind(NodeService.class).toInstance(nodeService); b.bind(NamedXContentRegistry.class).toInstance(xContentRegistry); b.bind(PluginsService.class).toInstance(pluginsService); b.bind(Client.class).toInstance(client); b.bind(NodeClient.class).toInstance(client); b.bind(Environment.class).toInstance(this.environment); b.bind(ThreadPool.class).toInstance(threadPool); b.bind(NodeEnvironment.class).toInstance(nodeEnvironment); b.bind(ResourceWatcherService.class).toInstance(resourceWatcherService); b.bind(CircuitBreakerService.class).toInstance(circuitBreakerService); b.bind(BigArrays.class).toInstance(bigArrays); b.bind(ScriptService.class).toInstance(scriptModule.getScriptService()); b.bind(AnalysisRegistry.class).toInstance(analysisModule.getAnalysisRegistry()); b.bind(IngestService.class).toInstance(ingestService); b.bind(UsageService.class).toInstance(usageService); b.bind(NamedWriteableRegistry.class).toInstance(namedWriteableRegistry); b.bind(MetaDataUpgrader.class).toInstance(metaDataUpgrader); b.bind(MetaStateService.class).toInstance(metaStateService); b.bind(IndicesService.class).toInstance(indicesService); b.bind(SearchService.class).toInstance(newSearchService(clusterService, indicesService, threadPool, scriptModule.getScriptService(), bigArrays, searchModule.getFetchPhase(), responseCollectorService)); b.bind(SearchTransportService.class).toInstance(searchTransportService); b.bind(SearchPhaseController.class).toInstance(new SearchPhaseController(settings, bigArrays, scriptModule.getScriptService())); b.bind(Transport.class).toInstance(transport); b.bind(TransportService.class).toInstance(transportService); b.bind(NetworkService.class).toInstance(networkService); b.bind(UpdateHelper.class).toInstance(new UpdateHelper(settings, scriptModule.getScriptService())); b.bind(MetaDataIndexUpgradeService.class).toInstance(metaDataIndexUpgradeService); b.bind(ClusterInfoService.class).toInstance(clusterInfoService); b.bind(GatewayMetaState.class).toInstance(gatewayMetaState); b.bind(Discovery.class).toInstance(discoveryModule.getDiscovery()); { RecoverySettings recoverySettings = new RecoverySettings(settings, settingsModule.getClusterSettings()); processRecoverySettings(settingsModule.getClusterSettings(), recoverySettings); b.bind(PeerRecoverySourceService.class).toInstance( new PeerRecoverySourceService(settings, transportService, indicesService, recoverySettings)); b.bind(PeerRecoveryTargetService.class).toInstance( new PeerRecoveryTargetService(settings, threadPool, transportService, recoverySettings, clusterService)); } httpBind.accept(b); pluginComponents.stream().forEach(p -> b.bind((Class) p.getClass()).toInstance(p)); } ); injector = modules.createInjector(); clusterModule.getAllocationService().setGatewayAllocator(injector.getInstance(GatewayAllocator.class)); List<LifecycleComponent> pluginLifecycleComponents = pluginComponents.stream() .filter(p -> p instanceof LifecycleComponent) .map(p -> (LifecycleComponent) p).collect(Collectors.toList()); pluginLifecycleComponents.addAll(pluginsService.getGuiceServiceClasses().stream() .map(injector::getInstance).collect(Collectors.toList())); resourcesToClose.addAll(pluginLifecycleComponents); this.pluginLifecycleComponents = Collections.unmodifiableList(pluginLifecycleComponents); client.initialize(injector.getInstance(new Key<Map<GenericAction, TransportAction>>() {}), () -> clusterService.localNode().getId()); if (NetworkModule.HTTP_ENABLED.get(settings)) { logger.debug("initializing HTTP handlers ..."); actionModule.initRestHandlers(() -> clusterService.state().nodes()); } success = true; } catch (IOException ex) { } finally { } }

• 33.elasticsearch里面的组件基本都进行进行了模块化管理，elasticsearch对guice进行了封装通过ModulesBuilder类构建es的模块.

node启动过程

 public Node start() throws NodeValidationException { if (!lifecycle.moveToStarted()) { return this; } Logger logger = Loggers.getLogger(Node.class, NODE_NAME_SETTING.get(settings)); logger.info("starting ..."); pluginLifecycleComponents.forEach(LifecycleComponent::start); injector.getInstance(MappingUpdatedAction.class).setClient(client); injector.getInstance(IndicesService.class).start(); injector.getInstance(IndicesClusterStateService.class).start(); injector.getInstance(SnapshotsService.class).start(); injector.getInstance(SnapshotShardsService.class).start(); injector.getInstance(RoutingService.class).start(); injector.getInstance(SearchService.class).start(); nodeService.getMonitorService().start(); final ClusterService clusterService = injector.getInstance(ClusterService.class); final NodeConnectionsService nodeConnectionsService = injector.getInstance(NodeConnectionsService.class); nodeConnectionsService.start(); clusterService.setNodeConnectionsService(nodeConnectionsService); injector.getInstance(ResourceWatcherService.class).start(); injector.getInstance(GatewayService.class).start(); Discovery discovery = injector.getInstance(Discovery.class); clusterService.getMasterService().setClusterStatePublisher(discovery::publish); // Start the transport service now so the publish address will be added to the local disco node in ClusterService TransportService transportService = injector.getInstance(TransportService.class); transportService.getTaskManager().setTaskResultsService(injector.getInstance(TaskResultsService.class)); transportService.start(); assert localNodeFactory.getNode() != null; assert transportService.getLocalNode().equals(localNodeFactory.getNode()) : "transportService has a different local node than the factory provided"; final MetaData onDiskMetadata; try { // we load the global state here (the persistent part of the cluster state stored on disk) to // pass it to the bootstrap checks to allow plugins to enforce certain preconditions based on the recovered state. if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.isDataNode(settings)) { onDiskMetadata = injector.getInstance(GatewayMetaState.class).loadMetaState(); } else { onDiskMetadata = MetaData.EMPTY_META_DATA; } assert onDiskMetadata != null : "metadata is null but shouldn't"; // this is never null } catch (IOException e) { throw new UncheckedIOException(e); } validateNodeBeforeAcceptingRequests(new BootstrapContext(settings, onDiskMetadata), transportService.boundAddress(), pluginsService .filterPlugins(Plugin .class) .stream() .flatMap(p -> p.getBootstrapChecks().stream()).collect(Collectors.toList())); clusterService.addStateApplier(transportService.getTaskManager()); // start after transport service so the local disco is known discovery.start(); // start before cluster service so that it can set initial state on ClusterApplierService clusterService.start(); assert clusterService.localNode().equals(localNodeFactory.getNode()) : "clusterService has a different local node than the factory provided"; transportService.acceptIncomingRequests(); discovery.startInitialJoin(); // tribe nodes don't have a master so we shouldn't register an observer s final TimeValue initialStateTimeout = DiscoverySettings.INITIAL_STATE_TIMEOUT_SETTING.get(settings); if (initialStateTimeout.millis() > 0) { final ThreadPool thread = injector.getInstance(ThreadPool.class); ClusterState clusterState = clusterService.state(); ClusterStateObserver observer = new ClusterStateObserver(clusterState, clusterService, null, logger, thread.getThreadContext()); if (clusterState.nodes().getMasterNodeId() == null) { logger.debug("waiting to join the cluster. timeout [{}]", initialStateTimeout); final CountDownLatch latch = new CountDownLatch(1); observer.waitForNextChange(new ClusterStateObserver.Listener() { @Override public void onNewClusterState(ClusterState state) { latch.countDown(); } @Override public void onClusterServiceClose() { latch.countDown(); } @Override public void onTimeout(TimeValue timeout) { logger.warn("timed out while waiting for initial discovery state - timeout: {}", initialStateTimeout); latch.countDown(); } }, state -> state.nodes().getMasterNodeId() != null, initialStateTimeout); try { latch.await(); } catch (InterruptedException e) { throw new ElasticsearchTimeoutException("Interrupted while waiting for initial discovery state"); } } } if (NetworkModule.HTTP_ENABLED.get(settings)) { injector.getInstance(HttpServerTransport.class).start(); } if (WRITE_PORTS_FILE_SETTING.get(settings)) { if (NetworkModule.HTTP_ENABLED.get(settings)) { HttpServerTransport http = injector.getInstance(HttpServerTransport.class); writePortsFile("http", http.boundAddress()); } TransportService transport = injector.getInstance(TransportService.class); writePortsFile("transport", transport.boundAddress()); } pluginsService.filterPlugins(ClusterPlugin.class).forEach(ClusterPlugin::onNodeStarted); return this; }

• 34.通过 injector 获取各个类的对象，调用 start() 方法启动（实际进入各个类的中 doStart 方法）: LifecycleComponent、IndicesService、IndicesClusterStateService、SnapshotsService、SnapshotShardsService、RoutingService、SearchService、MonitorService、NodeConnectionsService、ResourceWatcherService、GatewayService、Discovery、TransportService
• 35.IndicesService：索引管理
  IndicesClusterStateService：跨集群同步

SnapshotsService：负责创建快照
SnapshotShardsService：此服务在数据和主节点上运行，并控制这些节点上当前快照的分片。
RoutingService：侦听集群状态，当它收到集群改变事件将验证集群状态，路由表可能会更新
SearchService：搜索服务
MonitorService：监控
NodeConnectionsService：此组件负责在节点添加到群集状态后连接到节点，并在删除它们时断开连接。 此外，它会定期检查所有连接是否仍处于打开状态，并在需要时还原它们。 请注意，如果节点断开/不响应ping，则此组件不负责从群集中删除节点。 这是由NodesFaultDetection完成的。 主故障检测由链接MasterFaultDetection完成。
ResourceWatcherService：通用资源观察器服务
GatewayService：网关

• 36.集群发现与监控等，启动 HttpServerTransport， 绑定服务端口。

线程池

 public ThreadPool(final Settings settings, final ExecutorBuilder<?>... customBuilders) { super(settings); assert Node.NODE_NAME_SETTING.exists(settings); final Map<String, ExecutorBuilder> builders = new HashMap<>(); final int availableProcessors = EsExecutors.numberOfProcessors(settings); final int halfProcMaxAt5 = halfNumberOfProcessorsMaxFive(availableProcessors); final int halfProcMaxAt10 = halfNumberOfProcessorsMaxTen(availableProcessors); final int genericThreadPoolMax = boundedBy(4 * availableProcessors, 128, 512); builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30))); builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200)); builders.put(Names.BULK, new FixedExecutorBuilder(settings, Names.BULK, availableProcessors, 200)); builders.put(Names.GET, new FixedExecutorBuilder(settings, Names.GET, availableProcessors, 1000)); builders.put(Names.SEARCH, new AutoQueueAdjustingExecutorBuilder(settings, Names.SEARCH, searchThreadPoolSize(availableProcessors), 1000, 1000, 1000, 2000)); builders.put(Names.MANAGEMENT, new ScalingExecutorBuilder(Names.MANAGEMENT, 1, 5, TimeValue.timeValueMinutes(5))); builders.put(Names.LISTENER, new FixedExecutorBuilder(settings, Names.LISTENER, halfProcMaxAt10, -1)); builders.put(Names.FLUSH, new ScalingExecutorBuilder(Names.FLUSH, 1, halfProcMaxAt5, TimeValue.timeValueMinutes(5))); builders.put(Names.REFRESH, new ScalingExecutorBuilder(Names.REFRESH, 1, halfProcMaxAt10, TimeValue.timeValueMinutes(5))); builders.put(Names.WARMER, new ScalingExecutorBuilder(Names.WARMER, 1, halfProcMaxAt5, TimeValue.timeValueMinutes(5))); builders.put(Names.SNAPSHOT, new ScalingExecutorBuilder(Names.SNAPSHOT, 1, halfProcMaxAt5, TimeValue.timeValueMinutes(5))); builders.put(Names.FETCH_SHARD_STARTED, new ScalingExecutorBuilder(Names.FETCH_SHARD_STARTED, 1, 2 * availableProcessors, TimeValue.timeValueMinutes(5))); builders.put(Names.FORCE_MERGE, new FixedExecutorBuilder(settings, Names.FORCE_MERGE, 1, -1)); builders.put(Names.FETCH_SHARD_STORE, new ScalingExecutorBuilder(Names.FETCH_SHARD_STORE, 1, 2 * availableProcessors, TimeValue.timeValueMinutes(5))); for (final ExecutorBuilder<?> builder : customBuilders) { if (builders.containsKey(builder.name())) { throw new IllegalArgumentException("builder with name [" + builder.name() + "] already exists"); } builders.put(builder.name(), builder); } this.builders = Collections.unmodifiableMap(builders); threadContext = new ThreadContext(settings); final Map<String, ExecutorHolder> executors = new HashMap<>(); for (@SuppressWarnings("unchecked") final Map.Entry<String, ExecutorBuilder> entry : builders.entrySet()) { final ExecutorBuilder.ExecutorSettings executorSettings = entry.getValue().getSettings(settings); final ExecutorHolder executorHolder = entry.getValue().build(executorSettings, threadContext); if (executors.containsKey(executorHolder.info.getName())) { } logger.debug("created thread pool: {}", entry.getValue().formatInfo(executorHolder.info)); executors.put(entry.getKey(), executorHolder); } executors.put(Names.SAME, new ExecutorHolder(DIRECT_EXECUTOR, new Info(Names.SAME, ThreadPoolType.DIRECT))); this.executors = unmodifiableMap(executors); this.scheduler = Scheduler.initScheduler(settings); TimeValue estimatedTimeInterval = ESTIMATED_TIME_INTERVAL_SETTING.get(settings); this.cachedTimeThread = new CachedTimeThread(EsExecutors.threadName( settings, "[timer]"), estimatedTimeInterval.millis()); this.cachedTimeThread.start(); }

线程池类型 ThreadPoolType

fixed（固定）：fixed线程池拥有固定数量的线程来处理请求，在没有空闲线程时请求将被挂在队列中。queue_size参数可以控制在没有空闲线程时，能排队挂起的请求数
fixed_auto_queue_size：此类型为实验性的，将被更改或删除，不关注
scaling（弹性）：scaling线程池拥有的线程数量是动态的，这个数字介于core和max参数的配置之间变化。keep_alive参数用来控制线程在线程池中空闲的最长时间
direct：此类线程是一种不支持关闭的线程,就意味着一旦使用,则会一直存活下去.

一些重要的线程池

generic：用于通用的请求（例如：后台节点发现），线程池类型为 scaling。
index：用于index/delete请求，线程池类型为 fixed， 大小的为处理器数量，队列大小为200，最大线程数为 1 + 处理器数量。
search：用于count/search/suggest请求。线程池类型为 fixed， 大小的为 int((处理器数量 3) / 2) +1，队列大小为1000。*
get：用于get请求。线程池类型为 fixed，大小的为处理器数量，队列大小为1000。
analyze：用于analyze请求。线程池类型为 fixed，大小的1，队列大小为16
write：用于单个文档的 index/delete/update 请求以及 bulk 请求，线程池类型为 fixed，大小的为处理器数量，队列大小为200，最大线程数为 1 + 处理器数量。
snapshot：用于snaphost/restore请求。线程池类型为 scaling，线程保持存活时间为5分钟，最大线程数为min(5, (处理器数量)/2)。
warmer：用于segment warm-up请求。线程池类型为 scaling，线程保持存活时间为5分钟，最大线程数为min(5, (处理器数量)/2)。
refresh：用于refresh请求。线程池类型为 scaling，线程空闲保持存活时间为5分钟，最大线程数为min(10, (处理器数量)/2)。
listener：主要用于Java客户端线程监听器被设置为true时执行动作。线程池类型为 scaling，最大线程数为min(10, (处理器数量)/2)。

PluginsService插件服务

 public static final String ES_PLUGIN_PROPERTIES = "plugin-descriptor.properties"; public static final String ES_PLUGIN_POLICY = "plugin-security.policy"; public static PluginInfo readFromProperties(final Path path) throws IOException { final Path descriptor = path.resolve(ES_PLUGIN_PROPERTIES); final Properties props = new Properties(); try (InputStream stream = Files.newInputStream(descriptor)) { props.load(stream); } final String name = props.getProperty("name"); if (name == null || name.isEmpty()) { throw new IllegalArgumentException( "property [name] is missing in [" + descriptor + "]"); } final String description = props.getProperty("description"); if (description == null) { throw new IllegalArgumentException( "property [description] is missing for plugin [" + name + "]"); } final String version = props.getProperty("version"); if (version == null) { throw new IllegalArgumentException( "property [version] is missing for plugin [" + name + "]"); } final String esVersionString = props.getProperty("elasticsearch.version"); if (esVersionString == null) { throw new IllegalArgumentException( "property [elasticsearch.version] is missing for plugin [" + name + "]"); } final Version esVersion = Version.fromString(esVersionString); if (esVersion.equals(Version.CURRENT) == false) { final String message = String.format( Locale.ROOT, "plugin [%s] is incompatible with version [%s]; was designed for version [%s]", name, Version.CURRENT.toString(), esVersionString); throw new IllegalArgumentException(message); } final String javaVersionString = props.getProperty("java.version"); if (javaVersionString == null) { throw new IllegalArgumentException( "property [java.version] is missing for plugin [" + name + "]"); } JarHell.checkVersionFormat(javaVersionString); JarHell.checkJavaVersion(name, javaVersionString); final String classname = props.getProperty("classname"); if (classname == null) { throw new IllegalArgumentException( "property [classname] is missing for plugin [" + name + "]"); } final String hasNativeControllerValue = props.getProperty("has.native.controller"); final boolean hasNativeController; if (hasNativeControllerValue == null) { hasNativeController = false; } else { switch (hasNativeControllerValue) { case "true": hasNativeController = true; break; case "false": hasNativeController = false; break; default: final String message = String.format( Locale.ROOT, "property [%s] must be [%s], [%s], or unspecified but was [%s]", "has_native_controller", "true", "false", hasNativeControllerValue); throw new IllegalArgumentException(message); } } final String requiresKeystoreValue = props.getProperty("requires.keystore", "false"); final boolean requiresKeystore; try { requiresKeystore = Booleans.parseBoolean(requiresKeystoreValue); } catch (IllegalArgumentException e) { throw new IllegalArgumentException("property [requires.keystore] must be [true] or [false]," + " but was [" + requiresKeystoreValue + "]", e); } return new PluginInfo(name, description, version, classname, hasNativeController, requiresKeystore); }

• 读取模块的配置文件 plugin-descriptor.properties，解析出内容并存储到 Map中。
• 分别校验 name, description, version, elasticsearch.version, java.version, classname, extended.plugins, has.native.controller, requires.keystore 这些配置项，缺失或者不按要求则抛出异常。
• 根据配置项构造一个 PluginInfo 对象返回。

参考文章

Elasticsearch 6.3.2 启动过程