Java使用ES Client 调用滚动查询及Elasticsearch滚动查询Scrolling机制

POST /your_index/_search?scroll=5m
{
  "size": 100,       // 每次返回的结果数量
  "query": { ... }   // 查询条件
}

POST /_search/scroll
{
  "scroll": "5m",
  "scroll_id": "DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAD4WYm9laVYtZndUQlNsdDcwakFMNjU1QQ=="
}

POST /_search/scroll
 {
   "scroll": "5m",
   "scroll_id": "FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFDJPRXc0WWdCY1BLWlo1MTk4MmR3AAAAAAAAAXYWcWgwSW5CQUtScEd2T2QtRGtYaWliQQ=="
 }

DELETE /_search/scroll
{
  "scroll_id": [
    "DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAD4WYm9laVYtZndUQlNsdDcwakFMNjU1QQ=="
  ]
}

public static void main(String[] args) {
    long start = System.currentTimeMillis();
    //构建es HttpHost对象
    HttpHost httpHost1 = new HttpHost("192.168.1.1", 9200, "http");
    // 滚动时间窗口
    long scrollTime = 1L;
    // 每次返回的文档数量
    int batchSize = 20000;
    //索引名
    String indexName = "你的索引名称";
    try (RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(httpHost1))) {
        //构建查询请求
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.boolQuery());
        searchSourceBuilder.size(batchSize);
        //设置查询返回字段
        String[] includes = {};
        searchSourceBuilder.fetchSource(includes, null);
        // 滚动查询请求
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(searchSourceBuilder);
        //设置请求滚动时间窗口时间
        searchRequest.scroll(TimeValue.timeValueMinutes(scrollTime));
        //执行首次检索
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        //首次检索返回scrollId，用于下一次的滚动查询
        String scrollId = searchResponse.getScrollId();
        //获取首次检索命中结果
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        //计数
        int count = 0;
        // 处理第一批结果
        for (SearchHit hit : searchHits) {
            // 处理单个文档
            JSONObject dataJson = (JSONObject) JSON.parse(hit.getSourceAsString());
            System.out.println("====对首次请求的进行处理，当前计数：" + count++);
        }
        // 处理滚动结果
        while (searchHits != null && searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(TimeValue.timeValueMinutes(scrollTime));
            searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
            for (SearchHit hit : searchHits) {
                JSONObject dataJson = (JSONObject) JSON.parse(hit.getSourceAsString());
                System.out.println("====滚动查询，当前计数：" + count++);
            }
        }
        // 清理滚动上下文
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        boolean succeeded = clearScrollResponse.isSucceeded();
        long end = System.currentTimeMillis();
        System.out.println("共执行时间：" + (end - start) / 1000 + " s");
    } catch (Exception e) {
        System.out.println("===error==" + e.getMessage());
        e.printStackTrace();
    }
}

import org.elasticsearch.action.search .*;
import org.elasticsearch.client .*;
import org.elasticsearch.common.unit .*;
import org.elasticsearch.index.query .*;
import org.elasticsearch.search .*;
import org.elasticsearch.search.builder .*;
import org.springframework.beans.factory.annotation .*;
import org.springframework.data.elasticsearch.core .*;
import org.springframework.data.elasticsearch.core.query .*;
public class ScrollSearchExample {
    @Autowired
    private ElasticsearchOperations elasticsearchOperations;
    public void performScrollSearch() {
        String scrollTime = "1m";  // 滚动时间窗口
        int batchSize = 100;  // 每次返回的文档数量
        QueryBuilder queryBuilder = QueryBuilders.matchQuery("field", "value");
        NativeSearchQueryBuilder searchQuery = new NativeSearchQueryBuilder();
        searchQuery.withQuery(queryBuilder).withPageable(PageRequest.of(0, batchSize)).build();
        SearchResponse searchResponse = elasticsearchOperations.startScroll(
                scrollTime,
                searchQuery,
                YourEntityClass.class,
                IndexCoordinates.of("your_index")
        );
        String scrollId = searchResponse.getScrollId();
        SearchHits<YourEntityClass> searchHits = searchResponse.getSearchHits();
        // 处理第一批结果
        for (SearchHit<YourEntityClass> hit : searchHits) {
            YourEntityClass entity = hit.getContent();
            // 处理单个文档
        }
        // 处理滚动结果
        while (searchHits != null && searchHits.hasSearchHits()) {
            searchResponse = elasticsearchOperations.continueScroll(scrollId, scrollTime, YourEntityClass.class);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getSearchHits();
            for (SearchHit<YourEntityClass> hit : searchHits) {
                YourEntityClass entity = hit.getContent();
                // 处理单个文档
            }
        }
        // 清理滚动上下文
        elasticsearchOperations.clearScroll(scrollId);
    }
}