Elasticsearch Transport 模块创建及启动分析

Elasticsearch 通信模块的分析从宏观上介绍了ES Transport模块总体功能，于是就很好奇ElasticSearch是怎么把服务启动起来，以接收Client发送过来的Index索引操作、GET获取文档操作 等一系列操作的呢？本文分析：ElasticSearch6.3.2 Netty Http Server 服务的启动过程。ES节点启动，就是启动各个服务，初始化各个服务代码实现 在 org.elasticsearch.node.Node的构造方法中，从创建 org.elasticsearch.common.network.NetworkModule 对象开始，NetworkModule 就是ES中所有关于网络通信相关的功能的创建与注册吧。

final NetworkModule networkModule = new NetworkModule(settings, false, pluginsService.filterPlugins(NetworkPlugin.class),
threadPool, bigArrays, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, xContentRegistry,
networkService, restController);

在创建NetworkModule对象时，主要是创建2个用于通信的Server

• 一个是Sever是用来接收用户发起的各种操作请求的(external REST clients)，比如GET、INDEX、BULK WRITE、DELETE...这个Server叫HttpServerTransport（具体实现是Netty4HttpServerTransport）。
• 另一个Server用于节点之间的通信(transport layer)，比如：节点之间相互发送ping命令、集群各个节点之间的信息交换、还有，当GET index/_doc/1 这样的用户操作发送到coordinator 节点上，当docid为1的文档不在本机节点上，那么就会使用TcpTransport（具体实现是Netty4TcpTransport）将命令转发到目标节点上

A client can either be retrieved from a org.elasticsearch.node.Node started, or connected remotely to one or more nodes using org.elasticsearch.client.transport.TransportClient. Every node in the cluster can handle HTTP and Transport traffic by default. The transport layer is used exclusively for communication between nodes and the Java TransportClient; the HTTP layer is used only by external REST clients.

Netty4HttpServerTransport 对象创建如下，Netty4TcpTransport 也是类似的逻辑。
org.elasticsearch.common.network.NetworkModule#NetworkModule

                Map<String, Supplier<HttpServerTransport>> httpTransportFactory = plugin.getHttpTransports(settings,threadPool,bigArrays,circuitBreakerService,namedWriteableRegistry, xContentRegistry, networkService, dispatcher);
                for (Map.Entry<String, Supplier<HttpServerTransport>> entry : httpTransportFactory.entrySet()) {
                    registerHttpTransport(entry.getKey(), entry.getValue());
                }

Netty4Plugin#getHttpTransports 创建 Netty Http Server：Netty4HttpServerTransport

    @Override
    public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,CircuitBreakerService，circuitBreakerService,NamedWriteableRegistry namedWriteableRegistry,NamedXContentRegistry xContentRegistry,NetworkService networkService,HttpServerTransport.Dispatcher dispatcher) {
        return Collections.singletonMap(NETTY_HTTP_TRANSPORT_NAME,
            () -> new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher));
    }

将构造好的 Transport 对象封装到 TransportService

//获取构造好的 Netty4Transport
final Transport transport = networkModule.getTransportSupplier().get();
//将 Netty4Transport 封装到 TransportService
final TransportService transportService = newTransportService(settings, transport, threadPool,
networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings(), taskHeaders);

然后其他需要使用通信功能的模块，只需要封装 TransportService 对象即可。比如执行用户SEARCH操作的搜索模块 TransportSearchAction，它有一个实例属性SearchTransportService，而SearchTransportService就封装了 TransportService，这样TransportSearchAction就能使用TcpTransport进行通信了。如下代码所示：
Node.java 构造方法：

//构造SearchTransportService对象时f需要TransportService，TransportService对象 是一个"公共连接对象"，许多服务都会用到它
final SearchTransportService searchTransportService =  new SearchTransportService(settings,transportService,SearchExecutionStatsCollector.makeWrapper(responseCollectorService));

这里额外提一句：各种Action对象所依赖的Service，应该都是在Node.java的构造方法里面创建的：比如TransportSearchAction依赖的SearchTransportService、ClusterService等都是在节点启动时创建的。

