es 读流程源码解析-Toy模板网

这篇具有很好参考价值的文章主要介绍了es 读流程源码解析。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

本文源码基于es6.8.0版本

es 读流程源码解析

search 分为两部分，query + fetch

节点角色划分

协调节点负责接收请求，然后构造查询分发给其他的数据节点，然后从各个分片上获取数据。数据最终汇聚到协调节点，然后再讲结果做合并。然后返回查询结果。

而数据节点，则只负责将自己的分片上的数据做一次查询。然后把数据发给协调节点。

1.请求到协调节点，解析请求

Rest层用于解析Http请求参数，RestRequest解析并转化为SearchRequest，然后再对SearchRequest做处理，这块的逻辑在RestSearchAction.prepareRequest(final RestRequest request, final NodeClient client)

2.转换请求，从Rest层到Transport层

NodeClient在处理SearchRequest请求时，会将请求的action转化为对应Transport层的action，然后再由Transport层的action来处理SearchRequest

3.协调节点在TransportAction

TransportAction#execute(Request request, ActionListener listener) -> TransportAction#execute(Task task, Request request, ActionListener listener) -> TransportAction#proceed(Task task, String actionName, Request request, ActionListener listener)。TransportAction会调用一个请求过滤链来处理请求，如果相关的插件定义了对该action的过滤处理，则先会执行插件的处理逻辑，然后再进入TransportAction的处理逻辑

4.构造查询的分片列表

TransportSearchAction.doExecute()方法里边，会获取到远程集群和本地集群对应的索引列表

TransportSearchAction.executeSearch()构造查询的分片列表，设置默认的查询策略QUERY_THEN_FETCH，设置默认的缓存策略，不开启缓存。setMaxConcurrentShardRequests() 方法设置最大的并发数最大并发分片数，最大是256：Math.min(256, Math.max(nodeCount, 1)* IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getDefault(Settings.EMPTY))。

翻译一下上边的公式（Math.min(256, Math.max(数据节点的个数, 1)* 索引设置的分片数）

最后再调用searchAsyncAction()，去执行真正的查询操作。执行查询用到的线程池是：search

接着又调用了AbstractSearchAsyncAction的start方法。

5.执行query的操作，分发请求到各个分片上

TransportSearchAction.executeSearch() 方法里边调用了searchAsyncAction()方法! 接着再看searchAsyncAction()，它是真正执行查询操作的方法。这个方法里边，先判断是否执行过滤，如果不过滤，去判断是用那种方式来执行查询（默认QUERY_THEN_FETCH）。 searchAsyncAction()方法会根据查询类型，来返回AbstractSearchAsyncAction的实现类，如果是QUERY_THEN_FETCH就返回SearchQueryThenFetchAsyncAction对象，如果是DFS_QUERY_THEN_FETCH则返回SearchDfsQueryThenFetchAsyncAction对象。这个对象还在TransportSearchAction.executeSearch() 方法里边，并且最终调用了该对象的start()方法。注意这里边的调用关系：AbstractSearchAsyncAction类的start()方法调用了executePhase()方法，在executePhase()方法中又调用了SearchPhase类的 run()方法。而SearchPhase是一个抽象类。

接着再看继承了SearchPhase的类，只有以下截图中的四个。我们重点关注的是 InitialSearchPhase类，

es 读流程源码解析

InitialSearchPhase.run() 在该方法中，并发去获取每个分片上的数据（并发度为 Math.min(设置的最大并发数maxConcurrentShardRequests，分片数)）。调用 performPhaseOnShard()方法，去每个分片上取数据。等待所有的并发的线程把所有的分片上的数据都取到，才结束。执行过程中，如果某个分片失败了，也会记录下来。 shardsIts是本次查询涉及的所有分片，shardRoutings.nextOrNull()从某个分片中主或者所有副本中选一个。在该方法中，shardsIts是本次查询涉及的所有分片，shardRoutings.nextOrNull()从某个分片中主或者所有副本中选一个。InitialSearchPhase.run() 中最后去调用了performPhaseOnShard()方法，在performPhaseOnShard()方法里边调用了executePhaseOnShard()方法。还是在InitialSearchPhase类中的performPhaseOnShard()方法里边，executePhaseOnShard()方法是一个抽象方法，定义在InitialSearchPhase类中，它的实现类有以下截图上的几个

es 读流程源码解析

其中红色框是我们重点关注的，这个是和上边的TransportAction.executeSearch() 方法里边的searchAsyncAction()方法里边的searchAsyncAction()方法就照应起来了。如果是QUETY_THEN_FETCH搜索模式，则继续走的是SearchQueryThenFetchAsyncAction类的executePhaseOnShard()方法，在该方法中，调用了SearchTransportService类的sendExecuteQuery()方法，在sendExecuteQuery()方法中，继续调用了transportService的sendChildRequest()方法向具体的分片发送Query阶段的子任务进行异步处理（如果想看是如何发送请求给分片，可以接着看！）。

到这里协调及节点的查询算是结束了，剩下的就是等各个分片将执行的结果返回了。（ TODO ）然后剩下的查询流程是数据节点上的分片具体的处理协调协调节点分配的查询任务了。

这里再说一下InitialSearchPhase.run()，它对一次一次的分发请求给所有的分片，即使是同一个节点的，也是分发的！

6.关于协调节点分发查询请求后结果的汇聚的调用链路

在上边提到的InitialSearchPhase.run()中调用performPhaseOnShard()方法，performPhaseOnShard()方法调用了executePhaseOnShard()方法，在这个方法中，创建了一个SearchActionListener监听器，用来回调分片的查询结果。然后调用InitialSearchPhase.onShardResult()，将结果合并到一起。前边还提到了，一个查询任务，可能是有多个线程并发执行的，所以在创建的SearchActionListener的时候，调用了InitialSearchPhase.maybeFork()方法，来合并多个线程的任务。InitialSearchPhase.onShardResult()，做的操作是把全部线程执行的结果都合并在协调节点。InitialSearchPhase.onShardResult()方法做了两件事，第一件事是合并结果，调用了onShardSuccess()方法，第二件事是调用successfulShardExecution()方法，来检查是否所有的分片都结束了。如果结束则触发进行下一个阶段的，在successfulShardExecution()方法中，如果都结束了则调用onPhaseDone()方法中的executeNextPhase()来进入 fetch阶段。

7.在协调节点，query阶段，数据汇聚的细节（此时汇聚的只有文档id，真正的数据的汇聚是在fetch阶段进行的）

针对上述的结果合并，onShardSuccess()方法中调用results.consumeResult()方法，多做了一部检查，检查分片的数据是否重复，然后再调用consumeInternal()方法，关于在协调节点，各个分片结果合并的操作都在consumeInternal()方法里边。其中包括了agg聚类结果的合并，和search出来的TopDocs的合并。agg聚类结果的合并调用consumeInternal()中的InternalAggregations.reduce(Arrays.asList(aggsBuffer), reduceContext)，search出来的topDoc是通过mergeTopDocs(Arrays.asList(topDocsBuffer), querySearchResult.from() + querySearchResult.size(), 0)来合并的。