【ES源码分析】Transport模块的初始化

【ES源码分析】Transport模块的初始化

文章目录

Transport模块的初始化

ActionModule的初始化NetworkModule的初始化

NetworkPluginNetty4TransportNetty4HttpServerTransport TransportService的初始化

Transport模块的初始化

源码基于6.7.2

传输模块的初始化主要的在节点启动时的构造函数中完成的。

节点启动时，主要在构建函数中，进行通信模块的初始化。

    protected Node(
            final Environment environment, Collection> classpathPlugins, boolean forbidPrivateIndexSettings) {
        logger = LogManager.getLogger(Node.class);
        final List resourcesToClose = new ArrayList<>(); // register everything we need to release in the case of an error
        boolean success = false;
        try {
            ...
            ...
            ...
            // 过滤出ActionPlugin插件列表作为参数传入 ActionModule构造函数,在ActionModule中会对TCP和HTTP的请求和处理类进行注册和绑定。
            ActionModule actionModule = new ActionModule(false, settings, clusterModule.getIndexNameexpressionResolver(),    settingsModule.getIndexScopedSettings(), settingsModule.getClusterSettings(), settingsModule.getSettingsFilter(), threadPool, pluginsService.filterPlugins(ActionPlugin.class), client, circuitBreakerService, usageService);
            ...

            final RestController restController = actionModule.getRestController();
            // 过滤出NetworkPlugin插件列表作为参数传入 NetworkModule构造函数
            final NetworkModule networkModule = new NetworkModule(settings, false, pluginsService.filterPlugins(NetworkPlugin.class),
                threadPool, bigArrays, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, xContentRegistry,
                networkService, restController);
            ...
            ...
            ...  
            // 通过网络模块获取已经初始化的Transport
            final Transport transport = networkModule.getTransportSupplier().get();
            Set taskHeaders = Stream.concat(
                pluginsService.filterPlugins(ActionPlugin.class).stream().flatMap(p -> p.getTaskHeaders().stream()),
                Stream.of(Task.X_OPAQUE_ID)
            ).collect(Collectors.toSet());
            // 基于网络模块的Transport构建TransportService
            final TransportService transportService = newTransportService(settings, transport, threadPool,
                networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings(), taskHeaders);
            final ResponseCollectorService responseCollectorService = new ResponseCollectorService(clusterService);
            // 基于TransportService构建searchTransportService服务
            final SearchTransportService searchTransportService =  new SearchTransportService(transportService,
                SearchExecutionStatsCollector.makeWrapper(responseCollectorService));
            final Consumer httpBind;
            final HttpServerTransport httpServerTransport;
            // 通过网络模块获取已经初始化的Transport，HTTP还可以关闭？
            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;
            }
            ...
            ...
            ...    
            if (NetworkModule.HTTP_ENABLED.get(settings)) {
                logger.debug("initializing HTTP handlers ...");
                // 初始化REST的请求和处理类的映射
                actionModule.initRestHandlers(() -> clusterService.state().nodes());
            }
            logger.info("initialized");

            success = true;
            ...
            ...
            ...  
    }

ActionModule的初始化

ActionModule的内部初始化是通过插件的方式加载的，主要完成注册Action与处理类的映射和RestController的创建。

    public ActionModule(boolean transportClient, Settings settings, IndexNameexpressionResolver indexNameexpressionResolver,
                        IndexScopedSettings indexScopedSettings, ClusterSettings clusterSettings, SettingsFilter settingsFilter,
                        ThreadPool threadPool, List actionPlugins, NodeClient nodeClient,
            CircuitBreakerService circuitBreakerService, UsageService usageService) {
        this.transportClient = transportClient;
        this.settings = settings;
        this.indexNameexpressionResolver = indexNameexpressionResolver;
        this.indexScopedSettings = indexScopedSettings;
        this.clusterSettings = clusterSettings;
        this.settingsFilter = settingsFilter;
        this.actionPlugins = actionPlugins;
        // 进行Action设置，action注册和对应handler的映射绑定
        actions = setupActions(actionPlugins);
        // 对action过滤器进行配置
        actionFilters = setupActionFilters(actionPlugins);
        autoCreateIndex = transportClient ? null : new AutoCreateIndex(settings, clusterSettings, indexNameexpressionResolver);
        destructiveOperations = new DestructiveOperations(settings, clusterSettings);
        Set headers = Stream.concat(
            actionPlugins.stream().flatMap(p -> p.getRestHeaders().stream()),
            Stream.of(Task.X_OPAQUE_ID)
        ).collect(Collectors.toSet());
        // rest包装器
        UnaryOperator restWrapper = null;
        for (ActionPlugin plugin : actionPlugins) {
            UnaryOperator 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;
            }
        }
        mappingRequestValidators = new TransportPutMappingAction.RequestValidators(
            actionPlugins.stream().flatMap(p -> p.mappingRequestValidators().stream()).collect(Collectors.toList())
        );

        if (transportClient) {
            restController = null;
        } else {
            // 构建出RestController对象
            restController = new RestController(headers, restWrapper, nodeClient, circuitBreakerService, usageService);
        }
    }