当Netty4HttpServerTransport创建完毕后，就需要绑定端口，启动服务。在org.elasticsearch.node.Node.start方法是ES节点中所有服务的启动入口(当然也包括Netty Http Server了)
org.elasticsearch.node.Node#start方法

        if (NetworkModule.HTTP_ENABLED.get(settings)) {
            injector.getInstance(HttpServerTransport.class).start();
        }

因为Netty4HttpServerTransport继承了AbstractLifecycleComponent，因此它的启动逻辑在org.elasticsearch.common.component.AbstractLifecycleComponent.start中实现，执行doStart()启动Netty Http Server，并绑定端口到9200
Netty4HttpServerTransport#doStart()

protected void doStart() {
        boolean success = false;
        try {
            this.serverOpenChannels = new Netty4OpenChannelsHandler(logger);//---> es for test

            serverBootstrap = new ServerBootstrap();//workerCount=8, elasticsearch[debug_node][http_server_worker]
            //channel一旦分配给EventLoopGroup里面的某个EventLoop线程后,该channel上的所有的事件都将由这个EventLoop线程处理
            serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
                HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
            serverBootstrap.channel(NioServerSocketChannel.class);//处理连接请求,每个连接建立后创建一个'child channel'处理该连接的所有IO事件
            //为child channel 绑定一个handler, 即用该handler处理该 channel 上的io event
            serverBootstrap.childHandler(configureServerChannelHandler());//--->Netty4HttpRequestHandler
            //指定 child channel 一些配置参数 (父channel是处理连接请求的channel, child channel是已建立的连接的事件处理通道)
            serverBootstrap.childOption(ChannelOption.TCP_NODELAY, SETTING_HTTP_TCP_NO_DELAY.get(settings));
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, SETTING_HTTP_TCP_KEEP_ALIVE.get(settings));
            //---> TCP 发送缓冲区大小
            final ByteSizeValue tcpSendBufferSize = SETTING_HTTP_TCP_SEND_BUFFER_SIZE.get(settings);
            if (tcpSendBufferSize.getBytes() > 0) {
                serverBootstrap.childOption(ChannelOption.SO_SNDBUF, Math.toIntExact(tcpSendBufferSize.getBytes()));
            }
            //---> TCP 接收缓冲区大小
            final ByteSizeValue tcpReceiveBufferSize = SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE.get(settings);
            if (tcpReceiveBufferSize.getBytes() > 0) {
                serverBootstrap.childOption(ChannelOption.SO_RCVBUF, Math.toIntExact(tcpReceiveBufferSize.getBytes()));
            }

            serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator);
            serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator);

            final boolean reuseAddress = SETTING_HTTP_TCP_REUSE_ADDRESS.get(settings);
            serverBootstrap.option(ChannelOption.SO_REUSEADDR, reuseAddress);
            serverBootstrap.childOption(ChannelOption.SO_REUSEADDR, reuseAddress);

            this.boundAddress = createBoundHttpAddress();//--->ServerBootStrap绑定端口
            if (logger.isInfoEnabled()) {
                logger.info("{}", boundAddress);
            }
            success = true;
        } finally {
            if (success == false) {
                doStop(); // otherwise we leak threads since we never moved to started
            }
        }
    }

Netty Http Server的worker线程数量是：节点所在的机器上的可用CPU核数：(Runtime.getRuntime().availableProcessors()*2)
其他的一些默认配置如下：
TCP_NODELAY=true, SO_KEEPALIVE=true

ServerBootstrap(ServerBootstrapConfig(group: NioEventLoopGroup, channelFactory: NioServerSocketChannel.class, options: {RCVBUF_ALLOCATOR=io.netty.channel.FixedRecvByteBufAllocator@72ce8a9b, SO_REUSEADDR=true}, childGroup: NioEventLoopGroup, childOptions: {TCP_NODELAY=true, SO_KEEPALIVE=true, RCVBUF_ALLOCATOR=io.netty.channel.FixedRecvByteBufAllocator@72ce8a9b, SO_REUSEADDR=true}, childHandler: org.elasticsearch.http.netty4.Netty4HttpServerTransport$HttpChannelHandler@56ec6ac0))

