andler2、往数据节点发送查询请求的action(TransportSearchAction)3、通过transportService把查询请求发送到指定的数据节点

二、数据节点收到请求的处理逻辑1、尝试从缓存中加载查询结果2、不通过缓存查询直接执行查询(1)executeQuery和executeRank两种查询方式(2)、根据搜索上下文在查询之前添加各种查询搜集器(3)

这里只是HTTP发送查询请求到主节点主节点再转发到数据节点数据节点再到调用lucene.search实际查询数据之前如何处理的源码逻辑

在ActionModule.java

RestSearchAction(restController.getSearchUsageHolder()));actions.register(SearchAction.INSTANCE,

下面这个类RestSearchAction有长该省略的方法我都已经省略了首先通过routes请求到这个方法的prepareRequest(主要是组装searchRequest)这个方法内部会调用parseSearchSource(主要是组装searchSourceBuilder)

public

/{index}/_search),Route.builder(GET,

/{index}/{type}/_search).deprecated(TYPES_DEPRECATION_MESSAGE,

RestApiVersion.V_7).build(),Route.builder(POST,

/{index}/{type}/_search).deprecated(TYPES_DEPRECATION_MESSAGE,

RestApiVersion.V_7).build());}//入口

Overridepublic

(request.hasParam(min_compatible_shard_node))

{searchRequest

SearchRequest(Version.fromString(request.param(min_compatible_shard_node)));}

else

null。

*///组装SearchRequestIntConsumer

setSize

searchRequest.source().size(size);request.withContentOrSourceParamParserOrNull(parser

parseSearchRequest(searchRequest,

request,

client.getNamedWriteableRegistry(),

setSize,

searchUsageHolder));//请求发送return

channel

RestCancellableNodeClient(client,

request.getHttpChannel());cancelClient.execute(SearchAction.INSTANCE,

searchRequest,

RestChunkedToXContentListener(channel));};}//组装searchRequestpublic

static

parseSearchRequest(SearchRequest

request,Nullable

requestContentParser,NamedWriteableRegistry

namedWriteableRegistry,IntConsumer

setSize,Nullable

{request.param(type);deprecationLogger.compatibleCritical(search_with_types,

TYPES_DEPRECATION_MESSAGE);}//如果搜索请求的源为空创建一个新的

SearchSourceBuilder

SearchSourceBuilder());}//将请求中的索引参数解析为一个索引数组并设置到搜索请求中。

searchRequest.indices(Strings.splitStringByCommaToArray(request.param(index)));//如果提供了

null)

{searchRequest.source().parseXContent(requestContentParser,

true);}

{searchRequest.source().parseXContent(requestContentParser,

true,

searchUsageHolder);}}//设置批量减少大小参数。

final

int

request.paramAsInt(batched_reduce_size,

searchRequest.getBatchedReduceSize());searchRequest.setBatchedReduceSize(batchedReduceSize);//如果请求中包含了

(request.hasParam(pre_filter_shard_size))

{searchRequest.setPreFilterShardSize(request.paramAsInt(pre_filter_shard_size,

SearchRequest.DEFAULT_PRE_FILTER_SHARD_SIZE));}//如果请求中包含了

(request.hasParam(enable_fields_emulation))

{//

errorsrequest.paramAsBoolean(enable_fields_emulation,

max_concurrent_shard_requests(最大并发分片请求数)

(request.hasParam(max_concurrent_shard_requests))

{final

request.paramAsInt(max_concurrent_shard_requests,searchRequest.getMaxConcurrentShardRequests());searchRequest.setMaxConcurrentShardRequests(maxConcurrentShardRequests);}//如果请求中包含了

allow_partial_search_results(允许部分搜索结果)

(request.hasParam(allow_partial_search_results))

{//仅当我们传递了参数以覆盖集群级默认值时才设置searchRequest.allowPartialSearchResults(request.paramAsBoolean(allow_partial_search_results,

null));}//设置搜索类型参数。

searchRequest.searchType(request.param(search_type));//调用

parseSearchSource

方法解析搜索源。

parseSearchSource(searchRequest.source(),

request,

setSize);//设置请求缓存参数searchRequest.requestCache(request.paramAsBoolean(request_cache,

searchRequest.requestCache()));//解析并设置滚动参数。

String

scroll

scroll)));}//设置路由参数。

searchRequest.routing(request.param(routing));//设置首选项参数。

searchRequest.preference(request.param(preference));//设置索引选项。

searchRequest.indicesOptions(IndicesOptions.fromRequest(request,

searchRequest.indicesOptions()));//验证搜索请求。

