elasticsearch restHighLevelClient 游标查询全量数据示例

Love The Way You Lie 2021-10-19 08:38 845阅读 0赞

# 查询所有数据 #

我这边用了很多次，所以我抽成了工具类

## 调用 ##

//admin_log是索引名
    			SearchRequest searchRequest = new SearchRequest("admin_log");
    			SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    			searchSourceBuilder.query(QueryBuilders.matchQuery("app", "xxx"));
    			searchRequest.source(searchSourceBuilder);
    			try { 
    				//SCROLL_TIMEOUT是快照保存时间
                    List<SearchHit> searchHits = SearchHitUtils.scrollSearchAll(restHighLevelClient, SCROLL_TIMEOUT, searchRequest);
                    //logList就是查询到的所有结果了
                    List<AdminLog> logList= SearchHitUtils.getAdminLogForSearchHit(searchHits);
                } catch (IOException e) { 
                    LOGGER.error("################ 高危操作日志游标查询异常！ ##################");
                } catch (IllegalAccessException | InvocationTargetException e) { 
                    LOGGER.error("################ 高危操作日志游标查询结果，转换AdminLog对象出现异常！ ##################");
                }

## 工具类 ##

import com.vivo.admin_log_web.entity.AdminLog;
    import org.apache.commons.beanutils.BeanUtils;
    import org.apache.commons.lang3.ArrayUtils;
    import org.elasticsearch.action.search.ClearScrollRequest;
    import org.elasticsearch.action.search.SearchRequest;
    import org.elasticsearch.action.search.SearchResponse;
    import org.elasticsearch.action.search.SearchScrollRequest;
    import org.elasticsearch.client.RestHighLevelClient;
    import org.elasticsearch.search.Scroll;
    import org.elasticsearch.search.SearchHit;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.lang.reflect.InvocationTargetException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    
    import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
    
    /** * @author GaoJinShan */
    public class SearchHitUtils { 
        public static final Logger LOGGER = LoggerFactory.getLogger(SearchHitUtils.class);
    
        private SearchHitUtils() { 
        }
    
        /** * SearchHit数组转换成AdminLog对象集合 * * @param resultHits * @return * @throws InvocationTargetException * @throws IllegalAccessException */
        public static List<AdminLog> getAdminLogForSearchHit(SearchHit[] resultHits) throws InvocationTargetException, IllegalAccessException { 
            ArrayList<AdminLog> adminLogs = new ArrayList<>();
            for (SearchHit hit : resultHits) { 
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                AdminLog adminLog = new AdminLog();
                BeanUtils.populate(adminLog, sourceAsMap);
                adminLogs.add(adminLog);
            }
            return adminLogs;
        }
    
        /** * SearchHit集合转换成AdminLog对象集合 * * @param resultHits * @return * @throws InvocationTargetException * @throws IllegalAccessException */
        public static List<AdminLog> getAdminLogForSearchHit(List<SearchHit> resultHits) throws InvocationTargetException, IllegalAccessException { 
            ArrayList<AdminLog> adminLogs = new ArrayList<>();
            for (SearchHit hit : resultHits) { 
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                AdminLog adminLog = new AdminLog();
                BeanUtils.populate(adminLog, sourceAsMap);
                adminLogs.add(adminLog);
            }
            return adminLogs;
        }
    
        /** * 使用游标获取全部结果，返回SearchHit集合 * * @param restHighLevelClient * @param scrollTimeOut * @param searchRequest * @return * @throws IOException */
        public static List<SearchHit> scrollSearchAll(RestHighLevelClient restHighLevelClient, Long scrollTimeOut, SearchRequest searchRequest) throws IOException { 
            Scroll scroll = new Scroll(timeValueMillis(scrollTimeOut));
            searchRequest.scroll(scroll);
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest);
            String scrollId = searchResponse.getScrollId();
            SearchHit[] hits = searchResponse.getHits().getHits();
            List<SearchHit> resultSearchHit = new ArrayList<>();
            while (ArrayUtils.isNotEmpty(hits)) { 
                for (SearchHit hit : hits) { 
                    resultSearchHit.add(hit);
                }
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                SearchResponse searchScrollResponse = restHighLevelClient.searchScroll(searchScrollRequest);
                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits().getHits();
            }
            //及时清除es快照，释放资源
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            restHighLevelClient.clearScroll(clearScrollRequest);
            return resultSearchHit;
        }
    }

**注：** 本文是参考官网然后自己摸索修改的，如有问题请大佬指正