ES Server 接收用户请求(GET/WRITE/DELETE...)的起始处理点 在哪里？

由于ES Server(实在找不到其他更好的名字来描述了...)是基于 Netty的，那肯定有个ChannelHandler负责处理发生在SocketChannel上的事件。而这个ChannelHandler就是：org.elasticsearch.http.netty4.Netty4HttpRequestHandler
org.elasticsearch.http.netty4.Netty4HttpServerTransport.HttpChannelHandler#initChannel 方法中注册了Netty4HttpRequestHandler，因此用户请求就交给Netty4HttpRequestHandler来处理了。

            ch.pipeline().addLast("handler", requestHandler);//Netty4HttpRequestHandler 业务逻辑处理

那根据Netty框架，毫无疑问 接收用户请求的起始处理点在 org.elasticsearch.http.netty4.Netty4HttpRequestHandler#channelRead0 方法里面了。
因此，如果想debug一下INDEX操作、GET操作、DELETE操作的入口，在入口点： org.elasticsearch.http.netty4.Netty4HttpRequestHandler#channelRead0 打上debug断点，在返回处：org.elasticsearch.http.netty4.Netty4HttpChannel#sendResponse 打上debug断点，根据IDEA的 dubuger frames 栈追踪 查看各个操作的执行路径。

既然所有的用户操作都是统一的入口，那么又是如何解析这些操作，并最终传递给合适的 TransportXXXAction 来处理的呢？其大概步骤如下：

• 1，ES每个操作(JAVA API/rest api)都有对应的Action类，比如：DELETE APID的Action类是：RestDeleteAction；GET API 的Action类是：RestGetAction。
• 2，每个Action类都重写了父类的org.elasticsearch.rest.BaseRestHandler#prepareRequest方法，构造出相应的Action对象，并在方法中返回一个lambda表达式，代表需要执行该操作。接下来，该操作在 BaseRestHandler#handleRequest 方法中的 action.accept(channel)语句触发执行。
• 3，触发执行后，这些Action操作由 NodeClient#doExecute 方法发送到相应的节点上执行：先获得 执行Action操作所对应的 TransportXXXAction类，再通过 execute(request,listener) 执行，代码如下：

      return transportAction(action).execute(request, listener)
  

    TransportAction#execute(Request, org.elasticsearch.action.ActionListener<Response>)是执行各种Action操作的统一入口，最终在在：TransportAction.RequestFilterChain#proceed 中`this.action.doExecute(task, request, listener);`调用每个实现类TransportXXXAction#doExecute()执行对应的操作！
  

比如说：GET操作由：TransportSingleShardAction#doExecute处理；DELETE操作由：TransportBulkAction#doExecute(Task,BulkRequest, ActionListener)处理。

• 4，继续深入分析DELETE操作。TransportBulkAction#doExecute 调用 org.elasticsearch.action.bulk.TransportBulkAction#executeBulk启动一个新任务：BulkOperation。由于DELETE操作是与分片相关的操作，即需要从分片上删除数据，因此在org.elasticsearch.action.bulk.TransportBulkAction.BulkOperation#doRun 方法中判断该操作是一个DELETE类型的操作，并执行：shardBulkAction.execute(bulkShardRequest, new ActionListener<BulkShardResponse>(){...});将删除操作提交给"分片处理Action"---TransportShardBulkAction执行。

• 5，TransportShardBulkAction继承自TransportAction，execute当然还是走“相同的”逻辑到这个方法里面：TransportAction#execute(Task,Request,ActionListener)，再到processed()方法里面this.action.doExecute(task, request, listener);，这时就是调用：TransportShardBulkAction的doExecute方法了。而TransportShardBulkAction的doExecute()方法是继承自TransportReplicationAction，可以看到在这里面执行的是ReroutePhase任务，这也很好理解，因为删除一篇文档，需要知道这篇文档在哪个分片上，需要把删除请求发送到这个分片上去，这也是为什么需要ReroutePhase的原因吧：

      protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
      new ReroutePhase((ReplicationTask) task, request, listener).run();
  }
  