validateSearchRequest(request,

(searchRequest.pointInTimeBuilder()

null)

{preparePointInTime(searchRequest,

request,

参数searchRequest.setCcsMinimizeRoundtrips(request.paramAsBoolean(ccs_minimize_roundtrips,

searchRequest.isCcsMinimizeRoundtrips()));}//如果请求中包含了

(request.paramAsBoolean(force_synthetic_source,

false))

{searchRequest.setForceSyntheticSource(true);}}//组装searchSourceBuilderprivate

static

{//RestRequest对象的URL参数转换为QueryBuilder对象QueryBuilder

queryBuilder

RestActions.urlParamsToQueryBuilder(request);//并将其设置为SearchSourceBuilder对象的查询条件if

(queryBuilder

{searchSourceBuilder.query(queryBuilder);}//如果RestRequest对象包含from参数则将其转换为整数并设置为SearchSourceBuilder对象的from属性if

{searchSourceBuilder.from(request.paramAsInt(from,

0));}if

treatmentdeprecationLogger.compatibleCritical(search-api-size-1,Using

search

{searchSourceBuilder.explain(request.paramAsBoolean(explain,

null));}if

{searchSourceBuilder.version(request.paramAsBoolean(version,

null));}if

(request.hasParam(seq_no_primary_term))

{searchSourceBuilder.seqNoAndPrimaryTerm(request.paramAsBoolean(seq_no_primary_term,

null));}if

{searchSourceBuilder.timeout(request.paramAsTime(timeout,

null));}if

(request.hasParam(terminate_after))

{int

request.paramAsInt(terminate_after,

SearchContext.DEFAULT_TERMINATE_AFTER);searchSourceBuilder.terminateAfter(terminateAfter);}StoredFieldsContext

storedFieldsContext

StoredFieldsContext.fromRestRequest(SearchSourceBuilder.STORED_FIELDS_FIELD.getPreferredName(),request);if

(storedFieldsContext

{searchSourceBuilder.storedFields(storedFieldsContext);}String

sDocValueFields

request.param(docvalue_fields);if

(sDocValueFields

(Strings.hasText(sDocValueFields))

{String[]

Strings.splitStringByCommaToArray(sDocValueFields);for

(String

{searchSourceBuilder.docValueField(field,

fetchSourceContext

FetchSourceContext.parseFromRestRequest(request);if

(fetchSourceContext

{searchSourceBuilder.fetchSource(fetchSourceContext);}if

(request.hasParam(track_scores))

{searchSourceBuilder.trackScores(request.paramAsBoolean(track_scores,

false));}if

(request.hasParam(track_total_hits))

{if

(Booleans.isBoolean(request.param(track_total_hits)))

{searchSourceBuilder.trackTotalHits(request.paramAsBoolean(track_total_hits,

true));}

{searchSourceBuilder.trackTotalHitsUpTo(request.paramAsInt(track_total_hits,

SearchContext.DEFAULT_TRACK_TOTAL_HITS_UP_TO));}}String

sSorts

Strings.splitStringByCommaToArray(sSorts);for

(String

{searchSourceBuilder.sort(sortField,

SortOrder.ASC);}

{searchSourceBuilder.sort(sortField,

SortOrder.DESC);}}

{searchSourceBuilder.sort(sort);}}}String

sStats

{searchSourceBuilder.stats(Arrays.asList(Strings.splitStringByCommaToArray(sStats)));}SuggestBuilder

suggestBuilder

parseSuggestUrlParameters(request);if

(suggestBuilder

{searchSourceBuilder.suggest(suggestBuilder);}}

2、往数据节点发送查询请求的action(TransportSearchAction)

下面这个TransportSearchAction也有点长主要流程是doExecute-executeLocalSearch-executeSearch-接口SearchPhaseProvider的实现类AsyncSearchActionProvider

public

HandledTransportActionSearchRequest,

SearchResponse

AsyncSearchActionProvider::new);}//主要功能是执行搜索请求并根据不同的情况选择执行本地搜索或远程搜索void

task,SearchRequest

original,ActionListenerSearchResponse

listener,FunctionActionListenerSearchResponse,

SearchPhaseProvider

SearchTimeProvider(original.getOrCreateAbsoluteStartMillis(),relativeStartNanos,System::nanoTime);//使用重写监听器对搜索请求进行重写并根据重写后的请求获取搜索上下文和远程集群索引。

ActionListenerSearchRequest

rewriteListener

{checkCCSVersionCompatibility(rewritten);}if

(rewritten.pointInTimeBuilder()

null)

rewritten.pointInTimeBuilder().getSearchContextId(namedWriteableRegistry);remoteClusterIndices

getIndicesFromSearchContexts(searchContext,

else

remoteClusterService.groupIndices(rewritten.indicesOptions(),

rewritten.indices())

remoteClusterService.groupIndices(rewritten.indicesOptions(),

remoteClusterIndices

RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY

localIndices

remoteClusterIndices.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);//获取当前集群状态

ClusterState

clusterService.state();//如果远程集群索引为空则执行本地搜索if

(remoteClusterIndices.isEmpty())

{executeLocalSearch(task,timeProvider,rewritten,localIndices,clusterState,SearchResponse.Clusters.EMPTY,searchContext,searchPhaseProvider.apply(listener));}

else

{//如果远程集群索引不为空则根据是否需要最小化往返次数选择执行远程搜索或本地搜索。

//省略目前不涉及到远程集群}}},

listener::onFailure);Rewriteable.rewriteAndFetch(original,

searchService.getRewriteContext(timeProvider::absoluteStartMillis),

clusterState,SearchResponse.Clusters

searchContext,SearchPhaseProvider

searchPhaseProvider)

task,timeProvider,searchRequest,localIndices,Collections.emptyList(),(clusterName,

nodeId)

null,clusterState,Collections.emptyMap(),clusterInfo,searchContext,searchPhaseProvider);}private

void

localIndices,ListSearchShardIterator

remoteShardIterators,BiFunctionString,

String,

remoteAliasMap,SearchResponse.Clusters

clusters,Nullable

searchContext,SearchPhaseProvider

searchPhaseProvider)

{//检查全局集群阻塞状态是否允许读取操作clusterState.blocks().globalBlockedRaiseException(ClusterBlockLevel.READ);//检查搜索请求中是否定义了allowPartialSearchResults(允许部分搜索结果)参数如果没有则应用集群服务的默认设置。

if

(searchRequest.allowPartialSearchResults()

null)

{//默认允许部分搜索结果searchRequest.allowPartialSearchResults(searchService.defaultAllowPartialSearchResults());}final

MapString,

concreteLocalIndices;//根据搜索上下文的存在与否确定本地和远程的索引和别名过滤器。

if

(searchContext

searchRequest.pointInTimeBuilder()

null;aliasFilter

searchContext.aliasFilter();concreteLocalIndices

localIndices

localIndices.indices();localShardIterators

getLocalLocalShardsIteratorFromPointInTime(clusterState,localIndices,searchRequest.getLocalClusterAlias(),searchContext,searchRequest.pointInTimeBuilder().getKeepAlive(),searchRequest.allowPartialSearchResults());}

else

{//解析本地索引获取Index对象数组indices。

final

Index[]

resolveLocalIndices(localIndices,

clusterState,

timeProvider);//将indices数组中的每个Index对象的名称提取出来并存储在concreteLocalIndices数组中。

concreteLocalIndices

Arrays.stream(indices).map(Index::getName).toArray(String[]::new);//解析索引名称表达式获取与搜索请求中的索引相关的索引和别名的集合indicesAndAliases。

final

SetString

indexNameExpressionResolver.resolveExpressions(clusterState,

searchRequest.indices());//构建索引别名过滤器aliasFilter

buildIndexAliasFilters(clusterState,

indicesAndAliases,

indices);//将remoteAliasMap中的所有映射添加到aliasFilter中aliasFilter.putAll(remoteAliasMap);//取本地分片迭代器localShardIteratorslocalShardIterators

getLocalShardsIterator(clusterState,searchRequest,searchRequest.getLocalClusterAlias(),indicesAndAliases,concreteLocalIndices);}//合并创建一个GroupShardsIteratorSearchShardIterator对象并赋值给shardIterators变量。

final

GroupShardsIteratorSearchShardIterator

shardIterators

mergeShardsIterators(localShardIterators,

remoteShardIterators);//检查shardIterators的大小是否超过了集群设定的分片数量限制如果超过则抛出异常。

failIfOverShardCountLimit(clusterService,

shardIterators.size());//WaitForCheckpointsbuwei1if

(searchRequest.getWaitForCheckpoints().isEmpty()

false)

(remoteShardIterators.isEmpty()

false)

IllegalArgumentException(Cannot

use

{validateAndResolveWaitForCheckpoint(clusterState,

searchRequest,

concreteLocalIndices);}}MapString,

Float

resolveIndexBoosts(searchRequest,

clusterState);//shardIterators的大小调整搜索类型。

adjustSearchType(searchRequest,

DiscoveryNodes

clusterState.nodes();//构建一个连接查询函数connectionLookup用于根据索引和节点名称获取连接对象。

BiFunctionString,

String,

buildConnectionLookup(searchRequest.getLocalClusterAlias(),nodes::get,remoteConnections,searchTransportService::getConnection);//创建一个异步搜索执行器asyncSearchExecutor用于执行异步搜索。

final

Executor

asyncSearchExecutor(concreteLocalIndices);//根据条件判断是否需要预过滤搜索分片。

final

boolean

shouldPreFilterSearchShards(clusterState,searchRequest,concreteLocalIndices,localShardIterators.size()

remoteShardIterators.size(),defaultPreFilterShardSize);//调用searchPhaseProvider的newSearchPhase方法开始执行搜索阶段//searchPhaseProvider的实现用的是AsyncSearchActionProvidersearchPhaseProvider.newSearchPhase(task,searchRequest,asyncSearchExecutor,shardIterators,timeProvider,connectionLookup,clusterState,Collections.unmodifiableMap(aliasFilter),concreteIndexBoosts,preFilterSearchShards,threadPool,clusters).start();}//一个接口interface

SearchPhaseProvider

executor,GroupShardsIteratorSearchShardIterator

shardIterators,SearchTimeProvider

String,

threadPool,SearchResponse.Clusters

clusters);}//接口SearchPhaseProvider的一个实现类private

class

listener;AsyncSearchActionProvider(ActionListenerSearchResponse

listener)

executor,GroupShardsIteratorSearchShardIterator

shardIterators,SearchTimeProvider

String,

threadPool,SearchResponse.Clusters

clusters)

searchPhaseController.newSearchPhaseResults(executor,circuitBreaker,task::isCancelled,task.getProgressListener(),searchRequest,shardIterators.size(),exc

searchTransportService.cancelSearchTask(task,

failed

searchRequest.searchType();return

new

SearchQueryThenFetchAsyncAction(logger,searchTransportService,connectionLookup,aliasFilter,concreteIndexBoosts,executor,queryResultConsumer,searchRequest,listener,shardIterators,timeProvider,clusterState,task,clusters);}}}}

}其中searchType

相同但初始散射阶段除外该阶段用于计算分布项频率以实现更准确的评分。

*/DFS_QUERY_THEN_FETCH((byte)

0),/**

对所有分片执行查询但仅返回足够的信息而不是文档内容。

然后对结果进行排序和排名并基于此*

仅要求相关分片提供实际文档内容。

返回的命中数与大小中指定的命中数完全相同因为它们是唯一被提取的命中数。

当索引有很多分片不是副本、分片

组时这非常方便。

*/QUERY_THEN_FETCH((byte)

1);//

}SearchQueryThenFetchAsyncAction的实现如下

class

SearchQueryThenFetchAsyncAction

extends

AbstractSearchAsyncActionSearchPhaseResult

{private

bottomSortCollector;SearchQueryThenFetchAsyncAction(final

Logger

GroupShardsIteratorSearchShardIterator

shardsIts,final

TransportSearchAction.SearchTimeProvider

clusters)

{super(query,logger,searchTransportService,nodeIdToConnection,aliasFilter,concreteIndexBoosts,executor,request,listener,shardsIts,timeProvider,clusterState,task,resultConsumer,request.getMaxConcurrentShardRequests(),clusters);//省略代码}//父类的performPhaseOnShard方法会调用这个方法protected

void

SearchActionListenerSearchPhaseResult

listener)

rewriteShardSearchRequest(super.buildShardSearchRequest(shardIt,

listener.requestIndex));getSearchTransport().sendExecuteQuery(getConnection(shard.getClusterAlias(),

shard.getNodeId()),

}其中上面的executeSearch方法中searchPhaseProvider.newSearchPhase().start()实际执行的是SearchQueryThenFetchAsyncAction的父类AbstractSearchAsyncAction中的start方法

abstract

AbstractSearchAsyncActionResult

extends

这是搜索的主要入口点。

此方法启动初始阶段的搜索执行。

*/public

final

{//省略return;}executePhase(this);}private

void

{//toSkipShardsIts中的每个SearchShardIterator对象调用skip()方法并断言其返回值为true然后调用skipShard()方法for

(final

iterator.skip();skipShard(iterator);}//如果shardsIts的大小大于0if

(shardsIts.size()

(request.allowPartialSearchResults()

false)

{//省略代码}//对于shardsIts中的每个索引获取对应的SearchShardIterator对象shardRoutings然后执行performPhaseOnShard()方法。

//这里会遍历每一个分片for

(int

shardIndexMap.containsKey(shardRoutings);int

shardIndex

shardIndexMap.get(shardRoutings);performPhaseOnShard(shardIndex,

shardRoutings,

shardRoutings.nextOrNull());}}}protected

void

{//该分片未分配给任何节点会触发onShardFailure方法处理该情况assert

assertExecuteOnStartThread();SearchShardTarget

unassignedShard

shardIt.getClusterAlias());onShardFailure(shardIndex,

unassignedShard,

NoShardAvailableActionException(shardIt.shardId()));}

else

{//创建一个Runnable对象并执行executePhaseOnShard方法来在分片上执行搜索操作。

final

PendingExecutions

pendingExecutionsPerNode.computeIfAbsent(shard.getNodeId(),

new

PendingExecutions(maxConcurrentRequestsPerNode)):

null;Runnable

SearchActionListenerResult(shard,

shardIndex)

{//定义了一个SearchActionListener的匿名子类用于处理搜索操作的响应。

Overridepublic

void

{在响应成功时会调用onShardResult方法处理搜索结果onShardResult(result,

shardIt);}

{//在响应失败时会调用onShardFailure方法处理错误情况onShardFailure(shardIndex,

shard,

{//无论成功还是失败最后都会调用executeNext方法执行下一个操作。

executeNext(pendingExecutions,

void

{executeNext(pendingExecutions,

thread);}}});}

{executeNext(pendingExecutions,

thread);}}};//如果throttleConcurrentRequests为true则会使用pendingExecutions对象来限制并发请求的数量。

否则直接执行r.run()方法。

if

else

这个抽象方法由子类SearchQueryThenFetchAsyncAction实现*

abstract

executePhaseOnShard(SearchShardIterator

shard,SearchActionListenerResult

listener);

}3、通过transportService把查询请求发送到指定的数据节点

public

registerRequestHandler(TransportService

transportService,

{transportService.registerRequestHandler(QUERY_ACTION_NAME,ThreadPool.Names.SAME,ShardSearchRequest::new,(request,

channel,

searchService.executeQueryPhase(request,(SearchShardTask)

task,new

ChannelActionListener(channel)));

}public

sendExecuteQuery(Transport.Connection

connection,final

anymore.//我们对此进行了优化如果我们在搜索请求中只有一个分片则期望

QueryFetchSearchResult这曾经是不再存在的QUERY_AND_FETCH。

final

boolean

null);Writeable.ReaderSearchPhaseResult

reader

responseWrapper.apply(connection,

listener);//上面根据QUERY_ACTION_NAME注册的实际调用的是

searchService.executeQueryPhasetransportService.sendChildRequest(connection,QUERY_ACTION_NAME,request,task,new

ConnectionCountingHandler(handler,

reader,

connection.getNode().getId()));}二、数据节点收到请求的处理逻辑

public

executeQueryPhase(ShardSearchRequest

request,

ActionListenerSearchPhaseResult

listener)

request.canReturnNullResponseIfMatchNoDocs()

false

shard;//根据request对象获取一个IndexShard对象final

IndexShard

getShard(request);//调用rewriteAndFetchShardRequest方法对shard和request进行重写和获取请求。

rewriteAndFetchShardRequest(shard,

request,

(orig.canReturnNullResponseIfMatchNoDocs())

{assert

{//创建一个ShardSearchRequest对象的副本clone并调用canMatch方法进行匹配检查。

ShardSearchRequest

clone

ShardSearchRequest(orig);canMatchResp

canMatch(clone,

{l.onResponse(QuerySearchResult.nullInstance());return;}}//其中会执行executeQueryPhase方法的递归调用。

ensureAfterSeqNoRefreshed(shard,

orig,

的引用计数将通过此方法递增。

调用方有责任确保在不再需要对象时正确递减引用计数。

*/private

SearchPhaseResult

executeQueryPhase(ShardSearchRequest

request,

createOrGetReaderContext(request);try

(//创建SearchContext对象并设置相关参数。

Releasable

scope

tracer.withScope(task);Releasable

ignored

readerContext.markAsUsed(getKeepAlive(request));SearchContext

context

{//开始跟踪执行查询阶段。

tracer.startTrace(executeQueryPhase,

Map.of());final

afterQueryTime;//使用SearchOperationListenerExecutor执行加载或执行查询阶段的操作。

try

(SearchOperationListenerExecutor

executor

SearchOperationListenerExecutor(context))

{loadOrExecuteQueryPhase(request,

context);//检查查询结果是否具有搜索上下文并根据需要释放ReaderContext对象。

if

(context.queryResult().hasSearchContext()

false

{freeReaderContext(readerContext.id());}afterQueryTime

executor.success();}

{//停止跟踪执行查询阶段。

tracer.stopTrace();}//根据条件判断是否需要执行提取阶段if

(request.source()

fetchcontext.addFetchResult();//如果需要执行提取阶段则将提取结果添加到SearchContext对象并调用executeFetchPhase方法执行提取阶段。

return

executeFetchPhase(readerContext,

context,

{//将RescoreDocIds对象传递给queryResult并返回context.queryResult()。

//将

rescoreDocIds

queryResult以将它们发送到协调节点并在提取阶段接收它们。

我们还将

rescoreDocIds

LegacyReaderContext以防搜索状态需要保留在数据节点中final

RescoreDocIds

context.rescoreDocIds();context.queryResult().setRescoreDocIds(rescoreDocIds);readerContext.setRescoreDocIds(rescoreDocIds);context.queryResult().incRef();return

catch

ElasticsearchException(e.getCause());}logger.trace(Query

phase

e);processFailure(readerContext,

e);throw

QuerySearchResultnullInstance}*

则返回

false因为协调器节点至少需要一个分片响应来构建全局响应。

*/public

boolean

canReturnNullResponseIfMatchNoDocs()

{return

canReturnNullResponseIfMatchNoDocs;}1、尝试从缓存中加载查询结果

/***

如果无法使用缓存请尝试从缓存加载查询结果或直接执行查询阶段。

*/private

void

IndicesService.canCache(request,

context);context.getSearchExecutionContext().freezeContext();if

(canCache)

{indicesService.loadIntoContext(request,

context);}

{QueryPhase.execute(context);}}/***

加载缓存结果根据需要通过执行查询阶段进行计算否则将缓存值反序列化为

{link

的组合允许进行单个加载操作这将导致具有相同密钥的其他请求等待直到其加载并重用相同的缓存。

*/public

void

loadIntoContext(ShardSearchRequest

request,

的搜索器searcher的阅读器对象BytesReference

cacheKey,

//cacheKey缓存键CheckedConsumerStreamOutput,

IOException

//一个带有StreamOutput参数的回调函数用于加载数据)

throws

{//创建一个IndexShardCacheEntity对象用于表示索引分片的缓存实体IndexShardCacheEntity

cacheEntity

IndexShardCacheEntity(shard);//创建一个CheckedSupplier对象用于生成缓存数据。

CheckedSupplierBytesReference,

IOException

{//这个对象内部使用BytesStreamOutput它允许指定预期的字节大小//

但默认使用16k作为页面大小。

为了避免对小查询结果浪费太多内存将预期大小设置为512字节。

final

int

512;//在BytesStreamOutput中执行loader回调函数将数据写入输出流中try

(BytesStreamOutput

BytesStreamOutput(expectedSizeInBytes))

{loader.accept(out);//将输出流的字节表示返回作为缓存数据。

return

out.bytes();}};//通过调用indicesRequestCache.getOrCompute方法使用缓存实体、缓存数据生成器、映射缓存键、目录阅读器和缓存键作为参数获取或计算缓存数据return

indicesRequestCache.getOrCompute(cacheEntity,

supplier,

cacheKey);}这里要知道supplier内部会执行loader.accept(out);而传过来的loader是如下

out

{QueryPhase.execute(context);context.queryResult().writeToNoId(out);loadedFromCache[0]

false;}其实意味着如果执行了loader说明缓存中没有而是直接查询的继续往下

BytesReference

cacheEntity,CheckedSupplierBytesReference,

IOException

mappingCacheKey,DirectoryReader

cacheKey)

ElasticsearchDirectoryReader.getESReaderCacheHelper(reader);assert

cacheHelper

{key.entity.onMiss();//看看这是否是我们第一次看到这个读取器并确保注册一个清理密钥CleanupKey

cleanupKey

(registeredClosedListeners.containsKey(cleanupKey)

false)

registeredClosedListeners.putIfAbsent(cleanupKey,

Boolean.TRUE);if

{cacheHelper.addClosedListener(cleanupKey);}}}

else

null则尝试使用给定的映射函数计算其值并将其输入到此映射中除非为

load

而第二个加载器函数将返回第一个加载器函数提供的结果包括执行第一个加载器函数期间引发的任何异常*/public

computeIfAbsent(K

fail//尝试从缓存中获取与给定键关联的值如果值已过期则会在获取前将其删除。

V

value

时按住段锁可能会导致由于依赖键加载而对另一个线程进行死锁;//

因此我们需要一种机制来确保最多调用一次

中//首先获取与给定键关联的缓存段CacheSegment。

CacheSegment

segment

getCacheSegment(key);CompletableFutureEntryK,

future;//创建一个CompletableFuture对象用于在加载完成后获取值。

CompletableFutureEntryK,

completableFuture

CompletableFuture();//使用段锁将键和CompletableFuture对象放入段的映射Map中。

try

(ReleasableLock

completableFuture);}BiFunction?

super

(segment.writeLock.acquire())并使用try-with-resources语句来确保锁被释放。

try

(ReleasableLock

{//检查segment.map是否为空如果不为空则尝试从segment.map中获取与key对应的CompletableFutureEntryK,

sanity

sanity.isCompletedExceptionally())

{//如果sanity不为空且已经完成异常则从segment.map中移除key。

segment.map.remove(key);if

{//如果segment.map为空则将其赋值为null。

segment.map

null;}}}return

completableValue;//如果该键之前不存在映射则说明当前线程赢得了竞争需要执行加载操作。

if

(future

completableFuture;completableValue

loaded;//调用加载器的load方法加载值并将其放入CompletableFuture对象中。

try

{loaded

{future.completeExceptionally(e);throw

new

value);future.completeExceptionally(npe);throw

new

{//将加载的值包装成一个Entry对象并完成CompletableFuture对象。

future.complete(new

Entry(key,

{//说明该键存在映射直接调用completableValue

future.handle(handler);}//通过completableValue.get()获取加载完成的值try

{value

(future.isCompletedExceptionally())

{future.get();

IllegalStateException(e);}}return

value;}这里面在从缓存中没有得到指定的CacheSegment

则会调用loader.load(key)

里面实际调用的是QueryPhase.execute(context);

这里就看一下QueryPhase.execute(context);的实现源码

/***

//搜索请求的查询阶段用于运行查询并从每个分片中获取有关匹配文档的信息*/

public

(searchContext.rankShardContext()

null)

{executeRank(searchContext);}}(1)executeQuery和executeRank两种查询方式

static

{//获取排名的上下文信息和查询结果信息RankShardContext

rankShardContext

searchContext.rankShardContext();QuerySearchResult

querySearchResult

searchContext.queryResult();//然后根据条件判断是否需要执行组合布尔查询以获取总命中数或聚合结果,if

(searchContext.trackTotalHitsUpTo()

null)

{//需要的话则size0再执行executeQuery,来获取总命中数和聚合结果searchContext.size(0);QueryPhase.executeQuery(searchContext);}

else

{//将查询结果的topDocs设置为空即命中文档为空。

searchContext.queryResult().topDocs(new

TopDocs(new

DocValueFormat[0]);}ListTopDocs

rrfRankResults

querySearchResult.searchTimedOut();long

serviceTimeEWMA

querySearchResult.serviceTimeEWMA();int

nodeQueueSize

querySearchResult.nodeQueueSize();//迭代rankShardContext.queries()中的每个排名查询来执行排名操作for

(Query

{break;}//对于每个排名查询创建一个RankSearchContext对象RankSearchContext

rankSearchContext

RankSearchContext(searchContext,

rankQuery,

rankShardContext.windowSize());//并添加收集器和搜索操作QueryPhase.addCollectorsAndSearch(rankSearchContext);//然后将查询结果添加到rrfRankResults列表中并更新服务时间、节点队列大小和搜索超时的状态。

QuerySearchResult

rrfQuerySearchResult

rankSearchContext.queryResult();rrfRankResults.add(rrfQuerySearchResult.topDocs().topDocs);serviceTimeEWMA

rrfQuerySearchResult.serviceTimeEWMA();nodeQueueSize

rrfQuerySearchResult.nodeQueueSize());searchTimedOut

rrfQuerySearchResult.searchTimedOut();}//将排名结果通过rankShardContext.combine方法进行合并并将相关的值记录到querySearchResult中querySearchResult.setRankShardResult(rankShardContext.combine(rrfRankResults));//包括搜索超时状态、服务时间和节点队列大小。

//

record

queriesquerySearchResult.searchTimedOut(searchTimedOut);querySearchResult.serviceTimeEWMA(serviceTimeEWMA);querySearchResult.nodeQueueSize(nodeQueueSize);}static

void

{//检查searchContext是否只有建议suggest操作如果是就执行建议阶段的操作并返回一个空的查询结果if

(searchContext.hasOnlySuggest())

{SuggestPhase.execute(searchContext);searchContext.queryResult().topDocs(new

TopDocs(new

SearchContextSourcePrinter(searchContext));}//

Pre-process

phase//聚合aggregation进行预处理操作AggregationPhase.preProcess(searchContext);//添加收集器collectors并执行搜索操作addCollectorsAndSearch(searchContext);//执行重新评分rescore阶段的操作RescorePhase.execute(searchContext);//再次执行建议阶段的操作。

SuggestPhase.execute(searchContext);//执行聚合阶段的操作。

AggregationPhase.execute(searchContext);//如果searchContext中包含性能分析器profiler则对查询阶段的性能结果进行分析。

if

null)

{searchContext.queryResult().profileResults(searchContext.getProfilers().buildQueryPhaseResults());}}这两种最后还是要调用

QueryPhase.addCollectorsAndSearch进行查询只是executeRank

会多一层判断执行两遍addCollectorsAndSearch

static

addCollectorsAndSearch(SearchContext

searchContext)

searcher.getIndexReader();QuerySearchResult

queryResult

searchContext.queryResult();//设置查询结果的超时状态queryResult.searchTimedOut(false);try

{//起始位置和大小。

queryResult.from(searchContext.from());queryResult.size(searchContext.size());//重写查询并通过断言确认查询已经重写。

Query

query

searchContext.rewrittenQuery();assert

query

searchContext.scrollContext();if

(scrollContext

hits//我们在这里可以优化的不多因为我们想收集所有文档以获得总点击数}

else

{//如果不是第一轮滚动查询根据排序条件判断是否可以提前终止查询并构建新的查询对象。

final

ScoreDoc

scrollContext.lastEmittedDoc;if

{//

doc//由于搜索排序是索引排序的前缀我们可以直接跳到所需的文档if

(after

BooleanQuery.Builder().add(query,

BooleanClause.Occur.MUST).add(new

SearchAfterSortedDocQuery(searchContext.sort().sort,

(FieldDoc)

BooleanClause.Occur.FILTER).build();}}}}//创建顶部文档收集器。

//

create

createTopDocsCollectorFactory(searchContext,searchContext.parsedPostFilter()

null

null);CollectorManagerCollector,

Void

wrapWithProfilerCollectorManagerIfNeeded(searchContext.getProfilers(),topDocsFactory.collectorManager(),topDocsFactory.profilerName);//根据条件添加收集器//如果设置了terminate_after参数添加一个用于终止查询的收集器。

if

(searchContext.terminateAfter()

SearchContext.DEFAULT_TERMINATE_AFTER)

{//

collectorsTerminateAfterCollector

new

TerminateAfterCollector(searchContext.terminateAfter());final

Collector

collectorManager.newCollector();collectorManager

wrapWithProfilerCollectorManagerIfNeeded(searchContext.getProfilers(),new

SingleThreadCollectorManager(MultiCollector.wrap(terminateAfterCollector,

collector)),REASON_SEARCH_TERMINATE_AFTER_COUNT,collector);}//如果存在后置过滤器添加一个用于过滤结果的收集器。

if

(searchContext.parsedPostFilter()

null)

searcher.createWeight(searcher.rewrite(searchContext.parsedPostFilter().query()),ScoreMode.COMPLETE_NO_SCORES,1f);final

Collector

collectorManager.newCollector();collectorManager

wrapWithProfilerCollectorManagerIfNeeded(searchContext.getProfilers(),new

SingleThreadCollectorManager(new

filterWeight)),REASON_SEARCH_POST_FILTER,collector);}//如果存在聚合操作添加一个用于聚合的收集器。

if

null)

collectorManager.newCollector();final

Collector

searchContext.aggregations().getAggsCollectorManager().newCollector();collectorManager

wrapWithProfilerCollectorManagerIfNeeded(searchContext.getProfilers(),new

SingleThreadCollectorManager(MultiCollector.wrap(collector,

aggsCollector)),REASON_SEARCH_MULTI,collector,aggsCollector);}//如果设置了最小分数添加一个用于过滤低分结果的收集器。

if

null)

collectorManager.newCollector();//

apply

wrapWithProfilerCollectorManagerIfNeeded(searchContext.getProfilers(),new

SingleThreadCollectorManager(new

MinimumScoreCollector(collector,

searchContext.minimumScore())),REASON_SEARCH_MIN_SCORE,collector);}//根据超时设置添加查询超时检查的任务。

final

Runnable

getTimeoutCheck(searchContext);if

(timeoutRunnable

{searcher.addQueryCancellation(timeoutRunnable);}try

{//使用收集器管理器执行查询并更新查询结果。

searchWithCollectorManager(searchContext,

searcher,

null);queryResult.topDocs(topDocsFactory.topDocsAndMaxScore(),

topDocsFactory.sortValueFormats);//获取线程池执行器对象并根据类型更新查询结果的节点队列大小和服务时间指数加权移动平均值。

ExecutorService

executor

searchContext.indexShard().getThreadPool().executor(ThreadPool.Names.SEARCH);assert

executor

TaskExecutionTimeTrackingEsThreadPoolExecutor||

(executor

TaskExecutionTimeTrackingEsThreadPoolExecutor

rExecutor)

{queryResult.nodeQueueSize(rExecutor.getCurrentQueueSize());queryResult.serviceTimeEWMA((long)

rExecutor.getTaskExecutionEWMA());}}

finally

{searcher.removeQueryCancellation(timeoutRunnable);}}}

catch

QueryPhaseExecutionException(searchContext.shardTarget(),

Failed

e);}}会通过searchWithCollectorManager

来执行查询

searchWithCollectorManager(SearchContext

searchContext,ContextIndexSearcher

searcher,Query

query,CollectorManagerCollector,

Void

{//如果profilers不为null则获取当前查询的分析器并将collectorManager设置为InternalProfileCollectorManager的getCollectorTree方法。

if

null)

{searchContext.getProfilers().getCurrentQueryProfiler().setCollectorManager(((InternalProfileCollectorManager)

collectorManager)::getCollectorTree);}//获取searchContext的查询结果对象queryResult。

QuerySearchResult

queryResult

searchContext.queryResult();try

{//使用searcher和collectorManager执行查询操作。

searcher.search(query,

collectorManager);}

(TerminateAfterCollector.EarlyTerminationException

{//如果查询被TerminateAfterCollector.EarlyTerminationException异常提前终止则将queryResult的terminatedEarly属性设置为true。

queryResult.terminatedEarly(true);}

catch

set;//如果查询超时且timeoutSet为true则检查searchContext的request是否允许部分搜索结果。

if

(searchContext.request().allowPartialSearchResults()

false)

{//如果不允许部分搜索结果则抛出QueryPhaseExecutionException异常指示查询超时。

//

Cant

QueryPhaseExecutionException(searchContext.shardTarget(),

Time

exceeded);}//如果允许部分搜索结果则将queryResult的searchTimedOut属性设置为true。

queryResult.searchTimedOut(true);}//如果searchContext的terminateAfter属性不等于SearchContext.DEFAULT_TERMINATE_AFTER且queryResult的terminatedEarly属性为null则将queryResult的terminatedEarly属性设置为false。

if

(searchContext.terminateAfter()

SearchContext.DEFAULT_TERMINATE_AFTER

null)

{queryResult.terminatedEarly(false);}}(3)

Overridepublic

{//通过collectorManager创建一个收集器(Collector)的实例firstCollector。

final

firstCollector

collectorManager.newCollector();//

Take

needed.//根据firstCollector的评分模式(scoreMode)判断是否需要评分,如果需要评分则使用rewrite方法对查询进行重写如果不需要评分则使用ConstantScoreQuery对查询进行重写。

query

firstCollector.scoreMode().needsScores()

rewrite(query)

ConstantScoreQuery(query));//根据重写后的查询(query)、评分模式(scoreMode)和权重(Weight)创建一个权重对象(weight)。

final

Weight

1);//调用search方法传入权重对象、收集器管理器和第一个收集器执行搜索操作并返回结果。

return

search(weight,

firstCollector);}/**每个元素表示一个分片的信息*LeafSlice的数量就代表了索引的分片数量。

每个LeafSlice对象代表了一个分片的信息和上下文。

*

如果一个索引在这个数据节点有5个分片则这个的长度为5*/private

final

{//如果queueSizeBasedExecutor为空或者leafSlices的长度小于等于1//LeafSlice关键字解释是IndexSearcher用来管理和表示索引搜索分片的类,如果小于等于1则此数据节点只有一个分片if

null

{//那么直接在leafContexts上执行搜索操作search(leafContexts,

weight,

firstCollector);//并通过collectorManager.reduce方法将结果收集起来返回。

return

collectorManager.reduce(Collections.singletonList(firstCollector));}

else

ArrayList(leafSlices.length);collectors.add(firstCollector);final

ScoreMode

firstCollector.scoreMode();//并使用collectorManager创建新的收集器。

for

(int

collectorManager.newCollector();collectors.add(collector);if

(scoreMode

IllegalStateException(CollectorManager

does

mode);}}//创建一个FutureTask列表listTasks用于异步执行搜索操作final

ListFutureTaskC

ArrayList();//遍历leafSlices对每个leafSlices创建一个FutureTask并将其添加到listTasks中。

for

(int

collector;});listTasks.add(task);}//使用queueSizeBasedExecutor的invokeAll方法执行所有的listTasks等待它们的执行完成。

queueSizeBasedExecutor.invokeAll(listTasks);RuntimeException

exception

ArrayList();//遍历listTasks获取每个任务的结果并将其添加到collectedCollectors列表中。

for

(FutureC

{collectedCollectors.add(future.get());//

TODO:

{//省略代码}}//通过collectorManager.reduce方法将所有收集器的结果进行组合并返回最终结果。

return

collectorManager.reduce(collectedCollectors);}}需要注意leafSlices

Overridepublic

{collector.setWeight(weight);for

(LeafReaderContext

collector);}}1、主节点将查询请求路由(根据分片找到数据节点)到对应的数据节点执行查询请求因为数据节点无法知道查询请求是否仅针对某一个具体的分片。

数据节点会在所有分片上执行查询操作并将结果进行合并、去重和处理以产生最终的结果。

2、因此即使主节点发送的查询请求只涉及一个分片但在实际查询过程中数据节点会遍历该数据节点上所有与该索引对应的分片以保证查询结果的完整性。

private

{//检查是否取消了搜索操作。

cancellable.checkCancelled();//获取当前叶子节点的LeafCollector。

用于收集匹配的文档。

final

LeafCollector

{//获取当前叶子节点的存活文档集合。

leafCollector

collector.getLeafCollector(ctx);}

catch

{return;}//获取当前叶子节点的存活文档集合位集合表示哪些文档未被删除Bits

liveDocs

ctx.reader().getLiveDocs();//将存活文档集合转换为稀疏位集合BitSet

liveDocsBitSet

getSparseBitSetOrNull(liveDocs);//如果存活文档集合不是稀疏位集合那么使用BulkScorer进行评分操作。

if

(liveDocsBitSet

CancellableBulkScorer(bulkScorer,

cancellable::checkCancelled);}try

{//使用BulkScorer对匹配的文档进行评分操作。

bulkScorer.score(leafCollector,

liveDocs);}

{//如果存活文档集合为稀疏位集合//获取权重对象的Scorer用于计算每个候选文档的得分Scorer

scorer

使用Scorer和稀疏位集合liveDocsBitSet对匹配的文档进行交集计算并进行评分操作。

intersectScorerAndBitSet(scorer,liveDocsBitSet,leafCollector,this.cancellable.isEnabled()

{});}

的作用是在搜索阶段过滤掉已被删除的文档并仅处理存活文档。

通过与其他集合的结合使用可以在执行搜索和评分操作时仅处理存活的有效文档可以大大减少内存占用。

从而提升搜索性能和准确性。

其他集合例如Bits、BulkScorer和Scorer可能用于执行搜索和评分操作但往往与存活文档集合无直接关联

//在给定的Scorer、BitSet、LeafCollector和checkCancelled函数的基础上计算它们的交集并将结果收集到collector中。

static

void

intersectScorerAndBitSet(Scorer

scorer,

{//将scorer设置为collector的scorer。

collector.setScorer(scorer);//

ConjunctionDISI

first://创建一个迭代器iterator通过将acceptDocs和scorer的迭代器传递给ConjunctionUtils.intersectIterators()方法来计算它们的交集。

DocIdSetIterator

iterator

ConjunctionUtils.intersectIterators(Arrays.asList(new

acceptDocs.approximateCardinality()),

seen

0;checkCancelled.run();//通过迭代器遍历交集中的文档对于每个文档如果满足一定条件则将其收集到collector中。

for

(int

{//在每次遍历一定数量的文档后调用checkCancelled函数检查是否取消操作。

if

(seen

CHECK_CANCELLED_SCORER_INTERVAL

{checkCancelled.run();}collector.collect(docId);}//最后再次调用checkCancelled函数。

checkCancelled.run();}