PUT wkl_test/_settings
{
   "index":{
        "max_result_window":2147483647
    }
}

GET wkl_test/_search?scroll=5m
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "seq": {
        "order": "asc"
      }
    }
  ],
  "size": 200
}

@Test
    public void testScroll(){
        RestHighLevelClient restHighLevelClient ;
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.mustNot(QueryBuilders.existsQuery("seq"));

        try {
            //滾動(dòng)查詢(xún)的Scroll,設(shè)置請(qǐng)求滾動(dòng)時(shí)間窗口時(shí)間
            Scroll scroll = new Scroll(TimeValue.timeValueMillis(180000));

            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            //加入query語(yǔ)句
            sourceBuilder.query(boolQueryBuilder);
            //每次滾動(dòng)的長(zhǎng)度
            sourceBuilder.size(SIZE);
            //加入排序字段
            sourceBuilder.sort("id", SortOrder.DESC);
            //構(gòu)建searchRequest
            //加入scroll和構(gòu)造器
            SearchRequest searchRequest = new SearchRequest()
                    .indices("wkl_test")
                    .source(sourceBuilder)
                    .scroll(scroll);
            //存儲(chǔ)scroll的list
            List<String> scrollIdList = new ArrayList<>();
            //執(zhí)行首次檢索
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            //首次檢索返回scrollId，用于下一次的滾動(dòng)查詢(xún)
            String scrollId = searchResponse.getScrollId();
            //拿到hits結(jié)果
            SearchHit[] hits = searchResponse.getHits().getHits();
            long value = searchResponse.getHits().getTotalHits().value;
            //保存返回結(jié)果List大小
            Long resultSize = 0L;
            scrollIdList.add(scrollId);
            try {
                //滾動(dòng)查詢(xún)將SearchHit封裝到result中
                while (ArrayUtils.isNotEmpty(hits) && hits.length > 0) {
                    BulkRequest bulkRequest = new BulkRequest();
                    JSONArray esArray = new JSONArray();
                    for (SearchHit hit : hits) {
                        String sourceAsString = hit.getSourceAsString();
                        String index = hit.getIndex();
                        JSONObject jsonObject = JSONObject.parseObject(sourceAsString);
                        String seq = jsonObject.getString("seq");
                        if(StringUtils.isBlank(seq) ){
                            esArray.add(jsonObject);
                            String uuid = jsonObject.getString("id");
                            jsonObject.put("is_del",1);
                            bulkRequest.add(new UpdateRequest(index, uuid).doc(jsonObject));
                        }
                    }
                    resultSize = resultSize+hits.length;

                    //發(fā)送請(qǐng)求
                    //實(shí)時(shí)更新
                    bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
                    BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    System.out.println(bulk.getTook()+"-------"+bulk.getItems().length);

                    //說(shuō)明滾動(dòng)完了，返回結(jié)果即可
                    if (resultSize > 20000) {
                        break;
                    }
                    //繼續(xù)滾動(dòng)，根據(jù)上一個(gè)游標(biāo)，得到這次開(kāi)始查詢(xún)位置
                    SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                    searchScrollRequest.scroll(scroll);
                    //得到結(jié)果
                    SearchResponse searchScrollResponse = restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
                    //定位游標(biāo)
                    scrollId = searchScrollResponse.getScrollId();
                    hits = searchScrollResponse.getHits().getHits();
                    scrollIdList.add(scrollId);
                }
                System.out.println("----徹底結(jié)束了-----");
            } finally {
                //清理scroll,釋放資源
                ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                clearScrollRequest.setScrollIds(scrollIdList);
                restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

GET wkl_test/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "seq": {
        "order": "asc"
      }
    }
  ],
  "size": 200
}

@Test
    public void testSearchAfter() throws IOException {
        RestHighLevelClient restHighLevelClient = es7UtilApi.getRestHighLevelClient();

        MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(matchAllQueryBuilder);
        searchSourceBuilder.from(0);
        searchSourceBuilder.size(200);
        searchSourceBuilder.sort("seq", SortOrder.ASC);
        searchSourceBuilder.trackTotalHits(true);

        SearchRequest searchRequest = new SearchRequest()
                .indices("wkl_test")
                .source(searchSourceBuilder);

        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();
        long value = hits.getTotalHits().value;
        System.out.println("查詢(xún)到記錄數(shù)=" + value);

        List<JSONObject> list = new ArrayList<>();
        SearchHit[] searchHists = hits.getHits();
        Object[] sortValues = searchHists[searchHists.length - 1].getSortValues();
        if (searchHists.length > 0) {
            for (SearchHit hit : searchHists) {
                String sourceAsString = hit.getSourceAsString();
                JSONObject jsonObject = JSON.parseObject(sourceAsString);
                jsonObject.put("_id", hit.getId());
                list.add(jsonObject);
            }
        }

        //往后的每次請(qǐng)求都攜帶上一次的sort_id進(jìn)行訪(fǎng)問(wèn)。
        while (ArrayUtils.isNotEmpty(searchHists) && searchHists.length > 0){
            searchSourceBuilder.searchAfter(sortValues);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponseAfter = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            hits = searchResponseAfter.getHits();
            searchHists = hits.getHits();
            sortValues = searchHists[searchHists.length - 1].getSortValues();
            if (searchHists.length > 0) {
                for (SearchHit hit : searchHists) {
                    String sourceAsString = hit.getSourceAsString();
                    JSONObject jsonObject = JSON.parseObject(sourceAsString);
                    jsonObject.put("_id", hit.getId());
                    list.add(jsonObject);
                }
            }
            if(list.size()>20000){
                break;
            }
            System.out.println("-----徹底結(jié)束了-------");
        }

    }

#keep_alive必須要加上，它表示這個(gè)pit能存在多久，這里設(shè)置的是1分鐘
POST wkl_test/_pit?keep_alive=1m

GET _search
{
  "query": {
    "match_all": {}
  },
  "pit":{
    "id":"t_yxAwEId2tsX3Rlc3QWU0hzbEJkYWNTVEd0ZGRoN0xsQVVNdwAWUGQtaXJpT0xTa2VUN0RGLXZfTlBvZwAAAAAACHG1fxY1UWNKX1RHOFMybXBaV20zbWx3enp3ARZTSHNsQmRhY1NUR3RkZGg3TGxBVU13AAA=",
    "keep_alive":"5m"
  },
  "sort": [
    {
      "seq": {
        "order": "asc"
      }
    }
  ],
  "size": 200
}

DELETE _pit
{
 "id":"t_yxAwEId2tsX3Rlc3QWU0hzbEJkYWNTVEd0ZGRoN0xsQVVNdwAWUGQtaXJpT0xTa2VUN0RGLXZfTlBvZwAAAAAACHG1fxY1UWNKX1RHOFMybXBaV20zbWx3enp3ARZTSHNsQmRhY1NUR3RkZGg3TGxBVU13AAA="
}