elasticsearch滚动查询工具类 (集成版)

elasticsearch滚动查询工具类 (集成版)

    /*public static void sendscrollSearchData(TransportClient client, SearchRequest searchRequest) {
        SearchResponse searchResponse = client.search(searchRequest).actionGet();
        KafkaSink kafkaSink = new KafkaSink();
        kafkaSink.open();

        assert searchResponse != null;
        String scrollId = sendData(searchResponse,kafkaSink);

        while (searchResponse.getHits().getHits().length > 0) {
            if (scrollId == null) {
                break;
            }
            SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId).scroll(new Scroll(new Timevalue(60, TimeUnit.SECONDS)));
            searchResponse = client.searchScroll(searchScrollRequest).actionGet();
            scrollId = sendData(searchResponse,kafkaSink);
        }
        //清除滚屏
        try {
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            client.clearScroll(clearScrollRequest).actionGet();
        } catch (Exception e) {
            log.error("执行es滚动查询数据时发生故障", e);
        }
        kafkaSink.close();
    }