scroll是ES用來(lái)解決全量遍歷數(shù)據(jù)的功能壶运。具體使用文檔見(jiàn)：

https://www.elastic.co/guide/en/elasticsearch/reference/current/paginate-search-results.html#scroll-search-results

本篇文章將為大家分享scroll的原理凑保。通過(guò)本篇文章樱蛤，大家將會(huì)明白，為什么scroll是怎么做到全量遍歷功能，為什么不推薦使用scroll用作深翻頁(yè)。

先從基本功能誓琼，看看scroll是如何使用的。

scroll是通過(guò)search接口觸發(fā)的肴捉，search接口默認(rèn)最多只能返回size=10000條記錄腹侣，通過(guò)scroll，可以繼續(xù)往后面遍歷數(shù)據(jù)齿穗。這個(gè)是如何做大的呢傲隶？

首先search接口帶上scroll參數(shù)，本次search就會(huì)返回一個(gè)scrollId窃页，比如如下請(qǐng)求：

POST /my-index-000001/_search?scroll=1m
{
  "query": {
    "match": {
      "message": "foo"
    }
  }
}

返回結(jié)果如下：

{
  "_scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFjY0MlkyRFFLVFdDc3RJd0UwT0VKcUEAAAAAAAAblxZfWjZMamlqb1JXeXJlcWtlN0xfUHNR",
  "took" : 11,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 10372087,
      "relation" : "eq"
    },
    ......
  }
}

可以看到返回多了一個(gè)_scroll_id伦籍，然后使用scroll接口，傳遞scroll_id腮出，就能不斷遍歷之前search DSL命中的全部數(shù)據(jù)。

POST /_search/scroll                                                               
{
  "scroll" : "1m",                                                                
  "scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFjY0MlkyRFFLVFdDc3RJd0UwT0VKcUEAAAAAAAAblxZfWjZMamlqb1JXeXJlcWtlN0xfUHNR" 
}

這里scroll設(shè)置為1m芝薇，說(shuō)明scrollId只保存1分鐘胚嘲，一分鐘后，scrollid就消失了洛二，再使用就會(huì)報(bào)如下錯(cuò)誤：

{
  "error" : {
    "root_cause" : [
      {
        "type" : "search_context_missing_exception",
        "reason" : "No search context found for id [7063]"
      }
    ],
    "type" : "search_phase_execution_exception",
    "reason" : "all shards failed",
    "phase" : "query",
    "grouped" : true,
    "failed_shards" : [
      {
        "shard" : -1,
        "index" : null,
        "reason" : {
          "type" : "search_context_missing_exception",
          "reason" : "No search context found for id [7063]"
        }
      }
    ],
    "caused_by" : {
      "type" : "search_context_missing_exception",
      "reason" : "No search context found for id [7063]"
    }
  },
  "status" : 404
}

接下來(lái)看看scroll是如何實(shí)現(xiàn)的馋劈。

scroll是以shard為單位保存scrollId的，

在search query流程中晾嘶，如果search請(qǐng)求傳遞了scroll參數(shù)妓雾，在創(chuàng)建ReaderContext時(shí)，就會(huì)創(chuàng)建的是LegacyReaderContext垒迂，LegacyReaderContext中會(huì)生成一個(gè)ScrollContext械姻。我們來(lái)看下ScrollContext的內(nèi)容：

public final class ScrollContext {
    public TotalHits totalHits = null;
    public float maxScore = Float.NaN;
    public ScoreDoc lastEmittedDoc;
    public Scroll scroll;
}

其中有個(gè)關(guān)鍵變量lastEmittedDoc，這個(gè)記錄了上次scroll遍歷到的docId位置机断。

以一次遍歷10000天記錄楷拳，遍歷索引全部數(shù)據(jù)為例绣夺，之前遍歷到了10000條doc，那么lastEmittedDoc就是10000欢揖，再執(zhí)行一次scroll陶耍，就能繼續(xù)往后遍歷10000條，lastEmittedDoc就設(shè)置為20000她混。所以scroll就能做到不斷往后遍歷烈钞，直到遍歷全量數(shù)據(jù)。

這里lastEmittedDoc還需要注意坤按，在多shard時(shí)毯欣，實(shí)際lastEmittedDoc的值，要以shard實(shí)際fetch到的數(shù)據(jù)為準(zhǔn)晋涣，所以lastEmittedDoc是在SearchService.executeFetchPhase方法中設(shè)置：