• 6，跟踪到ReroutePhase的doRun()方法里面看：删除操作在本机节点上执行performLocalAction，删除操作在其他远程节点上执行：performRemoteAction。这里，又通过TransportService#sendRequest 方法把请求发送出去了。。。烦，那我就继续跟踪，看看你翻跟斗到哪里去了……

  if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
              performLocalAction(state, primary, node, indexMetaData);
          } else {
              performRemoteAction(state, primary, node);
          }
  

• 7，那跟斗到底翻到哪里去了呢？其实这个也很好判断，这是一个DELETE操作，它所对应的Action执行是TransportReplicationAction，而且DELETE操作肯定是要走primary shard的，结果在TransportReplicationAction的内部类PrimaryOperationTransportHandler里面发现了接收方法：PrimaryOperationTransportHandler#messageReceived(ConcreteShardRequest,TransportChannel,Task)，里面创建AsyncPrimaryAction任务，在TransportReplicationAction.AsyncPrimaryAction#doRun里面，才是真正地开始在分片上获取访问锁，并删除文档。

• 8，AsyncPrimaryAction#doRun成功获取到锁（PrimaryShardReference）后，回调：AsyncPrimaryAction#onResponse，在createReplicatedOperation(...).execute()触发底层Lucene删除逻辑。

删除的时候，有相应的删除策略，具体实现在：org.elasticsearch.index.engine.InternalEngine#planDeletionAsPrimary

if (versionValue == null) {
            currentVersion = Versions.NOT_FOUND;
            currentlyDeleted = true;
        } else {
            currentVersion = versionValue.version;
            currentlyDeleted = versionValue.isDelete();
        }
        final DeletionStrategy plan;
        if (delete.versionType().isVersionConflictForWrites(currentVersion, delete.version(), currentlyDeleted)) {
            final VersionConflictEngineException e = new VersionConflictEngineException(shardId, delete, currentVersion, currentlyDeleted);
            plan = DeletionStrategy.skipDueToVersionConflict(e, currentVersion, currentlyDeleted);
        } else {
            plan = DeletionStrategy.processNormally(
                    currentlyDeleted,
                    generateSeqNoForOperation(delete),
                    delete.versionType().updateVersion(currentVersion, delete.version()));
        }
        return plan;

删除doc的时候，还要判断docid在不在，具体实现在：org.elasticsearch.index.engine.InternalEngine#loadCurrentVersionFromIndex

private long loadCurrentVersionFromIndex(Term uid) throws IOException {
        assert incrementIndexVersionLookup();
        try (Searcher searcher = acquireSearcher("load_version", SearcherScope.INTERNAL)) {
            return VersionsAndSeqNoResolver.loadVersion(searcher.reader(), uid);
        }
    }

另外在看源码的时候发现，delete-by-doc-id 是不会触发 段合并的。所以，delete by id 这种方式的删除是很快的且对集群负载影响很小：

    // NOTE: we don't throttle this when merges fall behind because delete-by-id does not create new segments:

最终在：org.elasticsearch.index.engine.InternalEngine#delete 方法里面进行Lucene层面上的文档删除：

if (delete.origin() == Operation.Origin.PRIMARY) {
                plan = planDeletionAsPrimary(delete);
            } else {
                plan = planDeletionAsNonPrimary(delete);
            }

            if (plan.earlyResultOnPreflightError.isPresent()) {
                deleteResult = plan.earlyResultOnPreflightError.get();
            } else if (plan.deleteFromLucene) {
                deleteResult = deleteInLucene(delete, plan);
            } else {
                deleteResult = new DeleteResult(
                        plan.versionOfDeletion, plan.seqNoOfDeletion, plan.currentlyDeleted == false);
            }

具体实现在：org.elasticsearch.index.engine.InternalEngine#deleteInLucene里面，代码就不贴了。以上，就是一个完整的 ES delete by doc id 的执行流程。感兴趣的可以再细究。

这篇文章最后，详细介绍了DELET API的执行路径，其他操作也是类似的，按这个分析即可。
原文：https://www.cnblogs.com/hapjin/p/11018479.html