@Overrideprotected void doExecute(CountRequest request, ActionListener<CountResponse> listener) {request.nowInMillis = System.currentTimeMillis();super.doExecute(request, listener);}

 @Overrideprotected void doExecute(Request request, ActionListener<Response> listener) {new AsyncBroadcastAction(request, listener).start();}

public void start() {　　　　　　//没有shardsif (shardsIts.size() == 0) {// no shardstry {listener.onResponse(newResponse(request, new AtomicReferenceArray(0), clusterState));} catch (Throwable e) {listener.onFailure(e);}return;}request.beforeStart();// count the local operations, and perform the non local onesint shardIndex = -1;　　　　　　　//遍历对每个shards进行操作for (final ShardIterator shardIt : shardsIts) {shardIndex++;final ShardRouting shard = shardIt.nextOrNull();if (shard != null) {performOperation(shardIt, shard, shardIndex);} else {// really, no shards active in this grouponOperation(null, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));}}}

 protected void performOperation(final ShardIterator shardIt, final ShardRouting shard, final int shardIndex) {if (shard == null) {//shard 为null抛出异常// no more active shards... (we should not really get here, just safety)onOperation(null, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));} else {try {final ShardRequest shardRequest = newShardRequest(shardIt.size(), shard, request);if (shard.currentNodeId().equals(nodes.localNodeId())) {//shard在本地执行shardOperation方法，并通过onOperation方法封装结果threadPool.executor(executor).execute(new Runnable() {@Overridepublic void run() {try {onOperation(shard, shardIndex, shardOperation(shardRequest));} catch (Throwable e) {onOperation(shard, shardIt, shardIndex, e);}}});} else {//不是本地shard，发送到对应节点。DiscoveryNode node = nodes.get(shard.currentNodeId());if (node == null) {// no node connected, act as failureonOperation(shard, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));} else {transportService.sendRequest(node, transportShardAction, shardRequest, new BaseTransportResponseHandler<ShardResponse>() {@Overridepublic ShardResponse newInstance() {return newShardResponse();}@Overridepublic String executor() {return ThreadPool.Names.SAME;}@Overridepublic void handleResponse(ShardResponse response) {onOperation(shard, shardIndex, response);}@Overridepublic void handleException(TransportException e) {onOperation(shard, shardIt, shardIndex, e);}});}}} catch (Throwable e) {onOperation(shard, shardIt, shardIndex, e);}}}

    @Overrideprotected ShardCountResponse shardOperation(ShardCountRequest request) throws ElasticsearchException {IndexService indexService = indicesService.indexServiceSafe(request.shardId().getIndex());//IndexShard indexShard = indexService.shardSafe(request.shardId().id());//构造查询contextSearchShardTarget shardTarget = new SearchShardTarget(clusterService.localNode().id(), request.shardId().getIndex(), request.shardId().id());SearchContext context = new DefaultSearchContext(0,new ShardSearchLocalRequest(request.types(), request.nowInMillis(), request.filteringAliases()),shardTarget, indexShard.acquireSearcher("count"), indexService, indexShard,scriptService, cacheRecycler, pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter());SearchContext.setCurrent(context);try {// TODO: min score should move to be "null" as a value that is not initialized...if (request.minScore() != -1) {context.minimumScore(request.minScore());}BytesReference source = request.querySource();if (source != null && source.length() > 0) {try {QueryParseContext.setTypes(request.types());context.parsedQuery(indexService.queryParserService().parseQuery(source));} finally {QueryParseContext.removeTypes();}}final boolean hasTerminateAfterCount = request.terminateAfter() != DEFAULT_TERMINATE_AFTER;boolean terminatedEarly = false;context.preProcess();try {long count;if (hasTerminateAfterCount) {//调用lucene的封装接口执行查询并返回结果final Lucene.EarlyTerminatingCollector countCollector =Lucene.createCountBasedEarlyTerminatingCollector(request.terminateAfter());terminatedEarly = Lucene.countWithEarlyTermination(context.searcher(), context.query(), countCollector);count = countCollector.count();} else {count = Lucene.count(context.searcher(), context.query());}return new ShardCountResponse(request.shardId(), count, terminatedEarly);} catch (Exception e) {throw new QueryPhaseExecutionException(context, "failed to execute count", e);}} finally {// this will also release the index searcher
            context.close();SearchContext.removeCurrent();}}