public void executeFetchPhase(ShardFetchRequest request, SearchShardTask task, ActionListener<FetchSearchResult> listener) {
        final ReaderContext readerContext = findReaderContext(request.contextId(), request);
        final ShardSearchRequest shardSearchRequest = readerContext.getShardSearchRequest(request.getShardSearchRequest());
        final Releasable markAsUsed = readerContext.markAsUsed(getKeepAlive(shardSearchRequest));
        runAsync(getExecutor(readerContext.indexShard()), () -> {
            try (SearchContext searchContext = createContext(readerContext, shardSearchRequest, task, false)) {
                if (request.lastEmittedDoc() != null) {
                    searchContext.scrollContext().lastEmittedDoc = request.lastEmittedDoc();
                }
                searchContext.assignRescoreDocIds(readerContext.getRescoreDocIds(request.getRescoreDocIds()));
                searchContext.searcher().setAggregatedDfs(readerContext.getAggregatedDfs(request.getAggregatedDfs()));
                searchContext.docIdsToLoad(request.docIds());
......
                return searchContext.fetchResult();
            } catch (Exception e) {
                assert TransportActions.isShardNotAvailableException(e) == false : new AssertionError(e);
                // we handle the failure in the failure listener below
                throw e;
            }
        }, wrapFailureListener(listener, readerContext, markAsUsed));
    }

然后scroll的是保存在SearchService的activeReaders對(duì)象中仪媒，activeReaders是一個(gè)Map對(duì)象，有個(gè)Reaper會(huì)定時(shí)檢查過(guò)期的ReaderContext谢鹊，將它從activeReaders清除：

//默認(rèn)一分鐘檢查一次
public static final Setting<TimeValue> KEEPALIVE_INTERVAL_SETTING = Setting.positiveTimeSetting(
        "search.keep_alive_interval",
        timeValueMinutes(1),
        Property.NodeScope
    );

// 啟動(dòng)清理定時(shí)任務(wù)
TimeValue keepAliveInterval = KEEPALIVE_INTERVAL_SETTING.get(settings);
this.keepAliveReaper = threadPool.scheduleWithFixedDelay(new Reaper(), keepAliveInterval, Names.SAME);

    //清理任務(wù)
    class Reaper implements Runnable {
        @Override
        public void run() {
            for (ReaderContext context : activeReaders.values()) {
                if (context.isExpired()) {
                    logger.debug("freeing search context [{}]", context.id());
                    freeReaderContext(context.id());
                }
            }
        }
    }

所以手動(dòng)刪除scrollId的接口算吩，實(shí)現(xiàn)方式也很簡(jiǎn)單，就是清理掉ReaderContext對(duì)應(yīng)的scrollId佃扼。

接下來(lái)再看下scrollId的編碼偎巢。scrollId在ReaderContext中是以Long類型的id保存的。

然后在協(xié)調(diào)節(jié)點(diǎn)兼耀，會(huì)將全部shard的scrollId進(jìn)行編碼成一個(gè)字符串返回給客戶端压昼。

在TransportSearchHelper內(nèi)中，有buildScrollId和parseScrollId兩個(gè)方法用來(lái)進(jìn)行編解碼：