NetworkModule的初始化

构造NetworkModule对象，在执行构造函数的时候，通过插件的方式加载这3个成员对象。

主要数据成员对象

Map> transportFactories
Map> transportHttpFactories
List transportIntercetors

Transport ：负责内部节点的RPC请求

HttpServerTransport：负责客户端的REST服务

TransportInterceptor：传输层拦截器

    public NetworkModule(Settings settings, boolean transportClient, List plugins, ThreadPool threadPool,
                         BigArrays bigArrays,
                         PageCacheRecycler pageCacheRecycler,
                         CircuitBreakerService circuitBreakerService,
                         NamedWriteableRegistry namedWriteableRegistry,
                         NamedXContentRegistry xContentRegistry,
                         NetworkService networkService, HttpServerTransport.Dispatcher dispatcher) {
        this.settings = settings;
        this.transportClient = transportClient;
        // 遍历插件，分别注册HttpTransport,Transport,TransportInterceptor
        for (NetworkPlugin plugin : plugins) {
            if (transportClient == false && HTTP_ENABLED.get(settings)) {
                Map> httpTransportFactory = plugin.getHttpTransports(settings, threadPool, bigArrays,
                    circuitBreakerService, namedWriteableRegistry, xContentRegistry, networkService, dispatcher);
                for (Map.Entry> entry : httpTransportFactory.entrySet()) {
                    // 其实检查不能使用TransportClient创建HTTP的通信传输，不能有同名的HttpServerTransport
                    registerHttpTransport(entry.getKey(), entry.getValue());
                }
            }
            Map> transportFactory = plugin.getTransports(settings, threadPool, pageCacheRecycler,
                circuitBreakerService, namedWriteableRegistry, networkService);
            for (Map.Entry> entry : transportFactory.entrySet()) {
                // 检查不有同名的Transport
                registerTransport(entry.getKey(), entry.getValue());
            }
            List transportInterceptors = plugin.getTransportInterceptors(namedWriteableRegistry,
                threadPool.getThreadContext());
            for (TransportInterceptor interceptor : transportInterceptors) {
                // 注册传输层拦截器
                registerTransportInterceptor(interceptor);
            }
        }
    }

在构造完成时候，通过getTransportSupplier，getHttpServerTransportSupplier，getTransportInterceptor对外提供服务。

NetworkPlugin

NetworkPlugin是一个接口，Netty4Plugin实现了它，同时继承了Plugin。

在Netty4Plugin中实现了NetworkPlugin的getTransports和getHttpTransports方法，分别构建了Netty4Transport和Netty4HttpServerTransport用于Transport（TCP）传输和HTTP传输。

Netty4Transport

通过类图，可以发现Netty4Transport继承了TcpTransport，TcpTransport实现了Transport接口，这应该是在实现传输层，控制数据的在传输层的交互。Netty4Transport中实现了doStart的抽象方法，用来启动TCP服务。在启动的时候，默认情况下，同时构建了Client端和Server端。主要借助netty4框架来实现这些功能。

    @Override
    protected void doStart() {
        boolean success = false;
        try {
            ThreadFactory threadFactory = daemonThreadFactory(settings, TRANSPORT_WORKER_THREAD_NAME_PREFIX);
            eventLoopGroup = new NioEventLoopGroup(workerCount, threadFactory);
            // 初始化客户端
            clientBootstrap = createClientBootstrap(eventLoopGroup);
            // 默认是开启服务端配置的，初始化Server端
            if (NetworkService.NETWORK_SERVER.get(settings)) {
                for (ProfileSettings profileSettings : profileSettings) {
                    createServerBootstrap(profileSettings, eventLoopGroup);
                    bindServer(profileSettings);
                }
            }
            super.doStart();
            success = true;
        } finally {
            if (success == false) {
                doStop();
            }
        }
    }

Netty4HttpServerTransport

根据类图发现 Netty4HttpServerTransport 同时继承了AbstractLifecycleComponent和实现了HttpServerTransport。

同样实现了doStart的抽象方法，用来启动HTTP Server服务。在HTTP Server服务中配置了监听端口和处理器。这里其实应该是通过Netty4来完成HTTP协议下的传输层的部分。

@Override
    protected void doStart() {
       ...
            // 配置了请求的处理类HttpChannelHandler
            serverBootstrap.childHandler(configureServerChannelHandler());
       ...
           // 绑定端口作为HTTP监听端口
            this.boundAddress = createBoundHttpAddress();
       ...
    }

configureServerChannelHandler方法中构建了一个HttpChannelHandler对象，HttpChannelHandler的构造函数中有两个成员变量Netty4HttpServerTransport和Netty4HttpRequestHandler。当收到请求的时候，会调用dispatchRequest对不同的请求执行相应的处理。dispatchRequest是接口HttpServerTransport类的内部接口Dispatcher中的方法。主要用来转发请求。它的主要实现类是RestController。

