<input id="ohw05"></input>
  • <table id="ohw05"><menu id="ohw05"></menu></table>
  • <var id="ohw05"></var>
  • <code id="ohw05"><cite id="ohw05"></cite></code>
    <label id="ohw05"></label>
    <var id="ohw05"></var>
  • ES系列(五):獲取單條數據get處理過程實現

      前面講的都是些比較大的東西,即框架層面的東西。今天咱們來個輕松點的,只講一個點:如題,get單條記錄的es查詢實現。

     

    1. get語義說明

      es中要實現get的查詢,直接就是一個url請求即可:

    curl http://localhost:9200/test/job/1

      get是用于搜索單條es的數據,是根據主鍵id查詢數據方式。類比關系型數據庫中的sql則相當于:

    select * from test where id = #{id};

      當然了,es中每個關鍵詞,都有相當多的附加描述詞匯。比如:指定輸出字段,版本號。。。

     

    2. get的實現簡要說明

      從語義上講,get的結果至多只有一條記錄。所以,雖然es是集群存儲數據的,但此處都需要從某節點取得一條數據即可。所以,理論上,只要能夠快速定位到數據在哪個es節點上,然后向其發起請求,即可獲取到結果了。

      另外,對于使用主鍵id來進行查詢數據,只要數據結構設計得當,應該會有非常高效的查詢能力。

      所以,通過本功能的實現方式分析,我們可以簡要理解es key的分布方式。

     

    3. get的具體實現

      get只是es語法中的一個小點,根據上一節我們分析,知道了如何可以找到處理get的處理器。此處,我們簡單再回顧下:

        // org.elasticsearch.rest.RestController#dispatchRequest
        @Override
        public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
            try {
                // 嘗試所有可能的處理器
                tryAllHandlers(request, channel, threadContext);
            } catch (Exception e) {
                try {
                    // 發生異常則響應異常信息
                    channel.sendResponse(new BytesRestResponse(channel, e));
                } catch (Exception inner) {
                    inner.addSuppressed(e);
                    logger.error(() ->
                        new ParameterizedMessage("failed to send failure response for uri [{}]", request.uri()), inner);
                }
            }
        }
        // org.elasticsearch.rest.RestController#tryAllHandlers
        private void tryAllHandlers(final RestRequest request, final RestChannel channel, final ThreadContext threadContext) throws Exception {
            // 讀取 header 信息
            for (final RestHeaderDefinition restHeader : headersToCopy) {
                final String name = restHeader.getName();
                final List<String> headerValues = request.getAllHeaderValues(name);
                if (headerValues != null && headerValues.isEmpty() == false) {
                    final List<String> distinctHeaderValues = headerValues.stream().distinct().collect(Collectors.toList());
                    if (restHeader.isMultiValueAllowed() == false && distinctHeaderValues.size() > 1) {
                        channel.sendResponse(
                            BytesRestResponse.
                                createSimpleErrorResponse(channel, BAD_REQUEST, "multiple values for single-valued header [" + name + "]."));
                        return;
                    } else {
                        threadContext.putHeader(name, String.join(",", distinctHeaderValues));
                    }
                }
            }
            // error_trace cannot be used when we disable detailed errors
            // we consume the error_trace parameter first to ensure that it is always consumed
            if (request.paramAsBoolean("error_trace", false) && channel.detailedErrorsEnabled() == false) {
                channel.sendResponse(
                        BytesRestResponse.createSimpleErrorResponse(channel, BAD_REQUEST, "error traces in responses are disabled."));
                return;
            }
    
            final String rawPath = request.rawPath();
            final String uri = request.uri();
            final RestRequest.Method requestMethod;
            try {
                // Resolves the HTTP method and fails if the method is invalid
                requestMethod = request.method();
                // Loop through all possible handlers, attempting to dispatch the request
                // 獲取可能的處理器,主要是有正則或者索引變量的存在,可能匹配多個處理器
                Iterator<MethodHandlers> allHandlers = getAllHandlers(request.params(), rawPath);
                while (allHandlers.hasNext()) {
                    final RestHandler handler;
                    // 一個處理器里支持多種請求方法
                    final MethodHandlers handlers = allHandlers.next();
                    if (handlers == null) {
                        handler = null;
                    } else {
                        handler = handlers.getHandler(requestMethod);
                    }
                    if (handler == null) {
                      // 未找到處理器不代表不能處理,有可能需要繼續查找,如果確定不能處理,則直接響應客戶端返回
                      if (handleNoHandlerFound(rawPath, requestMethod, uri, channel)) {
                          return;
                      }
                    } else {
                        // 找到了處理器,調用其方法
                        dispatchRequest(request, channel, handler);
                        return;
                    }
                }
            } catch (final IllegalArgumentException e) {
                handleUnsupportedHttpMethod(uri, null, channel, getValidHandlerMethodSet(rawPath), e);
                return;
            }
            // If request has not been handled, fallback to a bad request error.
            // 降級方法調用
            handleBadRequest(uri, requestMethod, channel);
        }
            // org.elasticsearch.rest.RestController#dispatchRequest
        private void dispatchRequest(RestRequest request, RestChannel channel, RestHandler handler) throws Exception {
            final int contentLength = request.contentLength();
            if (contentLength > 0) {
                final XContentType xContentType = request.getXContentType();
                if (xContentType == null) {
                    sendContentTypeErrorMessage(request.getAllHeaderValues("Content-Type"), channel);
                    return;
                }
                if (handler.supportsContentStream() && xContentType != XContentType.JSON && xContentType != XContentType.SMILE) {
                    channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(channel, RestStatus.NOT_ACCEPTABLE,
                        "Content-Type [" + xContentType + "] does not support stream parsing. Use JSON or SMILE instead"));
                    return;
                }
            }
            RestChannel responseChannel = channel;
            try {
                // 熔斷判定
                if (handler.canTripCircuitBreaker()) {
                    inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
                } else {
                    inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
                }
                // iff we could reserve bytes for the request we need to send the response also over this channel
                responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
                // TODO: Count requests double in the circuit breaker if they need copying?
                if (handler.allowsUnsafeBuffers() == false) {
                    request.ensureSafeBuffers();
                }
                if (handler.allowSystemIndexAccessByDefault() == false && request.header(ELASTIC_PRODUCT_ORIGIN_HTTP_HEADER) == null) {
                    // The ELASTIC_PRODUCT_ORIGIN_HTTP_HEADER indicates that the request is coming from an Elastic product with a plan
                    // to move away from direct access to system indices, and thus deprecation warnings should not be emitted.
                    // This header is intended for internal use only.
                    client.threadPool().getThreadContext().putHeader(SYSTEM_INDEX_ACCESS_CONTROL_HEADER_KEY, Boolean.FALSE.toString());
                }
                // 調用handler處理方法,該handler可能會被過濾器先執行
                handler.handleRequest(request, responseChannel, client);
            } catch (Exception e) {
                responseChannel.sendResponse(new BytesRestResponse(responseChannel, e));
            }
        }
        // org.elasticsearch.xpack.security.rest.SecurityRestFilter#handleRequest
        @Override
        public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
            if (licenseState.isSecurityEnabled() && request.method() != Method.OPTIONS) {
                // CORS - allow for preflight unauthenticated OPTIONS request
                if (extractClientCertificate) {
                    HttpChannel httpChannel = request.getHttpChannel();
                    SSLEngineUtils.extractClientCertificates(logger, threadContext, httpChannel);
                }
    
                final String requestUri = request.uri();
                authenticationService.authenticate(maybeWrapRestRequest(request), ActionListener.wrap(
                    authentication -> {
                        if (authentication == null) {
                            logger.trace("No authentication available for REST request [{}]", requestUri);
                        } else {
                            logger.trace("Authenticated REST request [{}] as {}", requestUri, authentication);
                        }
                        secondaryAuthenticator.authenticateAndAttachToContext(request, ActionListener.wrap(
                            secondaryAuthentication -> {
                                if (secondaryAuthentication != null) {
                                    logger.trace("Found secondary authentication {} in REST request [{}]", secondaryAuthentication, requestUri);
                                }
                                RemoteHostHeader.process(request, threadContext);
                                restHandler.handleRequest(request, channel, client);
                            },
                            e -> handleException("Secondary authentication", request, channel, e)));
                    }, e -> handleException("Authentication", request, channel, e)));
            } else {
                // 轉發到下一處理器責任鏈
                restHandler.handleRequest(request, channel, client);
            }
        }
    View Code

     

      一般地,很多具體的處理器都會繼承 BaseRestHandler, 即 handleRequest() 整體調用時序圖如下:

    handleRequest() 整體調用時序圖

     

     

     

      具體代碼實現如下:

        // org.elasticsearch.rest.BaseRestHandler#handleRequest
        @Override
        public final void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
            // prepare the request for execution; has the side effect of touching the request parameters
            // 看起來叫準備請求,實際非常重要,它會組裝后續的請求邏輯
            final RestChannelConsumer action = prepareRequest(request, client);
    
            // validate unconsumed params, but we must exclude params used to format the response
            // use a sorted set so the unconsumed parameters appear in a reliable sorted order
            final SortedSet<String> unconsumedParams =
                request.unconsumedParams().stream().filter(p -> !responseParams().contains(p)).collect(Collectors.toCollection(TreeSet::new));
    
            // validate the non-response params
            if (!unconsumedParams.isEmpty()) {
                final Set<String> candidateParams = new HashSet<>();
                candidateParams.addAll(request.consumedParams());
                candidateParams.addAll(responseParams());
                throw new IllegalArgumentException(unrecognized(request, unconsumedParams, candidateParams, "parameter"));
            }
    
            if (request.hasContent() && request.isContentConsumed() == false) {
                throw new IllegalArgumentException("request [" + request.method() + " " + request.path() + "] does not support having a body");
            }
    
            usageCount.increment();
            // execute the action
            // 即此處僅為調用前面設置好的方法
            action.accept(channel);
        }

      而處理get的處理器,我們實際上可以通過在最初注冊的時候,可以看到是 RestGetAction, 它實現的 prepareRequest() 體現了其處理方法。

        // org.elasticsearch.rest.action.document.RestGetAction#prepareRequest
        @Override
        public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
            GetRequest getRequest;
            if (request.hasParam("type")) {
                deprecationLogger.deprecate("get_with_types", TYPES_DEPRECATION_MESSAGE);
                getRequest = new GetRequest(request.param("index"), request.param("type"), request.param("id"));
            } else {
                getRequest = new GetRequest(request.param("index"), request.param("id"));
            }
    
            getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh()));
            getRequest.routing(request.param("routing"));
            getRequest.preference(request.param("preference"));
            getRequest.realtime(request.paramAsBoolean("realtime", getRequest.realtime()));
            if (request.param("fields") != null) {
                throw new IllegalArgumentException("the parameter [fields] is no longer supported, " +
                    "please use [stored_fields] to retrieve stored fields or [_source] to load the field from _source");
            }
            final String fieldsParam = request.param("stored_fields");
            if (fieldsParam != null) {
                final String[] fields = Strings.splitStringByCommaToArray(fieldsParam);
                if (fields != null) {
                    getRequest.storedFields(fields);
                }
            }
    
            getRequest.version(RestActions.parseVersion(request));
            getRequest.versionType(VersionType.fromString(request.param("version_type"), getRequest.versionType()));
    
            getRequest.fetchSourceContext(FetchSourceContext.parseFromRestRequest(request));
            // 封裝具體業務處理方法
            // 交由 NodeClient 處理
            return channel -> client.get(getRequest, new RestToXContentListener<GetResponse>(channel) {
                @Override
                protected RestStatus getStatus(final GetResponse response) {
                    return response.isExists() ? OK : NOT_FOUND;
                }
            });
        }
        // org.elasticsearch.search.fetch.subphase.FetchSourceContext#parseFromRestRequest
        public static FetchSourceContext parseFromRestRequest(RestRequest request) {
            Boolean fetchSource = null;
            String[] sourceExcludes = null;
            String[] sourceIncludes = null;
    
            String source = request.param("_source");
            if (source != null) {
                if (Booleans.isTrue(source)) {
                    fetchSource = true;
                } else if (Booleans.isFalse(source)) {
                    fetchSource = false;
                } else {
                    sourceIncludes = Strings.splitStringByCommaToArray(source);
                }
            }
    
            String sIncludes = request.param("_source_includes");
            if (sIncludes != null) {
                sourceIncludes = Strings.splitStringByCommaToArray(sIncludes);
            }
    
            String sExcludes = request.param("_source_excludes");
            if (sExcludes != null) {
                sourceExcludes = Strings.splitStringByCommaToArray(sExcludes);
            }
    
            if (fetchSource != null || sourceIncludes != null || sourceExcludes != null) {
                return new FetchSourceContext(fetchSource == null ? true : fetchSource, sourceIncludes, sourceExcludes);
            }
            return null;
        }

      上面就是get的處理實現前奏,可以看出其支持的參數:type, id, refresh, routing, preference, fields, stored_fields, version_type, _source, _source_includes... 果然還是有點復雜,選擇越多,越麻煩。

      其中,url的請求方式,我們可以從es的route信息中看到:

        // org.elasticsearch.rest.action.document.RestGetAction#routes
        @Override
        public List<Route> routes() {
            return unmodifiableList(asList(
                new Route(GET, "/{index}/_doc/{id}"),
                new Route(HEAD, "/{index}/_doc/{id}"),
                // Deprecated typed endpoints.
                new Route(GET, "/{index}/{type}/{id}"),
                new Route(HEAD, "/{index}/{type}/{id}")));
        }

      最終,會得到一個 TransportGetAction 的內部處理器。

        // org.elasticsearch.action.support.TransportAction#execute
        /**
         * Use this method when the transport action call should result in creation of a new task associated with the call.
         *
         * This is a typical behavior.
         */
        public final Task execute(Request request, ActionListener<Response> listener) {
            /*
             * While this version of execute could delegate to the TaskListener
             * version of execute that'd add yet another layer of wrapping on the
             * listener and prevent us from using the listener bare if there isn't a
             * task. That just seems like too many objects. Thus the two versions of
             * this method.
             */
            final Releasable unregisterChildNode = registerChildNode(request.getParentTask());
            final Task task;
            try {
                task = taskManager.register("transport", actionName, request);
            } catch (TaskCancelledException e) {
                unregisterChildNode.close();
                throw e;
            }
            execute(task, request, new ActionListener<Response>() {
                @Override
                public void onResponse(Response response) {
                    try {
                        Releasables.close(unregisterChildNode, () -> taskManager.unregister(task));
                    } finally {
                        listener.onResponse(response);
                    }
                }
    
                @Override
                public void onFailure(Exception e) {
                    try {
                        Releasables.close(unregisterChildNode, () -> taskManager.unregister(task));
                    } finally {
                        listener.onFailure(e);
                    }
                }
            });
            return task;
        }
    
        /**
         * Use this method when the transport action should continue to run in the context of the current task
         */
        public final void execute(Task task, Request request, ActionListener<Response> listener) {
            ActionRequestValidationException validationException = request.validate();
            if (validationException != null) {
                listener.onFailure(validationException);
                return;
            }
    
            if (task != null && request.getShouldStoreResult()) {
                listener = new TaskResultStoringActionListener<>(taskManager, task, listener);
            }
    
            RequestFilterChain<Request, Response> requestFilterChain = new RequestFilterChain<>(this, logger);
            requestFilterChain.proceed(task, actionName, request, listener);
        }

      TransportGetAction 繼承了 TransportSingleShardAction 繼承了 TransportAction . , 所以 doExecute 調用了 TransportSingleShardAction 的方法。

        // org.elasticsearch.action.support.single.shard.TransportSingleShardAction#doExecute
        @Override
        protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
            new AsyncSingleAction(request, listener).start();
        }
            // org.elasticsearch.action.support.single.shard.TransportSingleShardAction.AsyncSingleAction
            private AsyncSingleAction(Request request, ActionListener<Response> listener) {
                this.listener = listener;
    
                ClusterState clusterState = clusterService.state();
                if (logger.isTraceEnabled()) {
                    logger.trace("executing [{}] based on cluster state version [{}]", request, clusterState.version());
                }
                nodes = clusterState.nodes();
                ClusterBlockException blockException = checkGlobalBlock(clusterState);
                if (blockException != null) {
                    throw blockException;
                }
    
                String concreteSingleIndex;
                // 解析 index 參數
                if (resolveIndex(request)) {
                    concreteSingleIndex = indexNameExpressionResolver.concreteSingleIndex(clusterState, request).getName();
                } else {
                    concreteSingleIndex = request.index();
                }
                // 組裝內部請求
                this.internalRequest = new InternalRequest(request, concreteSingleIndex);
                resolveRequest(clusterState, internalRequest);
                // 再次檢測集群阻塞狀態
                blockException = checkRequestBlock(clusterState, internalRequest);
                if (blockException != null) {
                    throw blockException;
                }
                // 獲取所有分片信息,向集群發起請求
                this.shardIt = shards(clusterState, internalRequest);
            }
        // org.elasticsearch.cluster.metadata.IndexNameExpressionResolver#concreteSingleIndex
        /**
         * Utility method that allows to resolve an index expression to its corresponding single concrete index.
         * Callers should make sure they provide proper {@link org.elasticsearch.action.support.IndicesOptions}
         * that require a single index as a result. The indices resolution must in fact return a single index when
         * using this method, an {@link IllegalArgumentException} gets thrown otherwise.
         *
         * @param state             the cluster state containing all the data to resolve to expression to a concrete index
         * @param request           The request that defines how the an alias or an index need to be resolved to a concrete index
         *                          and the expression that can be resolved to an alias or an index name.
         * @throws IllegalArgumentException if the index resolution lead to more than one index
         * @return the concrete index obtained as a result of the index resolution
         */
        public Index concreteSingleIndex(ClusterState state, IndicesRequest request) {
            String indexExpression = CollectionUtils.isEmpty(request.indices()) ? null : request.indices()[0];
            Index[] indices = concreteIndices(state, request.indicesOptions(), indexExpression);
            if (indices.length != 1) {
                throw new IllegalArgumentException("unable to return a single index as the index and options" +
                    " provided got resolved to multiple indices");
            }
            return indices[0];
        }
    
        // org.elasticsearch.cluster.metadata.IndexNameExpressionResolver#concreteIndices
        /**
         * Translates the provided index expression into actual concrete indices, properly deduplicated.
         *
         * @param state             the cluster state containing all the data to resolve to expressions to concrete indices
         * @param options           defines how the aliases or indices need to be resolved to concrete indices
         * @param indexExpressions  expressions that can be resolved to alias or index names.
         * @return the resolved concrete indices based on the cluster state, indices options and index expressions
         * @throws IndexNotFoundException if one of the index expressions is pointing to a missing index or alias and the
         * provided indices options in the context don't allow such a case, or if the final result of the indices resolution
         * contains no indices and the indices options in the context don't allow such a case.
         * @throws IllegalArgumentException if one of the aliases resolve to multiple indices and the provided
         * indices options in the context don't allow such a case.
         */
        public Index[] concreteIndices(ClusterState state, IndicesOptions options, String... indexExpressions) {
            return concreteIndices(state, options, false, indexExpressions);
        }
    
        public Index[] concreteIndices(ClusterState state, IndicesOptions options, boolean includeDataStreams, String... indexExpressions) {
            Context context = new Context(state, options, false, false, includeDataStreams,
                isSystemIndexAccessAllowed());
            return concreteIndices(context, indexExpressions);
        }
    
        Index[] concreteIndices(Context context, String... indexExpressions) {
            if (indexExpressions == null || indexExpressions.length == 0) {
                indexExpressions = new String[]{Metadata.ALL};
            }
            Metadata metadata = context.getState().metadata();
            IndicesOptions options = context.getOptions();
            // If only one index is specified then whether we fail a request if an index is missing depends on the allow_no_indices
            // option. At some point we should change this, because there shouldn't be a reason why whether a single index
            // or multiple indices are specified yield different behaviour.
            final boolean failNoIndices = indexExpressions.length == 1 ? !options.allowNoIndices() : !options.ignoreUnavailable();
            List<String> expressions = Arrays.asList(indexExpressions);
            for (ExpressionResolver expressionResolver : expressionResolvers) {
                expressions = expressionResolver.resolve(context, expressions);
            }
    
            if (expressions.isEmpty()) {
                if (!options.allowNoIndices()) {
                    IndexNotFoundException infe;
                    if (indexExpressions.length == 1) {
                        if (indexExpressions[0].equals(Metadata.ALL)) {
                            infe = new IndexNotFoundException("no indices exist", (String)null);
                        } else {
                            infe = new IndexNotFoundException((String)null);
                        }
                    } else {
                        infe = new IndexNotFoundException((String)null);
                    }
                    infe.setResources("index_expression", indexExpressions);
                    throw infe;
                } else {
                    return Index.EMPTY_ARRAY;
                }
            }
    
            boolean excludedDataStreams = false;
            final Set<Index> concreteIndices = new LinkedHashSet<>(expressions.size());
            for (String expression : expressions) {
                IndexAbstraction indexAbstraction = metadata.getIndicesLookup().get(expression);
                if (indexAbstraction == null ) {
                    if (failNoIndices) {
                        IndexNotFoundException infe;
                        if (expression.equals(Metadata.ALL)) {
                            infe = new IndexNotFoundException("no indices exist", expression);
                        } else {
                            infe = new IndexNotFoundException(expression);
                        }
                        infe.setResources("index_expression", expression);
                        throw infe;
                    } else {
                        continue;
                    }
                } else if (indexAbstraction.getType() == IndexAbstraction.Type.ALIAS && context.getOptions().ignoreAliases()) {
                    if (failNoIndices) {
                        throw aliasesNotSupportedException(expression);
                    } else {
                        continue;
                    }
                } else if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM &&
                            context.includeDataStreams() == false) {
                    excludedDataStreams = true;
                    continue;
                }
    
                if (indexAbstraction.getType() == IndexAbstraction.Type.ALIAS && context.isResolveToWriteIndex()) {
                    IndexMetadata writeIndex = indexAbstraction.getWriteIndex();
                    if (writeIndex == null) {
                        throw new IllegalArgumentException("no write index is defined for alias [" + indexAbstraction.getName() + "]." +
                            " The write index may be explicitly disabled using is_write_index=false or the alias points to multiple" +
                            " indices without one being designated as a write index");
                    }
                    if (addIndex(writeIndex, context)) {
                        concreteIndices.add(writeIndex.getIndex());
                    }
                } else if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM && context.isResolveToWriteIndex()) {
                    IndexMetadata writeIndex = indexAbstraction.getWriteIndex();
                    if (addIndex(writeIndex, context)) {
                        concreteIndices.add(writeIndex.getIndex());
                    }
                } else {
                    if (indexAbstraction.getIndices().size() > 1 && !options.allowAliasesToMultipleIndices()) {
                        String[] indexNames = new String[indexAbstraction.getIndices().size()];
                        int i = 0;
                        for (IndexMetadata indexMetadata : indexAbstraction.getIndices()) {
                            indexNames[i++] = indexMetadata.getIndex().getName();
                        }
                        throw new IllegalArgumentException(indexAbstraction.getType().getDisplayName() + " [" + expression +
                            "] has more than one index associated with it " + Arrays.toString(indexNames) +
                            ", can't execute a single index op");
                    }
    
                    for (IndexMetadata index : indexAbstraction.getIndices()) {
                        if (shouldTrackConcreteIndex(context, options, index)) {
                            concreteIndices.add(index.getIndex());
                        }
                    }
                }
            }
    
            if (options.allowNoIndices() == false && concreteIndices.isEmpty()) {
                IndexNotFoundException infe = new IndexNotFoundException((String)null);
                infe.setResources("index_expression", indexExpressions);
                if (excludedDataStreams) {
                    // Allows callers to handle IndexNotFoundException differently based on whether data streams were excluded.
                    infe.addMetadata(EXCLUDED_DATA_STREAMS_KEY, "true");
                }
                throw infe;
            }
            checkSystemIndexAccess(context, metadata, concreteIndices, indexExpressions);
            return concreteIndices.toArray(new Index[concreteIndices.size()]);
        }
    
        // org.elasticsearch.action.get.TransportGetAction#resolveRequest
        @Override
        protected void resolveRequest(ClusterState state, InternalRequest request) {
            // update the routing (request#index here is possibly an alias)
            request.request().routing(state.metadata().resolveIndexRouting(request.request().routing(), request.request().index()));
            // Fail fast on the node that received the request.
            if (request.request().routing() == null && state.getMetadata().routingRequired(request.concreteIndex())) {
                throw new RoutingMissingException(request.concreteIndex(), request.request().type(), request.request().id());
            }
        }
    
        @Override
        protected ShardIterator shards(ClusterState state, InternalRequest request) {
            return clusterService.operationRouting()
                    .getShards(clusterService.state(), request.concreteIndex(), request.request().id(), request.request().routing(),
                        request.request().preference());
        }
    
            // org.elasticsearch.action.support.single.shard.TransportSingleShardAction.AsyncSingleAction#start
            public void start() {
                if (shardIt == null) {
                    // just execute it on the local node
                    final Writeable.Reader<Response> reader = getResponseReader();
                    transportService.sendRequest(clusterService.localNode(), transportShardAction, internalRequest.request(),
                        new TransportResponseHandler<Response>() {
                        @Override
                        public Response read(StreamInput in) throws IOException {
                            return reader.read(in);
                        }
    
                        @Override
                        public void handleResponse(final Response response) {
                            listener.onResponse(response);
                        }
    
                        @Override
                        public void handleException(TransportException exp) {
                            listener.onFailure(exp);
                        }
                    });
                } else {
                    perform(null);
                }
            }

     

            private void perform(@Nullable final Exception currentFailure) {
                Exception lastFailure = this.lastFailure;
                if (lastFailure == null || TransportActions.isReadOverrideException(currentFailure)) {
                    lastFailure = currentFailure;
                    this.lastFailure = currentFailure;
                }
                final ShardRouting shardRouting = shardIt.nextOrNull();
                if (shardRouting == null) {
                    Exception failure = lastFailure;
                    if (failure == null || isShardNotAvailableException(failure)) {
                        failure = new NoShardAvailableActionException(null,
                            LoggerMessageFormat.format("No shard available for [{}]", internalRequest.request()), failure);
                    } else {
                        logger.debug(() -> new ParameterizedMessage("{}: failed to execute [{}]", null,
                            internalRequest.request()), failure);
                    }
                    listener.onFailure(failure);
                    return;
                }
                DiscoveryNode node = nodes.get(shardRouting.currentNodeId());
                if (node == null) {
                    onFailure(shardRouting, new NoShardAvailableActionException(shardRouting.shardId()));
                } else {
                    internalRequest.request().internalShardId = shardRouting.shardId();
                    if (logger.isTraceEnabled()) {
                        logger.trace(
                                "sending request [{}] to shard [{}] on node [{}]",
                                internalRequest.request(),
                                internalRequest.request().internalShardId,
                                node
                        );
                    }
                    final Writeable.Reader<Response> reader = getResponseReader();
                    transportService.sendRequest(node, transportShardAction, internalRequest.request(),
                        new TransportResponseHandler<Response>() {
    
                            @Override
                            public Response read(StreamInput in) throws IOException {
                                return reader.read(in);
                            }
    
                            @Override
                            public void handleResponse(final Response response) {
                                listener.onResponse(response);
                            }
    
                            @Override
                            public void handleException(TransportException exp) {
                                onFailure(shardRouting, exp);
                            }
                    });
                }
            }

       sendRequest, 本意是向外部網絡發起請求,得到結果的過程。但es針對本節點的請求當然不會真正向外請求,我們以請求本節點的時序圖,來看看es處理get的sendRequest過程吧。

    sendRequest 調用時序圖

     

     

      具體代碼實現如下:

        // org.elasticsearch.transport.TransportService#sendRequest
        public <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
                                                                    final TransportRequest request,
                                                                    final TransportResponseHandler<T> handler) {
            sendRequest(node, action, request, TransportRequestOptions.EMPTY, handler);
        }
    
        public final <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
                                                                    final TransportRequest request,
                                                                    final TransportRequestOptions options,
                                                                    TransportResponseHandler<T> handler) {
            final Transport.Connection connection;
            try {
                connection = getConnection(node);
            } catch (final NodeNotConnectedException ex) {
                // the caller might not handle this so we invoke the handler
                handler.handleException(ex);
                return;
            }
            sendRequest(connection, action, request, options, handler);
        }
    
        /**
         * Returns either a real transport connection or a local node connection if we are using the local node optimization.
         * @throws NodeNotConnectedException if the given node is not connected
         */
        public Transport.Connection getConnection(DiscoveryNode node) {
            // 如果是當前節點,直接返回當前service即可,無須再訪問遠程節點
            if (isLocalNode(node)) {
                return localNodeConnection;
            } else {
                return connectionManager.getConnection(node);
            }
        }
        private boolean isLocalNode(DiscoveryNode discoveryNode) {
            return Objects.requireNonNull(discoveryNode, "discovery node must not be null").equals(localNode);
        }
    
        // org.elasticsearch.transport.TransportService#sendRequest
        /**
         * Sends a request on the specified connection. If there is a failure sending the request, the specified handler is invoked.
         *
         * @param connection the connection to send the request on
         * @param action     the name of the action
         * @param request    the request
         * @param options    the options for this request
         * @param handler    the response handler
         * @param <T>        the type of the transport response
         */
        public final <T extends TransportResponse> void sendRequest(final Transport.Connection connection, final String action,
                                                                    final TransportRequest request,
                                                                    final TransportRequestOptions options,
                                                                    final TransportResponseHandler<T> handler) {
            try {
                final TransportResponseHandler<T> delegate;
                if (request.getParentTask().isSet()) {
                    // If the connection is a proxy connection, then we will create a cancellable proxy task on the proxy node and an actual
                    // child task on the target node of the remote cluster.
                    //  ----> a parent task on the local cluster
                    //        |
                    //         ----> a proxy task on the proxy node on the remote cluster
                    //               |
                    //                ----> an actual child task on the target node on the remote cluster
                    // To cancel the child task on the remote cluster, we must send a cancel request to the proxy node instead of the target
                    // node as the parent task of the child task is the proxy task not the parent task on the local cluster. Hence, here we
                    // unwrap the connection and keep track of the connection to the proxy node instead of the proxy connection.
                    final Transport.Connection unwrappedConn = unwrapConnection(connection);
                    final Releasable unregisterChildNode = taskManager.registerChildConnection(request.getParentTask().getId(), unwrappedConn);
                    delegate = new TransportResponseHandler<T>() {
                        @Override
                        public void handleResponse(T response) {
                            unregisterChildNode.close();
                            handler.handleResponse(response);
                        }
    
                        @Override
                        public void handleException(TransportException exp) {
                            unregisterChildNode.close();
                            handler.handleException(exp);
                        }
    
                        @Override
                        public String executor() {
                            return handler.executor();
                        }
    
                        @Override
                        public T read(StreamInput in) throws IOException {
                            return handler.read(in);
                        }
    
                        @Override
                        public String toString() {
                            return getClass().getName() + "/[" + action + "]:" + handler.toString();
                        }
                    };
                } else {
                    delegate = handler;
                }
                asyncSender.sendRequest(connection, action, request, options, delegate);
            } catch (final Exception ex) {
                // the caller might not handle this so we invoke the handler
                final TransportException te;
                if (ex instanceof TransportException) {
                    te = (TransportException) ex;
                } else {
                    te = new TransportException("failure to send", ex);
                }
                handler.handleException(te);
            }
        }
    
        // org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor#interceptSender
        @Override
        public AsyncSender interceptSender(AsyncSender sender) {
            return new AsyncSender() {
                @Override
                public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
                                                                      TransportRequestOptions options, TransportResponseHandler<T> handler) {
                    final boolean requireAuth = shouldRequireExistingAuthentication();
                    // the transport in core normally does this check, BUT since we are serializing to a string header we need to do it
                    // ourselves otherwise we wind up using a version newer than what we can actually send
                    final Version minVersion = Version.min(connection.getVersion(), Version.CURRENT);
    
                    // Sometimes a system action gets executed like a internal create index request or update mappings request
                    // which means that the user is copied over to system actions so we need to change the user
                    if (AuthorizationUtils.shouldReplaceUserWithSystem(threadPool.getThreadContext(), action)) {
                        securityContext.executeAsUser(SystemUser.INSTANCE, (original) -> sendWithUser(connection, action, request, options,
                                new ContextRestoreResponseHandler<>(threadPool.getThreadContext().wrapRestorable(original)
                                        , handler), sender, requireAuth), minVersion);
                    } else if (AuthorizationUtils.shouldSetUserBasedOnActionOrigin(threadPool.getThreadContext())) {
                        AuthorizationUtils.switchUserBasedOnActionOriginAndExecute(threadPool.getThreadContext(), securityContext,
                                (original) -> sendWithUser(connection, action, request, options,
                                        new ContextRestoreResponseHandler<>(threadPool.getThreadContext().wrapRestorable(original)
                                                , handler), sender, requireAuth));
                    } else if (securityContext.getAuthentication() != null &&
                            securityContext.getAuthentication().getVersion().equals(minVersion) == false) {
                        // re-write the authentication since we want the authentication version to match the version of the connection
                        securityContext.executeAfterRewritingAuthentication(original -> sendWithUser(connection, action, request, options,
                            new ContextRestoreResponseHandler<>(threadPool.getThreadContext().wrapRestorable(original), handler), sender,
                            requireAuth), minVersion);
                    } else {
                        sendWithUser(connection, action, request, options, handler, sender, requireAuth);
                    }
                }
            };
        }
        // org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor#sendWithUser
        private <T extends TransportResponse> void sendWithUser(Transport.Connection connection, String action, TransportRequest request,
                                                                TransportRequestOptions options, TransportResponseHandler<T> handler,
                                                                AsyncSender sender, final boolean requireAuthentication) {
            if (securityContext.getAuthentication() == null && requireAuthentication) {
                // we use an assertion here to ensure we catch this in our testing infrastructure, but leave the ISE for cases we do not catch
                // in tests and may be hit by a user
                assertNoAuthentication(action);
                throw new IllegalStateException("there should always be a user when sending a message for action [" + action + "]");
            }
    
            try {
                sender.sendRequest(connection, action, request, options, handler);
            } catch (Exception e) {
                handler.handleException(new TransportException("failed sending request", e));
            }
        }
        // org.elasticsearch.transport.TransportService#sendRequestInternal
        private <T extends TransportResponse> void sendRequestInternal(final Transport.Connection connection, final String action,
                                                                       final TransportRequest request,
                                                                       final TransportRequestOptions options,
                                                                       TransportResponseHandler<T> handler) {
            if (connection == null) {
                throw new IllegalStateException("can't send request to a null connection");
            }
            DiscoveryNode node = connection.getNode();
    
            Supplier<ThreadContext.StoredContext> storedContextSupplier = threadPool.getThreadContext().newRestorableContext(true);
            ContextRestoreResponseHandler<T> responseHandler = new ContextRestoreResponseHandler<>(storedContextSupplier, handler);
            // TODO we can probably fold this entire request ID dance into connection.sendReqeust but it will be a bigger refactoring
            final long requestId = responseHandlers.add(new Transport.ResponseContext<>(responseHandler, connection, action));
            final TimeoutHandler timeoutHandler;
            if (options.timeout() != null) {
                timeoutHandler = new TimeoutHandler(requestId, connection.getNode(), action);
                responseHandler.setTimeoutHandler(timeoutHandler);
            } else {
                timeoutHandler = null;
            }
            try {
                if (lifecycle.stoppedOrClosed()) {
                    /*
                     * If we are not started the exception handling will remove the request holder again and calls the handler to notify the
                     * caller. It will only notify if toStop hasn't done the work yet.
                     */
                    throw new NodeClosedException(localNode);
                }
                if (timeoutHandler != null) {
                    assert options.timeout() != null;
                    timeoutHandler.scheduleTimeout(options.timeout());
                }
                // 請求發送
                connection.sendRequest(requestId, action, request, options); // local node optimization happens upstream
            } catch (final Exception e) {
                // usually happen either because we failed to connect to the node
                // or because we failed serializing the message
                final Transport.ResponseContext<? extends TransportResponse> contextToNotify = responseHandlers.remove(requestId);
                // If holderToNotify == null then handler has already been taken care of.
                if (contextToNotify != null) {
                    if (timeoutHandler != null) {
                        timeoutHandler.cancel();
                    }
                    // callback that an exception happened, but on a different thread since we don't
                    // want handlers to worry about stack overflows. In the special case of running into a closing node we run on the current
                    // thread on a best effort basis though.
                    final SendRequestTransportException sendRequestException = new SendRequestTransportException(node, action, e);
                    final String executor = lifecycle.stoppedOrClosed() ? ThreadPool.Names.SAME : ThreadPool.Names.GENERIC;
                    threadPool.executor(executor).execute(new AbstractRunnable() {
                        @Override
                        public void onRejection(Exception e) {
                            // if we get rejected during node shutdown we don't wanna bubble it up
                            logger.debug(
                                () -> new ParameterizedMessage(
                                    "failed to notify response handler on rejection, action: {}",
                                    contextToNotify.action()),
                                e);
                        }
                        @Override
                        public void onFailure(Exception e) {
                            logger.warn(
                                () -> new ParameterizedMessage(
                                    "failed to notify response handler on exception, action: {}",
                                    contextToNotify.action()),
                                e);
                        }
                        @Override
                        protected void doRun() throws Exception {
                            contextToNotify.handler().handleException(sendRequestException);
                        }
                    });
                } else {
                    logger.debug("Exception while sending request, handler likely already notified due to timeout", e);
                }
            }
        }
            // org.elasticsearch.transport.Transport.Connection#sendRequest
            @Override
            public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
                throws TransportException {
                sendLocalRequest(requestId, action, request, options);
            }
        // org.elasticsearch.transport.TransportService#sendLocalRequest
        private void sendLocalRequest(long requestId, final String action, final TransportRequest request, TransportRequestOptions options) {
            final DirectResponseChannel channel = new DirectResponseChannel(localNode, action, requestId, this, threadPool);
            try {
                onRequestSent(localNode, requestId, action, request, options);
                onRequestReceived(requestId, action);
                // 獲取 handler
                final RequestHandlerRegistry reg = getRequestHandler(action);
                if (reg == null) {
                    throw new ActionNotFoundTransportException("Action [" + action + "] not found");
                }
                final String executor = reg.getExecutor();
                if (ThreadPool.Names.SAME.equals(executor)) {
                    //noinspection unchecked
                    reg.processMessageReceived(request, channel);
                } else {
                    threadPool.executor(executor).execute(new AbstractRunnable() {
                        @Override
                        protected void doRun() throws Exception {
                            //noinspection unchecked
                            reg.processMessageReceived(request, channel);
                        }
    
                        @Override
                        public boolean isForceExecution() {
                            return reg.isForceExecution();
                        }
    
                        @Override
                        public void onFailure(Exception e) {
                            try {
                                channel.sendResponse(e);
                            } catch (Exception inner) {
                                inner.addSuppressed(e);
                                logger.warn(() -> new ParameterizedMessage(
                                        "failed to notify channel of error message for action [{}]", action), inner);
                            }
                        }
    
                        @Override
                        public String toString() {
                            return "processing of [" + requestId + "][" + action + "]: " + request;
                        }
                    });
                }
    
            } catch (Exception e) {
                try {
                    channel.sendResponse(e);
                } catch (Exception inner) {
                    inner.addSuppressed(e);
                    logger.warn(
                        () -> new ParameterizedMessage(
                            "failed to notify channel of error message for action [{}]", action), inner);
                }
            }
        }
        // org.elasticsearch.transport.RequestHandlerRegistry#processMessageReceived
        public void processMessageReceived(Request request, TransportChannel channel) throws Exception {
            final Task task = taskManager.register(channel.getChannelType(), action, request);
            Releasable unregisterTask = () -> taskManager.unregister(task);
            try {
                if (channel instanceof TcpTransportChannel && task instanceof CancellableTask) {
                    final TcpChannel tcpChannel = ((TcpTransportChannel) channel).getChannel();
                    final Releasable stopTracking = taskManager.startTrackingCancellableChannelTask(tcpChannel, (CancellableTask) task);
                    unregisterTask = Releasables.wrap(unregisterTask, stopTracking);
                }
                final TaskTransportChannel taskTransportChannel = new TaskTransportChannel(channel, unregisterTask);
                handler.messageReceived(request, taskTransportChannel, task);
                unregisterTask = null;
            } finally {
                Releasables.close(unregisterTask);
            }
        }
        // org.elasticsearch.action.get.TransportGetAction#asyncShardOperation
        @Override
        protected void asyncShardOperation(GetRequest request, ShardId shardId, ActionListener<GetResponse> listener) throws IOException {
            IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
            IndexShard indexShard = indexService.getShard(shardId.id());
            if (request.realtime()) { // we are not tied to a refresh cycle here anyway
                super.asyncShardOperation(request, shardId, listener);
            } else {
                indexShard.awaitShardSearchActive(b -> {
                    try {
                        super.asyncShardOperation(request, shardId, listener);
                    } catch (Exception ex) {
                        listener.onFailure(ex);
                    }
                });
            }
        }
        // org.elasticsearch.action.support.single.shard.TransportSingleShardAction#asyncShardOperation
        protected void asyncShardOperation(Request request, ShardId shardId, ActionListener<Response> listener) throws IOException {
            threadPool.executor(getExecutor(request, shardId))
                .execute(ActionRunnable.supply(listener, () -> shardOperation(request, shardId)));
        }
    
        // org.elasticsearch.action.get.TransportGetAction#shardOperation
        // 該方法為實現get操作的真正方法
        @Override
        protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
            IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
            IndexShard indexShard = indexService.getShard(shardId.id());
    
            if (request.refresh() && !request.realtime()) {
                indexShard.refresh("refresh_flag_get");
            }
            // 核心 get 實現
            GetResult result = indexShard.getService().get(request.type(), request.id(), request.storedFields(),
                    request.realtime(), request.version(), request.versionType(), request.fetchSourceContext());
            // 使用 GetResponse 包裝結果返回,方便結果響應輸出
            return new GetResponse(result);
        }

      以上,是 get 的處理框架,基本原理就是先看是否有路由配置,如果有按其規則來,如果沒有直接向某個節點發起請求,即可獲取到數據。整個get請求前置的轉發過程,因為考慮大量的遠程調用與復用,顯示比較復雜。其中,es中大量使用了包裝器模式和觀察者模式,值得我們注意和學習。到最后最核心的get的實現,其實是 shardOperation() 方法。它又是如何與lucene交互獲取數據的呢?稍后見分曉。

      路由是es的一個我特性,而shard則是一個核心概念。接下來,我們看兩個細節,即 es get 如何處理路由 和es get 如何處理shard問題。

        // 處理路由問題, 請求方式如: curl -XGET 'http://localhost:9200/test_index/job/1?routing=user123'
        // org.elasticsearch.action.get.TransportGetAction#resolveRequest
        @Override
        protected void resolveRequest(ClusterState state, InternalRequest request) {
            // update the routing (request#index here is possibly an alias)
            // 取出默認的 routing 規則,或者校驗傳入的routing規則
            request.request().routing(state.metadata().resolveIndexRouting(request.request().routing(), request.request().index()));
            // Fail fast on the node that received the request.
            if (request.request().routing() == null && state.getMetadata().routingRequired(request.concreteIndex())) {
                throw new RoutingMissingException(request.concreteIndex(), request.request().type(), request.request().id());
            }
        }
        // org.elasticsearch.cluster.metadata.Metadata#resolveIndexRouting
        /**
         * Returns indexing routing for the given index.
         */
        // TODO: This can be moved to IndexNameExpressionResolver too, but this means that we will support wildcards and other expressions
        // in the index,bulk,update and delete apis.
        public String resolveIndexRouting(@Nullable String routing, String aliasOrIndex) {
            if (aliasOrIndex == null) {
                return routing;
            }
            // 獲取索引對應的元數據信息,用于判定是否是別名型索引
            // 非別名型索引,routing 不變
            IndexAbstraction result = getIndicesLookup().get(aliasOrIndex);
            if (result == null || result.getType() != IndexAbstraction.Type.ALIAS) {
                return routing;
            }
            IndexAbstraction.Alias alias = (IndexAbstraction.Alias) result;
            if (result.getIndices().size() > 1) {
                rejectSingleIndexOperation(aliasOrIndex, result);
            }
            AliasMetadata aliasMd = alias.getFirstAliasMetadata();
            if (aliasMd.indexRouting() != null) {
                // 一個 alias 不允許有多個 routing key
                if (aliasMd.indexRouting().indexOf(',') != -1) {
                    throw new IllegalArgumentException("index/alias [" + aliasOrIndex + "] provided with routing value [" +
                        aliasMd.getIndexRouting() + "] that resolved to several routing values, rejecting operation");
                }
                if (routing != null) {
                    if (!routing.equals(aliasMd.indexRouting())) {
                        throw new IllegalArgumentException("Alias [" + aliasOrIndex + "] has index routing associated with it [" +
                            aliasMd.indexRouting() + "], and was provided with routing value [" + routing + "], rejecting operation");
                    }
                }
                // Alias routing overrides the parent routing (if any).
                return aliasMd.indexRouting();
            }
            return routing;
        }
    
        // 處理shard問題
        // org.elasticsearch.action.get.TransportGetAction#shards
        @Override
        protected ShardIterator shards(ClusterState state, InternalRequest request) {
            return clusterService.operationRouting()
                    .getShards(clusterService.state(), request.concreteIndex(), request.request().id(), request.request().routing(),
                        request.request().preference());
        }
        // org.elasticsearch.cluster.routing.OperationRouting#getShards
        public ShardIterator getShards(ClusterState clusterState, String index, String id, @Nullable String routing,
                                       @Nullable String preference) {
            // 先獲取shardingTable, 再生成Shard迭代器
            return preferenceActiveShardIterator(shards(clusterState, index, id, routing), clusterState.nodes().getLocalNodeId(),
                clusterState.nodes(), preference, null, null);
        }
        // org.elasticsearch.cluster.routing.OperationRouting#shards
        protected IndexShardRoutingTable shards(ClusterState clusterState, String index, String id, String routing) {
            int shardId = generateShardId(indexMetadata(clusterState, index), id, routing);
            return clusterState.getRoutingTable().shardRoutingTable(index, shardId);
        }
        // org.elasticsearch.cluster.routing.OperationRouting#indexMetadata
        protected IndexMetadata indexMetadata(ClusterState clusterState, String index) {
            IndexMetadata indexMetadata = clusterState.metadata().index(index);
            if (indexMetadata == null) {
                throw new IndexNotFoundException(index);
            }
            return indexMetadata;
        }
        // 生成shardId
        // org.elasticsearch.cluster.routing.OperationRouting#generateShardId
        public static int generateShardId(IndexMetadata indexMetadata, @Nullable String id, @Nullable String routing) {
            final String effectiveRouting;
            final int partitionOffset;
    
            if (routing == null) {
                assert(indexMetadata.isRoutingPartitionedIndex() == false) : "A routing value is required for gets from a partitioned index";
                effectiveRouting = id;
            } else {
                effectiveRouting = routing;
            }
    
            if (indexMetadata.isRoutingPartitionedIndex()) {
                partitionOffset = Math.floorMod(Murmur3HashFunction.hash(id), indexMetadata.getRoutingPartitionSize());
            } else {
                // we would have still got 0 above but this check just saves us an unnecessary hash calculation
                partitionOffset = 0;
            }
    
            return calculateScaledShardId(indexMetadata, effectiveRouting, partitionOffset);
        }
        // org.elasticsearch.cluster.routing.OperationRouting#calculateScaledShardId
        private static int calculateScaledShardId(IndexMetadata indexMetadata, String effectiveRouting, int partitionOffset) {
            final int hash = Murmur3HashFunction.hash(effectiveRouting) + partitionOffset;
    
            // we don't use IMD#getNumberOfShards since the index might have been shrunk such that we need to use the size
            // of original index to hash documents
            // routingFactor默認為1
            return Math.floorMod(hash, indexMetadata.getRoutingNumShards()) / indexMetadata.getRoutingFactor();
        }
        // org.elasticsearch.cluster.routing.RoutingTable#shardRoutingTable(java.lang.String, int)
        /**
         * All shards for the provided index and shard id
         * @return All the shard routing entries for the given index and shard id
         * @throws IndexNotFoundException if provided index does not exist
         * @throws ShardNotFoundException if provided shard id is unknown
         */
        public IndexShardRoutingTable shardRoutingTable(String index, int shardId) {
            IndexRoutingTable indexRouting = index(index);
            if (indexRouting == null) {
                throw new IndexNotFoundException(index);
            }
            return shardRoutingTable(indexRouting, shardId);
        }
        // org.elasticsearch.cluster.routing.RoutingTable#shardRoutingTable
        /**
         * Get's the {@link IndexShardRoutingTable} for the given shard id from the given {@link IndexRoutingTable}
         * or throws a {@link ShardNotFoundException} if no shard by the given id is found in the IndexRoutingTable.
         *
         * @param indexRouting IndexRoutingTable
         * @param shardId ShardId
         * @return IndexShardRoutingTable
         */
        public static IndexShardRoutingTable shardRoutingTable(IndexRoutingTable indexRouting, int shardId) {
            IndexShardRoutingTable indexShard = indexRouting.shard(shardId);
            if (indexShard == null) {
                throw new ShardNotFoundException(new ShardId(indexRouting.getIndex(), shardId));
            }
            return indexShard;
        }
        // org.elasticsearch.cluster.routing.OperationRouting#preferenceActiveShardIterator
        private ShardIterator preferenceActiveShardIterator(IndexShardRoutingTable indexShard, String localNodeId,
                                                            DiscoveryNodes nodes, @Nullable String preference,
                                                            @Nullable ResponseCollectorService collectorService,
                                                            @Nullable Map<String, Long> nodeCounts) {
            if (preference == null || preference.isEmpty()) {
                return shardRoutings(indexShard, nodes, collectorService, nodeCounts);
            }
            if (preference.charAt(0) == '_') {
                Preference preferenceType = Preference.parse(preference);
                if (preferenceType == Preference.SHARDS) {
                    // starts with _shards, so execute on specific ones
                    int index = preference.indexOf('|');
    
                    String shards;
                    if (index == -1) {
                        shards = preference.substring(Preference.SHARDS.type().length() + 1);
                    } else {
                        shards = preference.substring(Preference.SHARDS.type().length() + 1, index);
                    }
                    String[] ids = Strings.splitStringByCommaToArray(shards);
                    boolean found = false;
                    for (String id : ids) {
                        if (Integer.parseInt(id) == indexShard.shardId().id()) {
                            found = true;
                            break;
                        }
                    }
                    if (!found) {
                        return null;
                    }
                    // no more preference
                    if (index == -1 || index == preference.length() - 1) {
                        return shardRoutings(indexShard, nodes, collectorService, nodeCounts);
                    } else {
                        // update the preference and continue
                        preference = preference.substring(index + 1);
                    }
                }
                preferenceType = Preference.parse(preference);
                switch (preferenceType) {
                    case PREFER_NODES:
                        final Set<String> nodesIds =
                                Arrays.stream(
                                        preference.substring(Preference.PREFER_NODES.type().length() + 1).split(",")
                                ).collect(Collectors.toSet());
                        return indexShard.preferNodeActiveInitializingShardsIt(nodesIds);
                    case LOCAL:
                        return indexShard.preferNodeActiveInitializingShardsIt(Collections.singleton(localNodeId));
                    case ONLY_LOCAL:
                        return indexShard.onlyNodeActiveInitializingShardsIt(localNodeId);
                    case ONLY_NODES:
                        String nodeAttributes = preference.substring(Preference.ONLY_NODES.type().length() + 1);
                        return indexShard.onlyNodeSelectorActiveInitializingShardsIt(nodeAttributes.split(","), nodes);
                    default:
                        throw new IllegalArgumentException("unknown preference [" + preferenceType + "]");
                }
            }
            // if not, then use it as the index
            int routingHash = Murmur3HashFunction.hash(preference);
            if (nodes.getMinNodeVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
                // The AllocationService lists shards in a fixed order based on nodes
                // so earlier versions of this class would have a tendency to
                // select the same node across different shardIds.
                // Better overall balancing can be achieved if each shardId opts
                // for a different element in the list by also incorporating the
                // shard ID into the hash of the user-supplied preference key.
                routingHash = 31 * routingHash + indexShard.shardId.hashCode();
            }
            if (awarenessAttributes.isEmpty()) {
                return indexShard.activeInitializingShardsIt(routingHash);
            } else {
                return indexShard.preferAttributesActiveInitializingShardsIt(awarenessAttributes, nodes, routingHash);
            }
        }
    
        private ShardIterator shardRoutings(IndexShardRoutingTable indexShard, DiscoveryNodes nodes,
                @Nullable ResponseCollectorService collectorService, @Nullable Map<String, Long> nodeCounts) {
            if (awarenessAttributes.isEmpty()) {
                if (useAdaptiveReplicaSelection) {
                    return indexShard.activeInitializingShardsRankedIt(collectorService, nodeCounts);
                } else {
                    return indexShard.activeInitializingShardsRandomIt();
                }
            } else {
                return indexShard.preferAttributesActiveInitializingShardsIt(awarenessAttributes, nodes);
            }
        }
        // org.elasticsearch.cluster.routing.IndexShardRoutingTable#activeInitializingShardsRankedIt
        /**
         * Returns an iterator over active and initializing shards, ordered by the adaptive replica
         * selection formula. Making sure though that its random within the active shards of the same
         * (or missing) rank, and initializing shards are the last to iterate through.
         */
        public ShardIterator activeInitializingShardsRankedIt(@Nullable ResponseCollectorService collector,
                                                              @Nullable Map<String, Long> nodeSearchCounts) {
            final int seed = shuffler.nextSeed();
            if (allInitializingShards.isEmpty()) {
                return new PlainShardIterator(shardId,
                        rankShardsAndUpdateStats(shuffler.shuffle(activeShards, seed), collector, nodeSearchCounts));
            }
    
            ArrayList<ShardRouting> ordered = new ArrayList<>(activeShards.size() + allInitializingShards.size());
            List<ShardRouting> rankedActiveShards =
                    rankShardsAndUpdateStats(shuffler.shuffle(activeShards, seed), collector, nodeSearchCounts);
            ordered.addAll(rankedActiveShards);
            List<ShardRouting> rankedInitializingShards =
                    rankShardsAndUpdateStats(allInitializingShards, collector, nodeSearchCounts);
            ordered.addAll(rankedInitializingShards);
            return new PlainShardIterator(shardId, ordered);
        }
    View Code

     

    4. get的內部邏輯

      它是在 TransportGetAction 的 shardOperation() 中實現的同步調用:

      先來看個整體的調用時序圖:

    get查找核心時序圖

     

     

        // org.elasticsearch.action.get.TransportGetAction#shardOperation
        // 該方法為實現get操作的真正方法
        @Override
        protected GetResponse shardOperation(GetRequest request, ShardId shardId) {
            IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
            IndexShard indexShard = indexService.getShard(shardId.id());
    
            if (request.refresh() && !request.realtime()) {
                indexShard.refresh("refresh_flag_get");
            }
            // 核心 get 實現
            GetResult result = indexShard.getService().get(request.type(), request.id(), request.storedFields(),
                    request.realtime(), request.version(), request.versionType(), request.fetchSourceContext());
            return new GetResponse(result);
        }
        // org.elasticsearch.indices.IndicesService#indexServiceSafe
        /**
         * Returns an IndexService for the specified index if exists otherwise a {@link IndexNotFoundException} is thrown.
         */
        public IndexService indexServiceSafe(Index index) {
            IndexService indexService = indices.get(index.getUUID());
            if (indexService == null) {
                throw new IndexNotFoundException(index);
            }
            assert indexService.indexUUID().equals(index.getUUID()) : "uuid mismatch local: " + indexService.indexUUID() +
                " incoming: " + index.getUUID();
            return indexService;
        }
        // org.elasticsearch.index.IndexService#getShard
        /**
         * Return the shard with the provided id, or throw an exception if it doesn't exist.
         */
        public IndexShard getShard(int shardId) {
            IndexShard indexShard = getShardOrNull(shardId);
            if (indexShard == null) {
                throw new ShardNotFoundException(new ShardId(index(), shardId));
            }
            return indexShard;
        }
    
        // org.elasticsearch.index.shard.IndexShard#getService
        public ShardGetService getService() {
            return this.getService;
        }
    
        // org.elasticsearch.index.get.ShardGetService#get
        public GetResult get(String type, String id, String[] gFields, boolean realtime, long version,
                                VersionType versionType, FetchSourceContext fetchSourceContext) {
            return
                get(type, id, gFields, realtime, version, versionType, UNASSIGNED_SEQ_NO, UNASSIGNED_PRIMARY_TERM, fetchSourceContext);
        }
        // org.elasticsearch.index.get.ShardGetService#get
        private GetResult get(String type, String id, String[] gFields, boolean realtime, long version, VersionType versionType,
                              long ifSeqNo, long ifPrimaryTerm, FetchSourceContext fetchSourceContext) {
            currentMetric.inc();
            try {
                long now = System.nanoTime();
                GetResult getResult =
                    innerGet(type, id, gFields, realtime, version, versionType, ifSeqNo, ifPrimaryTerm, fetchSourceContext);
    
                if (getResult.isExists()) {
                    existsMetric.inc(System.nanoTime() - now);
                } else {
                    missingMetric.inc(System.nanoTime() - now);
                }
                return getResult;
            } finally {
                currentMetric.dec();
            }
        }
        // org.elasticsearch.index.get.ShardGetService#innerGet
        private GetResult innerGet(String type, String id, String[] gFields, boolean realtime, long version, VersionType versionType,
                                   long ifSeqNo, long ifPrimaryTerm, FetchSourceContext fetchSourceContext) {
            // 查詢字段信息校驗
            fetchSourceContext = normalizeFetchSourceContent(fetchSourceContext, gFields);
            // 支持無type搜索,即丟棄老版本的 type 字段, 一個索引下全是數據
            if (type == null || type.equals("_all")) {
                DocumentMapper mapper = mapperService.documentMapper();
                type = mapper == null ? null : mapper.type();
            }
    
            Engine.GetResult get = null;
            if (type != null) {
                Term uidTerm = new Term(IdFieldMapper.NAME, Uid.encodeId(id));
                // 鏈式構造一個 Get 實例,調用 get() 方法查找doc
                // 其中 indexShard 是前面根據shardId 推斷出來的, 也是基于一個doc只能在一個shard上的前提
                get = indexShard.get(new Engine.Get(realtime, realtime, type, id, uidTerm)
                    .version(version).versionType(versionType).setIfSeqNo(ifSeqNo).setIfPrimaryTerm(ifPrimaryTerm));
                assert get.isFromTranslog() == false || realtime : "should only read from translog if realtime enabled";
                if (get.exists() == false) {
                    get.close();
                }
            }
    
            if (get == null || get.exists() == false) {
                return new GetResult(shardId.getIndexName(), type, id, UNASSIGNED_SEQ_NO, UNASSIGNED_PRIMARY_TERM, -1, false, null, null, null);
            }
    
            try {
                // break between having loaded it from translog (so we only have _source), and having a document to load
                // 根據docId 查詢具體的doc信息
                return innerGetLoadFromStoredFields(type, id, gFields, fetchSourceContext, get, mapperService);
            } finally {
                get.close();
            }
        }
    
        /**
         * decides what needs to be done based on the request input and always returns a valid non-null FetchSourceContext
         */
        private FetchSourceContext normalizeFetchSourceContent(@Nullable FetchSourceContext context, @Nullable String[] gFields) {
            if (context != null) {
                return context;
            }
            if (gFields == null) {
                return FetchSourceContext.FETCH_SOURCE;
            }
            for (String field : gFields) {
                if (SourceFieldMapper.NAME.equals(field)) {
                    return FetchSourceContext.FETCH_SOURCE;
                }
            }
            return FetchSourceContext.DO_NOT_FETCH_SOURCE;
        }
        // org.elasticsearch.index.shard.IndexShard#get
        public Engine.GetResult get(Engine.Get get) {
            // 當前集群狀態判定, 是否可讀, 如果是 red 狀態則是不可讀的
            readAllowed();
            DocumentMapper mapper = mapperService.documentMapper();
            if (mapper == null || mapper.type().equals(mapperService.resolveDocumentType(get.type())) == false) {
                return GetResult.NOT_EXISTS;
            }
            // wrapSearcher 實現查詢結果封裝
            return getEngine().get(get, mapper, this::wrapSearcher);
        }
        // org.elasticsearch.index.shard.IndexShard#readAllowed
        public void readAllowed() throws IllegalIndexShardStateException {
            IndexShardState state = this.state; // one time volatile read
            if (readAllowedStates.contains(state) == false) {
                throw new IllegalIndexShardStateException(shardId, state, "operations only allowed when shard state is one of " +
                    readAllowedStates.toString());
            }
        }
        // org.elasticsearch.index.shard.IndexShard#getEngine
        Engine getEngine() {
            Engine engine = getEngineOrNull();
            if (engine == null) {
                throw new AlreadyClosedException("engine is closed");
            }
            return engine;
        }
        /**
         * NOTE: returns null if engine is not yet started (e.g. recovery phase 1, copying over index files, is still running), or if engine is
         * closed.
         */
        protected Engine getEngineOrNull() {
            return this.currentEngineReference.get();
        }
    
        private Engine.Searcher wrapSearcher(Engine.Searcher searcher) {
            assert ElasticsearchDirectoryReader.unwrap(searcher.getDirectoryReader())
                != null : "DirectoryReader must be an instance or ElasticsearchDirectoryReader";
            boolean success = false;
            try {
                final Engine.Searcher newSearcher = readerWrapper == null ? searcher : wrapSearcher(searcher, readerWrapper);
                assert newSearcher != null;
                success = true;
                return newSearcher;
            } catch (IOException ex) {
                throw new ElasticsearchException("failed to wrap searcher", ex);
            } finally {
                if (success == false) {
                    Releasables.close(success, searcher);
                }
            }
        }
    
    
        // org.elasticsearch.index.engine.InternalEngine#get
        @Override
        public GetResult get(Get get, DocumentMapper mapper, Function<Engine.Searcher, Engine.Searcher> searcherWrapper) {
            assert Objects.equals(get.uid().field(), IdFieldMapper.NAME) : get.uid().field();
            // 使用 ReentrantReadWriteLock 實現并發鎖
            try (ReleasableLock ignored = readLock.acquire()) {
                // 確保未close
                ensureOpen();
                if (get.realtime()) {
                    final VersionValue versionValue;
                    try (Releasable ignore = versionMap.acquireLock(get.uid().bytes())) {
                        // we need to lock here to access the version map to do this truly in RT
                        versionValue = getVersionFromMap(get.uid().bytes());
                    }
                    // 版本號為空,以下大片邏輯就不走了
                    if (versionValue != null) {
                        if (versionValue.isDelete()) {
                            return GetResult.NOT_EXISTS;
                        }
                        if (get.versionType().isVersionConflictForReads(versionValue.version, get.version())) {
                            throw new VersionConflictEngineException(shardId, get.id(),
                                get.versionType().explainConflictForReads(versionValue.version, get.version()));
                        }
                        if (get.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO && (
                            get.getIfSeqNo() != versionValue.seqNo || get.getIfPrimaryTerm() != versionValue.term
                        )) {
                            throw new VersionConflictEngineException(shardId, get.id(),
                                get.getIfSeqNo(), get.getIfPrimaryTerm(), versionValue.seqNo, versionValue.term);
                        }
                        if (get.isReadFromTranslog()) {
                            // this is only used for updates - API _GET calls will always read form a reader for consistency
                            // the update call doesn't need the consistency since it's source only + _parent but parent can go away in 7.0
                            if (versionValue.getLocation() != null) {
                                try {
                                    final Translog.Operation operation = translog.readOperation(versionValue.getLocation());
                                    if (operation != null) {
                                        return getFromTranslog(get, (Translog.Index) operation, mapper, searcherWrapper);
                                    }
                                } catch (IOException e) {
                                    maybeFailEngine("realtime_get", e); // lets check if the translog has failed with a tragic event
                                    throw new EngineException(shardId, "failed to read operation from translog", e);
                                }
                            } else {
                                trackTranslogLocation.set(true);
                            }
                        }
                        assert versionValue.seqNo >= 0 : versionValue;
                        refreshIfNeeded("realtime_get", versionValue.seqNo);
                    }
                    return getFromSearcher(get, acquireSearcher("realtime_get", SearcherScope.INTERNAL, searcherWrapper));
                } else {
                    // we expose what has been externally expose in a point in time snapshot via an explicit refresh
                    return getFromSearcher(get, acquireSearcher("get", SearcherScope.EXTERNAL, searcherWrapper));
                }
            }
        }
          // org.elasticsearch.index.engine.Engine#ensureOpen()
        protected final void ensureOpen() {
            ensureOpen(null);
        }
        protected final void ensureOpen(Exception suppressed) {
            if (isClosed.get()) {
                AlreadyClosedException ace = new AlreadyClosedException(shardId + " engine is closed", failedEngine.get());
                if (suppressed != null) {
                    ace.addSuppressed(suppressed);
                }
                throw ace;
            }
        }
        // org.elasticsearch.index.engine.LiveVersionMap#acquireLock
        /**
         * Acquires a releaseable lock for the given uId. All *UnderLock methods require
         * this lock to be hold by the caller otherwise the visibility guarantees of this version
         * map are broken. We assert on this lock to be hold when calling these methods.
         * @see KeyedLock
         */
        Releasable acquireLock(BytesRef uid) {
            return keyedLock.acquire(uid);
        }
        // org.elasticsearch.common.util.concurrent.KeyedLock#acquire
        /**
         * Acquires a lock for the given key. The key is compared by it's equals method not by object identity. The lock can be acquired
         * by the same thread multiple times. The lock is released by closing the returned {@link Releasable}.
         */
        public Releasable acquire(T key) {
            while (true) {
                KeyLock perNodeLock = map.get(key);
                if (perNodeLock == null) {
                    ReleasableLock newLock = tryCreateNewLock(key);
                    if (newLock != null) {
                        return newLock;
                    }
                } else {
                    assert perNodeLock != null;
                    int i = perNodeLock.count.get();
                    // 最終基于 ReentrantLock 實現鎖, KeyLock 實現了計數
                    if (i > 0 && perNodeLock.count.compareAndSet(i, i + 1)) {
                        perNodeLock.lock();
                        return new ReleasableLock(key, perNodeLock);
                    }
                }
            }
        }
    
        // org.elasticsearch.common.util.concurrent.KeyedLock#tryCreateNewLock
        private ReleasableLock tryCreateNewLock(T key) {
            // KeyLock extends ReentrantLock
            KeyLock newLock = new KeyLock(fair);
            newLock.lock();
            KeyLock keyLock = map.putIfAbsent(key, newLock);
            if (keyLock == null) {
                return new ReleasableLock(key, newLock);
            }
            return null;
        }
        // org.elasticsearch.index.engine.InternalEngine#getVersionFromMap
        private VersionValue getVersionFromMap(BytesRef id) {
            if (versionMap.isUnsafe()) {
                synchronized (versionMap) {
                    // we are switching from an unsafe map to a safe map. This might happen concurrently
                    // but we only need to do this once since the last operation per ID is to add to the version
                    // map so once we pass this point we can safely lookup from the version map.
                    if (versionMap.isUnsafe()) {
                        refresh("unsafe_version_map", SearcherScope.INTERNAL, true);
                    }
                    versionMap.enforceSafeAccess();
                }
            }
            return versionMap.getUnderLock(id);
        }
        // org.elasticsearch.index.engine.LiveVersionMap#getUnderLock(org.apache.lucene.util.BytesRef)
        /**
         * Returns the live version (add or delete) for this uid.
         */
        VersionValue getUnderLock(final BytesRef uid) {
            return getUnderLock(uid, maps);
        }
    
        private VersionValue getUnderLock(final BytesRef uid, Maps currentMaps) {
            assert assertKeyedLockHeldByCurrentThread(uid);
            // First try to get the "live" value:
            VersionValue value = currentMaps.current.get(uid);
            if (value != null) {
                return value;
            }
    
            value = currentMaps.old.get(uid);
            if (value != null) {
                return value;
            }
            // 返回 null
            return tombstones.get(uid);
        }
        // org.elasticsearch.index.engine.Engine#acquireSearcher
        public Searcher acquireSearcher(String source, SearcherScope scope, Function<Searcher, Searcher> wrapper) throws EngineException {
            SearcherSupplier releasable = null;
            try {
                // 獲取 searcher , soure=realtime_get
                SearcherSupplier reader = releasable = acquireSearcherSupplier(wrapper, scope);
                Searcher searcher = reader.acquireSearcher(source);
                releasable = null;
                return new Searcher(source, searcher.getDirectoryReader(), searcher.getSimilarity(),
                    searcher.getQueryCache(), searcher.getQueryCachingPolicy(), () -> Releasables.close(searcher, reader));
            } finally {
                Releasables.close(releasable);
            }
        }
        // org.elasticsearch.index.engine.Engine#acquireSearcherSupplier
        /**
         * Acquires a point-in-time reader that can be used to create {@link Engine.Searcher}s on demand.
         */
        public SearcherSupplier acquireSearcherSupplier(Function<Searcher, Searcher> wrapper, SearcherScope scope) throws EngineException {
            /* Acquire order here is store -> manager since we need
             * to make sure that the store is not closed before
             * the searcher is acquired. */
            if (store.tryIncRef() == false) {
                throw new AlreadyClosedException(shardId + " store is closed", failedEngine.get());
            }
            Releasable releasable = store::decRef;
            try {
                // 引用計數+1
                ReferenceManager<ElasticsearchDirectoryReader> referenceManager = getReferenceManager(scope);
                ElasticsearchDirectoryReader acquire = referenceManager.acquire();
                SearcherSupplier reader = new SearcherSupplier(wrapper) {
                    @Override
                    public Searcher acquireSearcherInternal(String source) {
                        assert assertSearcherIsWarmedUp(source, scope);
                        return new Searcher(source, acquire, engineConfig.getSimilarity(), engineConfig.getQueryCache(),
                            engineConfig.getQueryCachingPolicy(), () -> {});
                    }
    
                    @Override
                    protected void doClose() {
                        try {
                            referenceManager.release(acquire);
                        } catch (IOException e) {
                            throw new UncheckedIOException("failed to close", e);
                        } catch (AlreadyClosedException e) {
                            // This means there's a bug somewhere: don't suppress it
                            throw new AssertionError(e);
                        } finally {
                            store.decRef();
                        }
                    }
                };
                releasable = null; // success - hand over the reference to the engine reader
                return reader;
            } catch (AlreadyClosedException ex) {
                throw ex;
            } catch (Exception ex) {
                maybeFailEngine("acquire_reader", ex);
                ensureOpen(ex); // throw EngineCloseException here if we are already closed
                logger.error(() -> new ParameterizedMessage("failed to acquire reader"), ex);
                throw new EngineException(shardId, "failed to acquire reader", ex);
            } finally {
                Releasables.close(releasable);
            }
        }
            // org.elasticsearch.index.engine.Engine.SearcherSupplier#acquireSearcher
            public final Searcher acquireSearcher(String source) {
                if (released.get()) {
                    throw new AlreadyClosedException("SearcherSupplier was closed");
                }
                // 新建 searcher
                final Searcher searcher = acquireSearcherInternal(source);
                return CAN_MATCH_SEARCH_SOURCE.equals(source) ? searcher : wrapper.apply(searcher);
            }
            // org.elasticsearch.index.engine.Engine.Searcher#Searcher
            public Searcher(String source, IndexReader reader,
                            Similarity similarity, QueryCache queryCache, QueryCachingPolicy queryCachingPolicy,
                            Closeable onClose) {
                super(reader);
                setSimilarity(similarity);
                setQueryCache(queryCache);
                setQueryCachingPolicy(queryCachingPolicy);
                this.source = source;
                this.onClose = onClose;
            }
        // org.elasticsearch.index.shard.IndexShard#wrapSearcher
        private Engine.Searcher wrapSearcher(Engine.Searcher searcher) {
            assert ElasticsearchDirectoryReader.unwrap(searcher.getDirectoryReader())
                != null : "DirectoryReader must be an instance or ElasticsearchDirectoryReader";
            boolean success = false;
            try {
                final Engine.Searcher newSearcher = readerWrapper == null ? searcher : wrapSearcher(searcher, readerWrapper);
                assert newSearcher != null;
                success = true;
                return newSearcher;
            } catch (IOException ex) {
                throw new ElasticsearchException("failed to wrap searcher", ex);
            } finally {
                if (success == false) {
                    Releasables.close(success, searcher);
                }
            }
        }
        // org.elasticsearch.index.shard.IndexShard#wrapSearcher
        static Engine.Searcher wrapSearcher(Engine.Searcher engineSearcher,
                                            CheckedFunction<DirectoryReader, DirectoryReader, IOException> readerWrapper) throws IOException {
            assert readerWrapper != null;
            final ElasticsearchDirectoryReader elasticsearchDirectoryReader =
                ElasticsearchDirectoryReader.getElasticsearchDirectoryReader(engineSearcher.getDirectoryReader());
            if (elasticsearchDirectoryReader == null) {
                throw new IllegalStateException("Can't wrap non elasticsearch directory reader");
            }
            NonClosingReaderWrapper nonClosingReaderWrapper = new NonClosingReaderWrapper(engineSearcher.getDirectoryReader());
            DirectoryReader reader = readerWrapper.apply(nonClosingReaderWrapper);
            if (reader != nonClosingReaderWrapper) {
                if (reader.getReaderCacheHelper() != elasticsearchDirectoryReader.getReaderCacheHelper()) {
                    throw new IllegalStateException("wrapped directory reader doesn't delegate IndexReader#getCoreCacheKey," +
                        " wrappers must override this method and delegate to the original readers core cache key. Wrapped readers can't be " +
                        "used as cache keys since their are used only per request which would lead to subtle bugs");
                }
                if (ElasticsearchDirectoryReader.getElasticsearchDirectoryReader(reader) != elasticsearchDirectoryReader) {
                    // prevent that somebody wraps with a non-filter reader
                    throw new IllegalStateException("wrapped directory reader hides actual ElasticsearchDirectoryReader but shouldn't");
                }
            }
    
            if (reader == nonClosingReaderWrapper) {
                return engineSearcher;
            } else {
                // we close the reader to make sure wrappers can release resources if needed....
                // our NonClosingReaderWrapper makes sure that our reader is not closed
                return new Engine.Searcher(engineSearcher.source(), reader,
                    engineSearcher.getSimilarity(), engineSearcher.getQueryCache(), engineSearcher.getQueryCachingPolicy(),
                    () -> IOUtils.close(reader, // this will close the wrappers excluding the NonClosingReaderWrapper
                        engineSearcher)); // this will run the closeable on the wrapped engine reader
            }
        }

      獲取到searcher實例后,再由其進行get操作。

        // org.elasticsearch.index.engine.Engine#getFromSearcher
        protected final GetResult getFromSearcher(Get get, Engine.Searcher searcher) throws EngineException {
            final DocIdAndVersion docIdAndVersion;
            try {
                // 讀取docId, version, 其實就是核心的讀數據實現之一
                docIdAndVersion = VersionsAndSeqNoResolver.loadDocIdAndVersion(searcher.getIndexReader(), get.uid(), true);
            } catch (Exception e) {
                Releasables.closeWhileHandlingException(searcher);
                //TODO: A better exception goes here
                throw new EngineException(shardId, "Couldn't resolve version", e);
            }
    
            if (docIdAndVersion != null) {
                // 檢測版本是否沖突,如有則拋出異常
                if (get.versionType().isVersionConflictForReads(docIdAndVersion.version, get.version())) {
                    Releasables.close(searcher);
                    throw new VersionConflictEngineException(shardId, get.id(),
                            get.versionType().explainConflictForReads(docIdAndVersion.version, get.version()));
                }
                if (get.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO && (
                    get.getIfSeqNo() != docIdAndVersion.seqNo || get.getIfPrimaryTerm() != docIdAndVersion.primaryTerm
                )) {
                    Releasables.close(searcher);
                    throw new VersionConflictEngineException(shardId, get.id(),
                        get.getIfSeqNo(), get.getIfPrimaryTerm(), docIdAndVersion.seqNo, docIdAndVersion.primaryTerm);
                }
            }
    
            if (docIdAndVersion != null) {
                // don't release the searcher on this path, it is the
                // responsibility of the caller to call GetResult.release
                return new GetResult(searcher, docIdAndVersion, false);
            } else {
                Releasables.close(searcher);
                return GetResult.NOT_EXISTS;
            }
        }
        // org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver#loadDocIdAndVersion
        /**
         * Load the internal doc ID and version for the uid from the reader, returning<ul>
         * <li>null if the uid wasn't found,
         * <li>a doc ID and a version otherwise
         * </ul>
         */
        public static DocIdAndVersion loadDocIdAndVersion(IndexReader reader, Term term, boolean loadSeqNo) throws IOException {
            // feild: _id
            PerThreadIDVersionAndSeqNoLookup[] lookups = getLookupState(reader, term.field());
            List<LeafReaderContext> leaves = reader.leaves();
            // iterate backwards to optimize for the frequently updated documents
            // which are likely to be in the last segments
            for (int i = leaves.size() - 1; i >= 0; i--) {
                final LeafReaderContext leaf = leaves.get(i);
                PerThreadIDVersionAndSeqNoLookup lookup = lookups[leaf.ord];
                DocIdAndVersion result = lookup.lookupVersion(term.bytes(), loadSeqNo, leaf);
                if (result != null) {
                    return result;
                }
            }
            return null;
        }
        // org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver#getLookupState
        private static PerThreadIDVersionAndSeqNoLookup[] getLookupState(IndexReader reader, String uidField) throws IOException {
            // We cache on the top level
            // This means cache entries have a shorter lifetime, maybe as low as 1s with the
            // default refresh interval and a steady indexing rate, but on the other hand it
            // proved to be cheaper than having to perform a CHM and a TL get for every segment.
            // See https://github.com/elastic/elasticsearch/pull/19856.
            IndexReader.CacheHelper cacheHelper = reader.getReaderCacheHelper();
            CloseableThreadLocal<PerThreadIDVersionAndSeqNoLookup[]> ctl = lookupStates.get(cacheHelper.getKey());
            if (ctl == null) {
                // First time we are seeing this reader's core; make a new CTL:
                ctl = new CloseableThreadLocal<>();
                CloseableThreadLocal<PerThreadIDVersionAndSeqNoLookup[]> other = lookupStates.putIfAbsent(cacheHelper.getKey(), ctl);
                if (other == null) {
                    // Our CTL won, we must remove it when the reader is closed:
                    cacheHelper.addClosedListener(removeLookupState);
                } else {
                    // Another thread beat us to it: just use their CTL:
                    ctl = other;
                }
            }
    
            PerThreadIDVersionAndSeqNoLookup[] lookupState = ctl.get();
            if (lookupState == null) {
                lookupState = new PerThreadIDVersionAndSeqNoLookup[reader.leaves().size()];
                for (LeafReaderContext leaf : reader.leaves()) {
                    lookupState[leaf.ord] = new PerThreadIDVersionAndSeqNoLookup(leaf.reader(), uidField);
                }
                ctl.set(lookupState);
            }
    
            if (lookupState.length != reader.leaves().size()) {
                throw new AssertionError("Mismatched numbers of leaves: " + lookupState.length + " != " + reader.leaves().size());
            }
    
            if (lookupState.length > 0 && Objects.equals(lookupState[0].uidField, uidField) == false) {
                throw new AssertionError("Index does not consistently use the same uid field: ["
                        + uidField + "] != [" + lookupState[0].uidField + "]");
            }
    
            return lookupState;
        }
      // org.apache.lucene.index.IndexReader#leaves
      /**
       * Returns the reader's leaves, or itself if this reader is atomic.
       * This is a convenience method calling {@code this.getContext().leaves()}.
       * @see IndexReaderContext#leaves()
       */
      public final List<LeafReaderContext> leaves() {
        return getContext().leaves();
      }
      // org.apache.lucene.index.CompositeReader#getContext
      @Override
      public final CompositeReaderContext getContext() {
        ensureOpen();
        // lazy init without thread safety for perf reasons: Building the readerContext twice does not hurt!
        if (readerContext == null) {
          assert getSequentialSubReaders() != null;
          readerContext = CompositeReaderContext.create(this);
        }
        return readerContext;
      }
      // org.apache.lucene.index.CompositeReaderContext#leaves
      @Override
      public List<LeafReaderContext> leaves() throws UnsupportedOperationException {
        if (!isTopLevel)
          throw new UnsupportedOperationException("This is not a top-level context.");
        assert leaves != null;
        return leaves;
      }
          // org.elasticsearch.common.lucene.uid.PerThreadIDVersionAndSeqNoLookup#lookupVersion
        /** Return null if id is not found.
         * We pass the {@link LeafReaderContext} as an argument so that things
         * still work with reader wrappers that hide some documents while still
         * using the same cache key. Otherwise we'd have to disable caching
         * entirely for these readers.
         */
        public DocIdAndVersion lookupVersion(BytesRef id, boolean loadSeqNo, LeafReaderContext context)
            throws IOException {
            assert context.reader().getCoreCacheHelper().getKey().equals(readerKey) :
                "context's reader is not the same as the reader class was initialized on.";
            // docId 是個整型?
            int docID = getDocID(id, context);
    
            if (docID != DocIdSetIterator.NO_MORE_DOCS) {
                final long seqNo;
                final long term;
                if (loadSeqNo) {
                    seqNo = readNumericDocValues(context.reader(), SeqNoFieldMapper.NAME, docID);
                    term = readNumericDocValues(context.reader(), SeqNoFieldMapper.PRIMARY_TERM_NAME, docID);
                } else {
                    seqNo = UNASSIGNED_SEQ_NO;
                    term = UNASSIGNED_PRIMARY_TERM;
                }
                final long version = readNumericDocValues(context.reader(), VersionFieldMapper.NAME, docID);
                return new DocIdAndVersion(docID, version, seqNo, term, context.reader(), context.docBase);
            } else {
                return null;
            }
        }
        // org.elasticsearch.common.lucene.uid.PerThreadIDVersionAndSeqNoLookup#getDocID
        /**
         * returns the internal lucene doc id for the given id bytes.
         * {@link DocIdSetIterator#NO_MORE_DOCS} is returned if not found
         * */
        private int getDocID(BytesRef id, LeafReaderContext context) throws IOException {
            // termsEnum can possibly be null here if this leaf contains only no-ops.
            // lucene 接口: seekExact
            if (termsEnum != null && termsEnum.seekExact(id)) {
                final Bits liveDocs = context.reader().getLiveDocs();
                int docID = DocIdSetIterator.NO_MORE_DOCS;
                // there may be more than one matching docID, in the case of nested docs, so we want the last one:
                docsEnum = termsEnum.postings(docsEnum, 0);
                for (int d = docsEnum.nextDoc(); d != DocIdSetIterator.NO_MORE_DOCS; d = docsEnum.nextDoc()) {
                    if (liveDocs != null && liveDocs.get(d) == false) {
                        continue;
                    }
                    docID = d;
                }
                return docID;
            } else {
                return DocIdSetIterator.NO_MORE_DOCS;
            }
        }
    
        // org.elasticsearch.common.lucene.uid.PerThreadIDVersionAndSeqNoLookup#readNumericDocValues
        private static long readNumericDocValues(LeafReader reader, String field, int docId) throws IOException {
            final NumericDocValues dv = reader.getNumericDocValues(field);
            if (dv == null || dv.advanceExact(docId) == false) {
                assert false : "document [" + docId + "] does not have docValues for [" + field + "]";
                throw new IllegalStateException("document [" + docId + "] does not have docValues for [" + field + "]");
            }
            return dv.longValue();
        }
        // docId 包裝
        // org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver.DocIdAndVersion
        /** Wraps an {@link LeafReaderContext}, a doc ID <b>relative to the context doc base</b> and a version. */
        public static class DocIdAndVersion {
            public final int docId;
            public final long version;
            public final long seqNo;
            public final long primaryTerm;
            public final LeafReader reader;
            public final int docBase;
    
            public DocIdAndVersion(int docId, long version, long seqNo, long primaryTerm, LeafReader reader, int docBase) {
                this.docId = docId;
                this.version = version;
                this.seqNo = seqNo;
                this.primaryTerm = primaryTerm;
                this.reader = reader;
                this.docBase = docBase;
            }
        }
            // org.elasticsearch.index.VersionType#isVersionConflictForReads
            @Override
            public boolean isVersionConflictForReads(long currentVersion, long expectedVersion) {
                return isVersionConflict(currentVersion, expectedVersion, false);
            }
            private boolean isVersionConflict(long currentVersion, long expectedVersion, boolean deleted) {
                if (expectedVersion == Versions.MATCH_ANY) {
                    return false;
                }
                if (expectedVersion == Versions.MATCH_DELETED) {
                    return deleted == false;
                }
                if (currentVersion != expectedVersion) {
                    return true;
                }
                return false;
            }
            // 最后,構造 GetResult 返回
            // org.elasticsearch.index.engine.Engine.GetResult#GetResult
            public GetResult(Engine.Searcher searcher, DocIdAndVersion docIdAndVersion, boolean fromTranslog) {
                this(true, docIdAndVersion.version, docIdAndVersion, searcher, fromTranslog);
            }
    
            private GetResult(boolean exists, long version, DocIdAndVersion docIdAndVersion, Engine.Searcher searcher, boolean fromTranslog) {
                this.exists = exists;
                this.version = version;
                this.docIdAndVersion = docIdAndVersion;
                this.searcher = searcher;
                this.fromTranslog = fromTranslog;
                assert fromTranslog == false || searcher.getIndexReader() instanceof TranslogLeafReader;
            }

      前面根據id調用 lucene 接口獲取到了docId, 接下來根據docId查詢doc詳情:

        // org.elasticsearch.index.get.ShardGetService#innerGetLoadFromStoredFields
        private GetResult innerGetLoadFromStoredFields(String type, String id, String[] storedFields, FetchSourceContext fetchSourceContext,
                                                       Engine.GetResult get, MapperService mapperService) {
            assert get.exists() : "method should only be called if document could be retrieved";
    
            // check first if stored fields to be loaded don't contain an object field
            DocumentMapper docMapper = mapperService.documentMapper();
            if (storedFields != null) {
                for (String field : storedFields) {
                    Mapper fieldMapper = docMapper.mappers().getMapper(field);
                    if (fieldMapper == null) {
                        if (docMapper.mappers().objectMappers().get(field) != null) {
                            // Only fail if we know it is a object field, missing paths / fields shouldn't fail.
                            throw new IllegalArgumentException("field [" + field + "] isn't a leaf field");
                        }
                    }
                }
            }
    
            Map<String, DocumentField> documentFields = null;
            Map<String, DocumentField> metadataFields = null;
            BytesReference source = null;
            DocIdAndVersion docIdAndVersion = get.docIdAndVersion();
            // force fetching source if we read from translog and need to recreate stored fields
            boolean forceSourceForComputingTranslogStoredFields = get.isFromTranslog() && storedFields != null &&
                    Stream.of(storedFields).anyMatch(f -> TranslogLeafReader.ALL_FIELD_NAMES.contains(f) == false);
            // 構造字段訪問器
            FieldsVisitor fieldVisitor = buildFieldsVisitors(storedFields,
                forceSourceForComputingTranslogStoredFields ? FetchSourceContext.FETCH_SOURCE : fetchSourceContext);
            if (fieldVisitor != null) {
                try {
                    // 一個字段讀取
                    docIdAndVersion.reader.document(docIdAndVersion.docId, fieldVisitor);
                } catch (IOException e) {
                    throw new ElasticsearchException("Failed to get type [" + type + "] and id [" + id + "]", e);
                }
                source = fieldVisitor.source();
    
                // in case we read from translog, some extra steps are needed to make _source consistent and to load stored fields
                if (get.isFromTranslog()) {
                    // Fast path: if only asked for the source or stored fields that have been already provided by TranslogLeafReader,
                    // just make source consistent by reapplying source filters from mapping (possibly also nulling the source)
                    if (forceSourceForComputingTranslogStoredFields == false) {
                        try {
                            source = indexShard.mapperService().documentMapper().sourceMapper().applyFilters(source, null);
                        } catch (IOException e) {
                            throw new ElasticsearchException("Failed to reapply filters for [" + id + "] after reading from translog", e);
                        }
                    } else {
                        // Slow path: recreate stored fields from original source
                        assert source != null : "original source in translog must exist";
                        SourceToParse sourceToParse = new SourceToParse(shardId.getIndexName(), type, id, source,
                            XContentHelper.xContentType(source), fieldVisitor.routing());
                        ParsedDocument doc = indexShard.mapperService().documentMapper().parse(sourceToParse);
                        assert doc.dynamicMappingsUpdate() == null : "mapping updates should not be required on already-indexed doc";
                        // update special fields
                        doc.updateSeqID(docIdAndVersion.seqNo, docIdAndVersion.primaryTerm);
                        doc.version().setLongValue(docIdAndVersion.version);
    
                        // retrieve stored fields from parsed doc
                        fieldVisitor = buildFieldsVisitors(storedFields, fetchSourceContext);
                        for (IndexableField indexableField : doc.rootDoc().getFields()) {
                            IndexableFieldType fieldType = indexableField.fieldType();
                            if (fieldType.stored()) {
                                FieldInfo fieldInfo = new FieldInfo(indexableField.name(), 0, false, false, false, IndexOptions.NONE,
                                    DocValuesType.NONE, -1, Collections.emptyMap(), 0, 0, 0, false);
                                StoredFieldVisitor.Status status = fieldVisitor.needsField(fieldInfo);
                                if (status == StoredFieldVisitor.Status.YES) {
                                    if (indexableField.numericValue() != null) {
                                        fieldVisitor.objectField(fieldInfo, indexableField.numericValue());
                                    } else if (indexableField.binaryValue() != null) {
                                        fieldVisitor.binaryField(fieldInfo, indexableField.binaryValue());
                                    } else if (indexableField.stringValue() != null) {
                                        fieldVisitor.objectField(fieldInfo, indexableField.stringValue());
                                    }
                                } else if (status == StoredFieldVisitor.Status.STOP) {
                                    break;
                                }
                            }
                        }
                        // retrieve source (with possible transformations, e.g. source filters
                        source = fieldVisitor.source();
                    }
                }
    
                // put stored fields into result objects
                if (!fieldVisitor.fields().isEmpty()) {
                    fieldVisitor.postProcess(mapperService::fieldType,
                        mapperService.documentMapper() == null ? null : mapperService.documentMapper().type());
                    documentFields = new HashMap<>();
                    metadataFields = new HashMap<>();
                    for (Map.Entry<String, List<Object>> entry : fieldVisitor.fields().entrySet()) {
                        if (mapperService.isMetadataField(entry.getKey())) {
                            metadataFields.put(entry.getKey(), new DocumentField(entry.getKey(), entry.getValue()));
                        } else {
                            documentFields.put(entry.getKey(), new DocumentField(entry.getKey(), entry.getValue()));
                        }
                    }
                }
            }
    
            if (source != null) {
                // apply request-level source filtering
                if (fetchSourceContext.fetchSource() == false) {
                    source = null;
                } else if (fetchSourceContext.includes().length > 0 || fetchSourceContext.excludes().length > 0) {
                    Map<String, Object> sourceAsMap;
                    // TODO: The source might be parsed and available in the sourceLookup but that one uses unordered maps so different.
                    //  Do we care?
                    Tuple<XContentType, Map<String, Object>> typeMapTuple = XContentHelper.convertToMap(source, true);
                    XContentType sourceContentType = typeMapTuple.v1();
                    sourceAsMap = typeMapTuple.v2();
                    sourceAsMap = XContentMapValues.filter(sourceAsMap, fetchSourceContext.includes(), fetchSourceContext.excludes());
                    try {
                        source = BytesReference.bytes(XContentFactory.contentBuilder(sourceContentType).map(sourceAsMap));
                    } catch (IOException e) {
                        throw new ElasticsearchException("Failed to get id [" + id + "] with includes/excludes set", e);
                    }
                }
            }
    
            if (!fetchSourceContext.fetchSource()) {
                source = null;
            }
    
            if (source != null && get.isFromTranslog()) {
                // reapply source filters from mapping (possibly also nulling the source)
                try {
                    source = docMapper.sourceMapper().applyFilters(source, null);
                } catch (IOException e) {
                    throw new ElasticsearchException("Failed to reapply filters for [" + id + "] after reading from translog", e);
                }
            }
    
            if (source != null && (fetchSourceContext.includes().length > 0 || fetchSourceContext.excludes().length > 0)) {
                Map<String, Object> sourceAsMap;
                // TODO: The source might parsed and available in the sourceLookup but that one uses unordered maps so different. Do we care?
                Tuple<XContentType, Map<String, Object>> typeMapTuple = XContentHelper.convertToMap(source, true);
                XContentType sourceContentType = typeMapTuple.v1();
                sourceAsMap = typeMapTuple.v2();
                sourceAsMap = XContentMapValues.filter(sourceAsMap, fetchSourceContext.includes(), fetchSourceContext.excludes());
                try {
                    source = BytesReference.bytes(XContentFactory.contentBuilder(sourceContentType).map(sourceAsMap));
                } catch (IOException e) {
                    throw new ElasticsearchException("Failed to get type [" + type + "] and id [" + id + "] with includes/excludes set", e);
                }
            }
            // 構造新的 GetResult 返回, source 就是最終查詢的結果
            return new GetResult(shardId.getIndexName(), type, id, get.docIdAndVersion().seqNo, get.docIdAndVersion().primaryTerm,
                get.version(), get.exists(), source, documentFields, metadataFields);
        }
    
        private static FieldsVisitor buildFieldsVisitors(String[] fields, FetchSourceContext fetchSourceContext) {
            if (fields == null || fields.length == 0) {
                return fetchSourceContext.fetchSource() ? new FieldsVisitor(true) : null;
            }
    
            return new CustomFieldsVisitor(Sets.newHashSet(fields), fetchSourceContext.fetchSource());
        }

      最后,一點lucene的查找實現片段,供參考。

      // 快速定位到指定id的文檔位置,相當于索引定位
      // org.apache.lucene.codecs.blocktree.SegmentTermsEnum#seekExact(org.apache.lucene.util.BytesRef)
      @Override
      public boolean seekExact(BytesRef target) throws IOException {
    
        if (fr.index == null) {
          throw new IllegalStateException("terms index was not loaded");
        }
    
        if (fr.size() > 0 && (target.compareTo(fr.getMin()) < 0 || target.compareTo(fr.getMax()) > 0)) {
            return false;
        }
    
        term.grow(1 + target.length);
    
        assert clearEOF();
    
        // if (DEBUG) {
        //   System.out.println("\nBTTR.seekExact seg=" + fr.parent.segment + " target=" + fr.fieldInfo.name + ":" + brToString(target) + " current=" + brToString(term) + " (exists?=" + termExists + ") validIndexPrefix=" + validIndexPrefix);
        //   printSeekState(System.out);
        // }
    
        FST.Arc<BytesRef> arc;
        int targetUpto;
        BytesRef output;
    
        targetBeforeCurrentLength = currentFrame.ord;
    
        if (currentFrame != staticFrame) {
    
          // We are already seek'd; find the common
          // prefix of new seek term vs current term and
          // re-use the corresponding seek state.  For
          // example, if app first seeks to foobar, then
          // seeks to foobaz, we can re-use the seek state
          // for the first 5 bytes.
    
          // if (DEBUG) {
          //   System.out.println("  re-use current seek state validIndexPrefix=" + validIndexPrefix);
          // }
    
          arc = arcs[0];
          assert arc.isFinal();
          output = arc.output();
          targetUpto = 0;
              
          SegmentTermsEnumFrame lastFrame = stack[0];
          assert validIndexPrefix <= term.length();
    
          final int targetLimit = Math.min(target.length, validIndexPrefix);
    
          int cmp = 0;
    
          // TODO: reverse vLong byte order for better FST
          // prefix output sharing
    
          // First compare up to valid seek frames:
          while (targetUpto < targetLimit) {
            cmp = (term.byteAt(targetUpto)&0xFF) - (target.bytes[target.offset + targetUpto]&0xFF);
            // if (DEBUG) {
            //    System.out.println("    cycle targetUpto=" + targetUpto + " (vs limit=" + targetLimit + ") cmp=" + cmp + " (targetLabel=" + (char) (target.bytes[target.offset + targetUpto]) + " vs termLabel=" + (char) (term.bytes[targetUpto]) + ")"   + " arc.output=" + arc.output + " output=" + output);
            // }
            if (cmp != 0) {
              break;
            }
            arc = arcs[1+targetUpto];
            assert arc.label() == (target.bytes[target.offset + targetUpto] & 0xFF): "arc.label=" + (char) arc.label() + " targetLabel=" + (char) (target.bytes[target.offset + targetUpto] & 0xFF);
            if (arc.output() != BlockTreeTermsReader.NO_OUTPUT) {
              output = BlockTreeTermsReader.FST_OUTPUTS.add(output, arc.output());
            }
            if (arc.isFinal()) {
              lastFrame = stack[1+lastFrame.ord];
            }
            targetUpto++;
          }
    
          if (cmp == 0) {
            final int targetUptoMid = targetUpto;
    
            // Second compare the rest of the term, but
            // don't save arc/output/frame; we only do this
            // to find out if the target term is before,
            // equal or after the current term
            final int targetLimit2 = Math.min(target.length, term.length());
            while (targetUpto < targetLimit2) {
              cmp = (term.byteAt(targetUpto)&0xFF) - (target.bytes[target.offset + targetUpto]&0xFF);
              // if (DEBUG) {
              //    System.out.println("    cycle2 targetUpto=" + targetUpto + " (vs limit=" + targetLimit + ") cmp=" + cmp + " (targetLabel=" + (char) (target.bytes[target.offset + targetUpto]) + " vs termLabel=" + (char) (term.bytes[targetUpto]) + ")");
              // }
              if (cmp != 0) {
                break;
              }
              targetUpto++;
            }
    
            if (cmp == 0) {
              cmp = term.length() - target.length;
            }
            targetUpto = targetUptoMid;
          }
    
          if (cmp < 0) {
            // Common case: target term is after current
            // term, ie, app is seeking multiple terms
            // in sorted order
            // if (DEBUG) {
            //   System.out.println("  target is after current (shares prefixLen=" + targetUpto + "); frame.ord=" + lastFrame.ord);
            // }
            currentFrame = lastFrame;
    
          } else if (cmp > 0) {
            // Uncommon case: target term
            // is before current term; this means we can
            // keep the currentFrame but we must rewind it
            // (so we scan from the start)
            targetBeforeCurrentLength = lastFrame.ord;
            // if (DEBUG) {
            //   System.out.println("  target is before current (shares prefixLen=" + targetUpto + "); rewind frame ord=" + lastFrame.ord);
            // }
            currentFrame = lastFrame;
            currentFrame.rewind();
          } else {
            // Target is exactly the same as current term
            assert term.length() == target.length;
            if (termExists) {
              // if (DEBUG) {
              //   System.out.println("  target is same as current; return true");
              // }
              return true;
            } else {
              // if (DEBUG) {
              //   System.out.println("  target is same as current but term doesn't exist");
              // }
            }
            //validIndexPrefix = currentFrame.depth;
            //term.length = target.length;
            //return termExists;
          }
    
        } else {
    
          targetBeforeCurrentLength = -1;
          arc = fr.index.getFirstArc(arcs[0]);
    
          // Empty string prefix must have an output (block) in the index!
          assert arc.isFinal();
          assert arc.output() != null;
    
          // if (DEBUG) {
          //   System.out.println("    no seek state; push root frame");
          // }
    
          output = arc.output();
    
          currentFrame = staticFrame;
    
          //term.length = 0;
          targetUpto = 0;
          currentFrame = pushFrame(arc, BlockTreeTermsReader.FST_OUTPUTS.add(output, arc.nextFinalOutput()), 0);
        }
    
        // if (DEBUG) {
        //   System.out.println("  start index loop targetUpto=" + targetUpto + " output=" + output + " currentFrame.ord=" + currentFrame.ord + " targetBeforeCurrentLength=" + targetBeforeCurrentLength);
        // }
    
        // We are done sharing the common prefix with the incoming target and where we are currently seek'd; now continue walking the index:
        while (targetUpto < target.length) {
    
          final int targetLabel = target.bytes[target.offset + targetUpto] & 0xFF;
    
          final FST.Arc<BytesRef> nextArc = fr.index.findTargetArc(targetLabel, arc, getArc(1+targetUpto), fstReader);
    
          if (nextArc == null) {
    
            // Index is exhausted
            // if (DEBUG) {
            //   System.out.println("    index: index exhausted label=" + ((char) targetLabel) + " " + toHex(targetLabel));
            // }
                
            validIndexPrefix = currentFrame.prefix;
            //validIndexPrefix = targetUpto;
    
            currentFrame.scanToFloorFrame(target);
    
            if (!currentFrame.hasTerms) {
              termExists = false;
              term.setByteAt(targetUpto, (byte) targetLabel);
              term.setLength(1+targetUpto);
              // if (DEBUG) {
              //   System.out.println("  FAST NOT_FOUND term=" + brToString(term));
              // }
              return false;
            }
    
            currentFrame.loadBlock();
    
            final SeekStatus result = currentFrame.scanToTerm(target, true);            
            if (result == SeekStatus.FOUND) {
              // if (DEBUG) {
              //   System.out.println("  return FOUND term=" + term.utf8ToString() + " " + term);
              // }
              return true;
            } else {
              // if (DEBUG) {
              //   System.out.println("  got " + result + "; return NOT_FOUND term=" + brToString(term));
              // }
              return false;
            }
          } else {
            // Follow this arc
            arc = nextArc;
            term.setByteAt(targetUpto, (byte) targetLabel);
            // Aggregate output as we go:
            assert arc.output() != null;
            if (arc.output() != BlockTreeTermsReader.NO_OUTPUT) {
              output = BlockTreeTermsReader.FST_OUTPUTS.add(output, arc.output());
            }
    
            // if (DEBUG) {
            //   System.out.println("    index: follow label=" + toHex(target.bytes[target.offset + targetUpto]&0xff) + " arc.output=" + arc.output + " arc.nfo=" + arc.nextFinalOutput);
            // }
            targetUpto++;
    
            if (arc.isFinal()) {
              //if (DEBUG) System.out.println("    arc is final!");
              currentFrame = pushFrame(arc, BlockTreeTermsReader.FST_OUTPUTS.add(output, arc.nextFinalOutput()), targetUpto);
              //if (DEBUG) System.out.println("    curFrame.ord=" + currentFrame.ord + " hasTerms=" + currentFrame.hasTerms);
            }
          }
        }
    
        //validIndexPrefix = targetUpto;
        validIndexPrefix = currentFrame.prefix;
    
        currentFrame.scanToFloorFrame(target);
    
        // Target term is entirely contained in the index:
        if (!currentFrame.hasTerms) {
          termExists = false;
          term.setLength(targetUpto);
          // if (DEBUG) {
          //   System.out.println("  FAST NOT_FOUND term=" + brToString(term));
          // }
          return false;
        }
    
        currentFrame.loadBlock();
    
        final SeekStatus result = currentFrame.scanToTerm(target, true);            
        if (result == SeekStatus.FOUND) {
          // if (DEBUG) {
          //   System.out.println("  return FOUND term=" + term.utf8ToString() + " " + term);
          // }
          return true;
        } else {
          // if (DEBUG) {
          //   System.out.println("  got result " + result + "; return NOT_FOUND term=" + term.utf8ToString());
          // }
    
          return false;
        }
      }
      // org.apache.lucene.codecs.blocktree.SegmentTermsEnum#postings
      @Override
      public PostingsEnum postings(PostingsEnum reuse, int flags) throws IOException {
        assert !eof;
        //if (DEBUG) {
        //System.out.println("BTTR.docs seg=" + segment);
        //}
        currentFrame.decodeMetaData();
        //if (DEBUG) {
        //System.out.println("  state=" + currentFrame.state);
        //}
        return fr.parent.postingsReader.postings(fr.fieldInfo, currentFrame.state, reuse, flags);
      }
      // 根據docId 獲取文檔的詳情信息
      // org.apache.lucene.index.FilterLeafReader#document
      @Override
      public void document(int docID, StoredFieldVisitor visitor) throws IOException {
        ensureOpen();
        in.document(docID, visitor);
      }
      // org.apache.lucene.index.CodecReader#document
      @Override
      public final void document(int docID, StoredFieldVisitor visitor) throws IOException {
        checkBounds(docID);
        getFieldsReader().visitDocument(docID, visitor);
      }
      // org.apache.lucene.codecs.compressing.CompressingStoredFieldsReader#visitDocument
      @Override
      public void visitDocument(int docID, StoredFieldVisitor visitor)
          throws IOException {
    
        final SerializedDocument doc = document(docID);
    
        for (int fieldIDX = 0; fieldIDX < doc.numStoredFields; fieldIDX++) {
          final long infoAndBits = doc.in.readVLong();
          final int fieldNumber = (int) (infoAndBits >>> TYPE_BITS);
          final FieldInfo fieldInfo = fieldInfos.fieldInfo(fieldNumber);
    
          final int bits = (int) (infoAndBits & TYPE_MASK);
          assert bits <= NUMERIC_DOUBLE: "bits=" + Integer.toHexString(bits);
    
          switch(visitor.needsField(fieldInfo)) {
            case YES:
              readField(doc.in, visitor, fieldInfo, bits);
              break;
            case NO:
              if (fieldIDX == doc.numStoredFields - 1) {// don't skipField on last field value; treat like STOP
                return;
              }
              skipField(doc.in, bits);
              break;
            case STOP:
              return;
          }
        }
      }
      // org.apache.lucene.codecs.compressing.CompressingStoredFieldsReader#document
      SerializedDocument document(int docID) throws IOException {
          // 定位文檔位置,取出文檔。。。
        if (state.contains(docID) == false) {
          fieldsStream.seek(indexReader.getStartPointer(docID));
          state.reset(docID);
        }
        assert state.contains(docID);
        return state.document(docID);
      }
      // org.apache.lucene.codecs.compressing.FieldsIndexReader#getStartPointer
      @Override
      long getStartPointer(int docID) {
        FutureObjects.checkIndex(docID, maxDoc);
        long blockIndex = docs.binarySearch(0, numChunks, docID);
        if (blockIndex < 0) {
          blockIndex = -2 - blockIndex;
        }
        return startPointers.get(blockIndex);
      }
          // org.apache.lucene.codecs.compressing.CompressingStoredFieldsReader.BlockState#document
        /**
         * Get the serialized representation of the given docID. This docID has
         * to be contained in the current block.
         */
        SerializedDocument document(int docID) throws IOException {
          if (contains(docID) == false) {
            throw new IllegalArgumentException();
          }
    
          final int index = docID - docBase;
          final int offset = Math.toIntExact(offsets[index]);
          final int length = Math.toIntExact(offsets[index+1]) - offset;
          final int totalLength = Math.toIntExact(offsets[chunkDocs]);
          final int numStoredFields = Math.toIntExact(this.numStoredFields[index]);
    
          final BytesRef bytes;
          if (merging) {
            bytes = this.bytes;
          } else {
            bytes = new BytesRef();
          }
    
          final DataInput documentInput;
          if (length == 0) {
            // empty
            documentInput = new ByteArrayDataInput();
          } else if (merging) {
            // already decompressed
            documentInput = new ByteArrayDataInput(bytes.bytes, bytes.offset + offset, length);
          } else if (sliced) {
            fieldsStream.seek(startPointer);
            decompressor.decompress(fieldsStream, chunkSize, offset, Math.min(length, chunkSize - offset), bytes);
            documentInput = new DataInput() {
    
              int decompressed = bytes.length;
    
              void fillBuffer() throws IOException {
                assert decompressed <= length;
                if (decompressed == length) {
                  throw new EOFException();
                }
                final int toDecompress = Math.min(length - decompressed, chunkSize);
                decompressor.decompress(fieldsStream, toDecompress, 0, toDecompress, bytes);
                decompressed += toDecompress;
              }
    
              @Override
              public byte readByte() throws IOException {
                if (bytes.length == 0) {
                  fillBuffer();
                }
                --bytes.length;
                return bytes.bytes[bytes.offset++];
              }
    
              @Override
              public void readBytes(byte[] b, int offset, int len) throws IOException {
                while (len > bytes.length) {
                  System.arraycopy(bytes.bytes, bytes.offset, b, offset, bytes.length);
                  len -= bytes.length;
                  offset += bytes.length;
                  fillBuffer();
                }
                System.arraycopy(bytes.bytes, bytes.offset, b, offset, len);
                bytes.offset += len;
                bytes.length -= len;
              }
    
            };
          } else {
            fieldsStream.seek(startPointer);
            decompressor.decompress(fieldsStream, totalLength, offset, length, bytes);
            assert bytes.length == length;
            documentInput = new ByteArrayDataInput(bytes.bytes, bytes.offset, bytes.length);
          }
    
          return new SerializedDocument(documentInput, length, numStoredFields);
        }
    View Code

     

    5. 幾個數據結構

      核心的幾個get的外觀數據結構主要有,請求時GetRequest, 查找時DocIdAndVersion, 真正的結果GetResult, 以及響應時的出參GetResponse. 其實前面已經可以查看,不過在此處做個歸集,有意者閱之。

    0. get的請求數據結構

    // org.elasticsearch.action.get.GetRequest
    public class GetRequest extends SingleShardRequest<GetRequest> implements RealtimeRequest {
    
        private String type;
        private String id;
        private String routing;
        private String preference;
    
        private String[] storedFields;
    
        private FetchSourceContext fetchSourceContext;
    
        private boolean refresh = false;
    
        boolean realtime = true;
    
        private VersionType versionType = VersionType.INTERNAL;
        private long version = Versions.MATCH_ANY;
    
        public GetRequest() {
            type = MapperService.SINGLE_MAPPING_NAME;
        }
    
        GetRequest(StreamInput in) throws IOException {
            super(in);
            type = in.readString();
            id = in.readString();
            routing = in.readOptionalString();
            if (in.getVersion().before(Version.V_7_0_0)) {
                in.readOptionalString();
            }
            preference = in.readOptionalString();
            refresh = in.readBoolean();
            storedFields = in.readOptionalStringArray();
            realtime = in.readBoolean();
    
            this.versionType = VersionType.fromValue(in.readByte());
            this.version = in.readLong();
            fetchSourceContext = in.readOptionalWriteable(FetchSourceContext::new);
        }
        ...
    }
    View Code

    1. docId 和 version 包含數據結構

        // org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver.DocIdAndVersion
        /** Wraps an {@link LeafReaderContext}, a doc ID <b>relative to the context doc base</b> and a version. */
        public static class DocIdAndVersion {
            public final int docId;
            public final long version;
            public final long seqNo;
            public final long primaryTerm;
            public final LeafReader reader;
            public final int docBase;
    
            public DocIdAndVersion(int docId, long version, long seqNo, long primaryTerm, LeafReader reader, int docBase) {
                this.docId = docId;
                this.version = version;
                this.seqNo = seqNo;
                this.primaryTerm = primaryTerm;
                this.reader = reader;
                this.docBase = docBase;
            }
        }
    View Code

    2. GetResult 數據結構

    // org.elasticsearch.index.get.GetResult
    public class GetResult implements Writeable, Iterable<DocumentField>, ToXContentObject {
    
        public static final String _INDEX = "_index";
        public static final String _TYPE = "_type";
        public static final String _ID = "_id";
        private static final String _VERSION = "_version";
        private static final String _SEQ_NO = "_seq_no";
        private static final String _PRIMARY_TERM = "_primary_term";
        private static final String FOUND = "found";
        private static final String FIELDS = "fields";
    
        private String index;
        private String type;
        private String id;
        private long version;
        private long seqNo;
        private long primaryTerm;
        private boolean exists;
        private final Map<String, DocumentField> documentFields;
        private final Map<String, DocumentField> metaFields;
        private Map<String, Object> sourceAsMap;
        private BytesReference source;
        private byte[] sourceAsBytes;
    
        public GetResult(StreamInput in) throws IOException {
            index = in.readString();
            type = in.readOptionalString();
            id = in.readString();
            if (in.getVersion().onOrAfter(Version.V_6_6_0)) {
                seqNo = in.readZLong();
                primaryTerm = in.readVLong();
            } else {
                seqNo = UNASSIGNED_SEQ_NO;
                primaryTerm = UNASSIGNED_PRIMARY_TERM;
            }
            version = in.readLong();
            exists = in.readBoolean();
            if (exists) {
                source = in.readBytesReference();
                if (source.length() == 0) {
                    source = null;
                }
                if (in.getVersion().onOrAfter(Version.V_7_3_0)) {
                    documentFields = readFields(in);
                    metaFields = readFields(in);
                } else {
                    Map<String, DocumentField> fields = readFields(in);
                    documentFields = new HashMap<>();
                    metaFields = new HashMap<>();
                    fields.forEach((fieldName, docField) ->
                        (MapperService.META_FIELDS_BEFORE_7DOT8.contains(fieldName) ? metaFields : documentFields).put(fieldName, docField));
                }
            } else {
                metaFields = Collections.emptyMap();
                documentFields = Collections.emptyMap();
            }
        }
    
        public GetResult(String index, String type, String id, long seqNo, long primaryTerm, long version, boolean exists,
                         BytesReference source, Map<String, DocumentField> documentFields, Map<String, DocumentField> metaFields) {
            this.index = index;
            this.type = type;
            this.id = id;
            this.seqNo = seqNo;
            this.primaryTerm = primaryTerm;
            assert (seqNo == UNASSIGNED_SEQ_NO && primaryTerm == UNASSIGNED_PRIMARY_TERM) || (seqNo >= 0 && primaryTerm >= 1) :
                "seqNo: " + seqNo + " primaryTerm: " + primaryTerm;
            assert exists || (seqNo == UNASSIGNED_SEQ_NO && primaryTerm == UNASSIGNED_PRIMARY_TERM) :
                "doc not found but seqNo/primaryTerm are set";
            this.version = version;
            this.exists = exists;
            this.source = source;
            this.documentFields = documentFields == null ? emptyMap() : documentFields;
            this.metaFields = metaFields == null ? emptyMap() : metaFields;
        }
        ...
    }
    View Code

    3. get響應結果 GetResponse

    // org.elasticsearch.action.get.GetResponse
    public class GetResponse extends ActionResponse implements Iterable<DocumentField>, ToXContentObject {
    
        GetResult getResult;
    
        GetResponse(StreamInput in) throws IOException {
            super(in);
            getResult = new GetResult(in);
        }
    
        public GetResponse(GetResult getResult) {
            this.getResult = getResult;
        }
        ...
    }
    View Code

     

    posted @ 2021-05-24 10:19  等你歸去來  閱讀(1095)  評論(0編輯  收藏  舉報
    国产美女a做受大片观看