static String buildScrollId(AtomicArray<? extends SearchPhaseResult> searchPhaseResults) {
        try {
            BytesStreamOutput out = new BytesStreamOutput();
            out.writeString(INCLUDE_CONTEXT_UUID);
            out.writeString(searchPhaseResults.length() == 1 ? ParsedScrollId.QUERY_AND_FETCH_TYPE : ParsedScrollId.QUERY_THEN_FETCH_TYPE);
            out.writeCollection(searchPhaseResults.asList(), (o, searchPhaseResult) -> {
                o.writeString(searchPhaseResult.getContextId().getSessionId());
                o.writeLong(searchPhaseResult.getContextId().getId());
                SearchShardTarget searchShardTarget = searchPhaseResult.getSearchShardTarget();
                if (searchShardTarget.getClusterAlias() != null) {
                    o.writeString(
                        RemoteClusterAware.buildRemoteIndexName(searchShardTarget.getClusterAlias(), searchShardTarget.getNodeId())
                    );
                } else {
                    o.writeString(searchShardTarget.getNodeId());
                }
            });
            return Base64.getUrlEncoder().encodeToString(out.copyBytes().array());
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    static ParsedScrollId parseScrollId(String scrollId) {
        try {
            byte[] bytes = Base64.getUrlDecoder().decode(scrollId);
            ByteArrayStreamInput in = new ByteArrayStreamInput(bytes);
            final boolean includeContextUUID;
            final String type;
            final String firstChunk = in.readString();
            if (INCLUDE_CONTEXT_UUID.equals(firstChunk)) {
                includeContextUUID = true;
                type = in.readString();
            } else {
                includeContextUUID = false;
                type = firstChunk;
            }
            SearchContextIdForNode[] context = new SearchContextIdForNode[in.readVInt()];
            for (int i = 0; i < context.length; ++i) {
                final String contextUUID = includeContextUUID ? in.readString() : "";
                long id = in.readLong();
                String target = in.readString();
                String clusterAlias;
                final int index = target.indexOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR);
                if (index == -1) {
                    clusterAlias = null;
                } else {
                    clusterAlias = target.substring(0, index);
                    target = target.substring(index + 1);
                }
                context[i] = new SearchContextIdForNode(clusterAlias, target, new ShardSearchContextId(contextUUID, id));
            }
            if (in.getPosition() != bytes.length) {
                throw new IllegalArgumentException("Not all bytes were read");
            }
            return new ParsedScrollId(scrollId, type, context);
        } catch (Exception e) {
            throw new IllegalArgumentException("Cannot parse scroll id", e);
        }
    }

Scroll在5.x版本還增加另一個(gè)Sliced scroll功能瘤运。這是可以支持對(duì)一個(gè)DSL進(jìn)行并發(fā)scroll窍霞，提高拉取數(shù)據(jù)的性能。

Sliced scroll的原理是在scroll基礎(chǔ)上增加了分片功能拯坟。建議分片是按照shard數(shù)量的倍數(shù)來(lái)絮短。在一個(gè)shard中的sliced還能進(jìn)一步切分绒怨。

實(shí)現(xiàn)原理如下：

在DSL傳遞slice參數(shù)后，生成的Query，會(huì)wrapper上slice的query改基，具體代碼在SliceBuilder.createSliceQuery中：

    private Query createSliceQuery(int id, int max, SearchExecutionContext context, boolean isScroll) {
        if (field == null) {
            return isScroll ? new TermsSliceQuery(IdFieldMapper.NAME, id, max) : new DocIdSliceQuery(id, max);
        } else if (IdFieldMapper.NAME.equals(field)) {
            if (isScroll == false) {
                throw new IllegalArgumentException("cannot slice on [_id] when using [point-in-time]");
            }
            return new TermsSliceQuery(IdFieldMapper.NAME, id, max);
        } else {
            MappedFieldType type = context.getFieldType(field);
            if (type == null) {
                throw new IllegalArgumentException("field " + field + " not found");
            }
            if (type.hasDocValues() == false) {
                throw new IllegalArgumentException("cannot load numeric doc values on " + field);
            } else {
                IndexFieldData<?> ifm = context.getForField(type, MappedFieldType.FielddataOperation.SEARCH);
                if (ifm instanceof IndexNumericFieldData == false) {
                    throw new IllegalArgumentException("cannot load numeric doc values on " + field);
                }
                return new DocValuesSliceQuery(field, id, max);
            }
        }
    }

wrapper的slice query根據(jù)field的類型包括TermsSliceQuery魂奥、DocIdSliceQuery童社、DocValuesSliceQuery溯警。

他們的功能都是根據(jù)傳遞的id、max值年柠，確定在命中的docid集合中只返回slice id對(duì)應(yīng)的docid凿歼。這樣原DSL就只能返回slice query對(duì)應(yīng)的docId列表。

以上就是scroll的原理實(shí)現(xiàn)，所以可以看到scroll只能往后面遍歷毅往，而且設(shè)置了有效期牵咙，所以如果頻繁設(shè)置scroll用來(lái)進(jìn)行深翻頁(yè)，會(huì)導(dǎo)致生成過(guò)多的scrollId攀唯，而且scroll也不能跳頁(yè)洁桌，所以從功能定位上scroll是用來(lái)遍歷全量數(shù)據(jù)使用的。如果要進(jìn)行翻頁(yè)侯嘀，ES推薦的使用Search After功能另凌。

使用scroll還需要注意一個(gè)問(wèn)題，如果訪問(wèn)出現(xiàn)了超時(shí)戒幔，由于不確定本次scroll是否已經(jīng)進(jìn)行吠谢，所以可能導(dǎo)致獲取的數(shù)據(jù)缺失，所以要進(jìn)行全部重試诗茎，這里也可以用一些技巧工坊，通過(guò)修改DSL，定位到最新獲取的數(shù)據(jù)敢订，再重新使用scroll查詢也是可以的王污。