TransportService的初始化

在Node的启动过程中会在初始化Transport之后，基于Transport构建TransportService。newTransportService方法中其实是调用TransportService的构造函数。

    public TransportService(Settings settings, Transport transport, ThreadPool threadPool, TransportInterceptor transportInterceptor, Function localNodeFactory, @Nullable ClusterSettings clusterSettings, Set taskHeaders) {
        this(settings, transport, threadPool, transportInterceptor, localNodeFactory, clusterSettings, taskHeaders,
             // 创建连接管理器
            new ConnectionManager(settings, transport));
    }

ConnectionManager是Node传输连接的管理类，通过这个类的connectionManager.connectToNode(node, connectionProfile, connectionValidator(node));解析连接配置文件创建内部连接。在ConnectionManager的构造函数中调用ConnectionProfile.buildDefaultConnectionProfile(settings)，这个方法会根据配置文件提供的信息创建连接。在TransportRequestOptions.Type中发现连接的分类。

public class TransportRequestOptions {
  ...
  ...
  ...
  public enum Type {
        RECOVERY, // 用于恢复
        BULK,     // 用于批量写入
        REG,      // 不是很清楚，有一个是集群注册
        STATE,    // 传输集群的状态
        PING      // ping请求
    }
    
}

在TransportSettings类中发现各类TCP连接的连接数，默认合计13个。

public final class TransportSettings {
        ...
        ...
        ...        
    public static final Setting CONNECTIONS_PER_NODE_RECOVERY =
        intSetting("transport.connections_per_node.recovery", 2, 1, Setting.Property.NodeScope);
    public static final Setting CONNECTIONS_PER_NODE_BULK =
        intSetting("transport.connections_per_node.bulk", 3, 1, Setting.Property.NodeScope);
    public static final Setting CONNECTIONS_PER_NODE_REG =
        intSetting("transport.connections_per_node.reg", 6, 1, Setting.Property.NodeScope);
    public static final Setting CONNECTIONS_PER_NODE_STATE =
        intSetting("transport.connections_per_node.state", 1, 1, Setting.Property.NodeScope);
    public static final Setting CONNECTIONS_PER_NODE_PING =
        intSetting("transport.connections_per_node.ping", 1, 1, Setting.Property.NodeScope);
        ...
        ...
        ...
    }

public TransportService(Settings settings, Transport transport, ThreadPool threadPool, TransportInterceptor transportInterceptor,
                        Function localNodeFactory, @Nullable ClusterSettings clusterSettings,
                        Set taskHeaders, ConnectionManager connectionManager) {
    // The only time we do not want to validate node connections is when this is a transport client using the simple node sampler
    this.validateConnections = TransportClient.CLIENT_TYPE.equals(settings.get(Client.CLIENT_TYPE_SETTING_S.getKey())) == false ||
        TransportClient.CLIENT_TRANSPORT_SNIFF.get(settings);
    this.transport = transport;
    this.threadPool = threadPool;
    this.localNodeFactory = localNodeFactory;
    this.connectionManager = connectionManager;
    this.clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
    setTracerLogInclude(TransportSettings.TRACE_LOG_INCLUDE_SETTING.get(settings));
    setTracerLogExclude(TransportSettings.TRACE_LOG_EXCLUDE_SETTING.get(settings));
    tracerLog = Loggers.getLogger(logger, ".tracer");
    // 任务管理器服务，用于跟踪节点上当前正在运行的任务
    taskManager = createTaskManager(settings, threadPool, taskHeaders);
    this.interceptor = transportInterceptor;
    // 异步发送者-发送集群节点间的通信请求
    this.asyncSender = interceptor.interceptSender(this::sendRequestInternal);
    // 远程集群--CCR
    this.connectToRemoteCluster = RemoteClusterService.ENABLE_REMOTE_CLUSTERS.get(settings);
    remoteClusterService = new RemoteClusterService(settings, this);
    responseHandlers = transport.getResponseHandlers();
    if (clusterSettings != null) {
        clusterSettings.addSettingsUpdateConsumer(TransportSettings.TRACE_LOG_INCLUDE_SETTING, this::setTracerLogInclude);
        clusterSettings.addSettingsUpdateConsumer(TransportSettings.TRACE_LOG_EXCLUDE_SETTING, this::setTracerLogExclude);
        if (connectToRemoteCluster) {
            remoteClusterService.listenForUpdates(clusterSettings);
        }
    }
    // 注册心跳Action与handler的映射
    registerRequestHandler(
        HANDSHAKE_ACTION_NAME,
        () -> HandshakeRequest.INSTANCE,
        ThreadPool.Names.SAME,
        false, false,
        (request, channel) -> channel.sendResponse(
            new HandshakeResponse(localNode, clusterName, localNode.getVersion())));
}

伴随着在Node初始化的最后会调用，进行RestHandler的注册映射。通信模块的初始化部分就结束了。