ElasticSearch的scroll分批查询

落日映苍穹つ 2022-12-27 01:59 165阅读 0赞

## 项目集成RestHighLeverConfig ##

1.  application.yml添加配置参数

#ES集群rest接口
    rest-elasticsearch:
      ipAndPorts: 192.168.8.23:9200,192.168.8.24:9200,192.168.8.25:9200
      schema: http
      connectTimeOut: 1000
      socketTimeOut: 30000
      connectionRequestTimeOut: 500
      maxConnectNum: 100
      maxConnectPerRoute: 100
      username: elastic
      password: 123456

1.  配置类代码

@Configuration
    public class ESRestHighLevelConfig {
        private static final Logger LOGGER  = LoggerFactory.getLogger(ESRestHighLevelConfig.class);
        private static final String CLUSTER_NODES_SPLIT_SYMBOL = ",";
        private static final String HOST_PORT_SPLIT_SYMBOL = ":";
    
        @Value("${rest-elasticsearch.ipAndPorts}")
        private String ipAndPorts;
        @Value("${rest-elasticsearch.schema}")
        private String schema;
        @Value("${rest-elasticsearch.connectTimeOut}")
        private int connectTimeOut;
        @Value("${rest-elasticsearch.socketTimeOut}")
        private int socketTimeOut;
        @Value("${rest-elasticsearch.connectionRequestTimeOut}")
        private int connectionRequestTimeOut;
        @Value("${rest-elasticsearch.maxConnectNum}")
        private int maxConnectNum;
        @Value("${rest-elasticsearch.maxConnectPerRoute}")
        private int maxConnectPerRoute;
        @Value("${rest-elasticsearch.username}")
        private String username;
        @Value("${rest-elasticsearch.password}")
        private String password;
    
        @Bean(name = "restHighLevelClient")
        public RestHighLevelClient restClient() {
            return new RestHighLevelClient(initRestClientBuilder());
        }
    
        /**
         * 初始化ES配置信息
         */
        private RestClientBuilder  initRestClientBuilder() {
    
            LOGGER.info("elasticsearch  初始化...");
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
            String[] clusterNodeArray = ipAndPorts.trim().split(CLUSTER_NODES_SPLIT_SYMBOL);
            HttpHost[] httpHosts = new HttpHost[clusterNodeArray.length];
            for (int i = 0; i < clusterNodeArray.length;i++) {
                String[] clusterNodeInfoArray = clusterNodeArray[i].trim().split(HOST_PORT_SPLIT_SYMBOL);
                httpHosts[i] = new HttpHost(clusterNodeInfoArray[0],Integer.parseInt(clusterNodeInfoArray[1]),schema);
            }
            RestClientBuilder restClientBuilder = RestClient.builder(httpHosts)
                    .setHttpClientConfigCallback(httpClientBuilder -> {
                        //httpclient的连接数配置
                        httpClientBuilder.setMaxConnTotal(maxConnectNum);
                        httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
                        //设置认证信息
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    })
                    .setRequestConfigCallback(requestConfigBuilder -> {
                        requestConfigBuilder.setConnectTimeout(connectTimeOut);
                        requestConfigBuilder.setSocketTimeout(socketTimeOut);
                        requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
                        return requestConfigBuilder;
                    });
    
            LOGGER.info("elasticsearch  初始化成功...");
    
            return restClientBuilder;
        }
    
    }

1.  业务层

@Controller
    @RequestMapping("testEsScroll")
    public class TestEsController {
        @Autowired
        ESRestHighLevelConfig esRestHighLevelConfig;
    
        @RequestMapping(value = "scroll")
        @ResponseBody
        public void scroll(){
            RestHighLevelClient client = esRestHighLevelConfig.restClient();
            //初始化scroll
            final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1));
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("poc"); //指定索引名称
            searchRequest.scroll(scroll);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            //设置查询条件，如果要查询全部就用 
            // QueryBuilder mathAll = QueryBuilders.matchAllQuery();
            //多条件查询
            QueryBuilder q1 = QueryBuilders.termQuery("channelCode", "onlineBank");
             QueryBuilder q2 = QueryBuilders.rangeQuery("transTime").gte(1565937193000L).lte(1566973993000L);
            BoolQueryBuilder must = QueryBuilders.boolQuery().must(q1).must(q2);
            QueryBuilder mathAll = QueryBuilders.matchAllQuery();
            searchSourceBuilder.query(must);
            //searchSourceBuilder.query(QueryBuilders.termQuery("channelCode","onlineBank")); //term精准查询
            //searchSourceBuilder.query(QueryBuilders.rangeQuery("age").gte(20)); // 范围查询
            //searchSourceBuilder.query(QueryBuilders.rangeQuery("age").from(10).to(20)); //闭区间，范围查询
            //searchSourceBuilder.query(QueryBuilders.rangeQuery("age").from(15,false).to(20,false)); //开区间，范围查询
            searchSourceBuilder.size(10);  //每次返回的数据条数
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = null;
            try {
                searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            String scrollId = searchResponse.getScrollId();  //首次查询返回的油标scrollID
            SearchHit[] searchHits = searchResponse.getHits().getHits();  //获取结果数据
            for (SearchHit searchHit : searchHits) {
                System.out.println(searchHit.getSourceAsString());
            }
            //遍历搜索命中的数据，直到没有数据
            int count = 10;
            while (searchHits != null && searchHits.length > 0){
                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                try {
                    searchResponse = client.scroll(scrollRequest,RequestOptions.DEFAULT);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                scrollId = searchResponse.getScrollId(); //每次查询后更新scrollId
                searchHits = searchResponse.getHits().getHits();
                if(searchHits != null && searchHits.length > 0){
                    System.out.println("---------下一页---------");
                    for (SearchHit searchHit : searchHits) {
                        System.out.println(searchHit.getSourceAsString());
                    }
                }
                count = count + searchHits.length;
            }
            //清除滚屏
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            ClearScrollResponse clearScrollResponse = null;
            try {
                clearScrollResponse = client.clearScroll(clearScrollRequest,RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            boolean succeeded = clearScrollResponse.isSucceeded();
            System.out.println("succeeded: "+succeeded+",  共接收交易数据:"+count+" 条");
        }
    }