ElasticSearch 海量数据查询

1、ES数据格式配置 【ElasticSearch 海量数据查询】/** * ES数据格式配置 * * @author Lance * @version 1.0 * @data 2022/03/24 23:12 */public class SearchAllBuilder {/*** 索引名称*/private String indices;/*** 类型*/private String types;/*** 查询条件*/private SearchSourceBuilder searchSourceBuilder;/*** 符号*/private RequestOptions options = RequestOptions.DEFAULT;/*** 属性命名策略*/private Integer propertyNamingStrategy = Constant.ORIGINAL;public SearchAllBuilder() {}public SearchAllBuilder indices(String indices) {this.indices = indices;return this;}public SearchAllBuilder types(String types) {this.types = types;return this;}public SearchAllBuilder searchSourceBuilder(SearchSourceBuilder searchSourceBuilder) {this.searchSourceBuilder = searchSourceBuilder;return this;}public SearchAllBuilder options(RequestOptions options) {this.options = options;return this;}public SearchAllBuilder propertyNamingStrategy(Integer propertyNamingStrategy) {this.propertyNamingStrategy = propertyNamingStrategy;return this;}public String getIndices() {return indices;}public String getTypes() {return types;}public SearchSourceBuilder getSearchSourceBuilder() {return searchSourceBuilder;}public RequestOptions getOptions() {return options;}public Integer getPropertyNamingStrategy() {return propertyNamingStrategy;}} 2、ES 服务常量 /** * ES 服务常量 * * @author Lance * @version 1.0 * @data 2022/03/24 23:12 */public interface Constant {/*** ES原生转换*/Integer ORIGINAL = 1;/*** ES下划线转驼峰*/Integer UNDERSCORE_TO_CAMEL = 2;} 3、Lambda获取字段属性 /** * 获取字段名 * * @author Lance * @version 1.0 * @data 2022/03/24 16:15 */public class LambdaUtils {public static String getfieldName(TypeFunction typeFunction) {return TypeFunction.getLambdaColumnName(typeFunction);}} 4、函数式接口 /** * 函数式接口 -反射 Lambda获取属性值 * * @param * @param * @author Lance * @version 1.0 * @data 2022/03/24 15:11 */@FunctionalInterfacepublic interface TypeFunction extends Serializable, Function {/*** 获取列名称** @param lambda* @return String*/static String getLambdaColumnName(Serializable lambda) {try {Method method = lambda.getClass().getDeclaredMethod("writeReplace");method.setAccessible(Boolean.TRUE);SerializedLambda serializedLambda = (SerializedLambda) method.invoke(lambda);String getter = serializedLambda.getImplMethodName();String fieldName = Introspector.decapitalize(getter.replace("get", ""));return fieldName;} catch (ReflectiveOperationException e) {throw new RuntimeException(e);}}} 5、ES 查询接口 /** * ES 查询接口 * * @author Lance * @version 1.0 * @data 2022/03/24 23:25 */public interface EsSearchApiService {/*** 查询全量（其实也是分页循环查询）** @param searchAllBuilder 查询所有的builder* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/ List searchAll(SearchAllBuilder searchAllBuilder, Class rClass) throws IOException;/*** 指定字段查询** @param indexName 索引名称* @param field字段名称* @param value字段值* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/ T searchField(String indexName, String field, Object value, Class rClass) throws IOException;/*** 同一个字段条件查询* 例如：orderSn["2033-SHEN-AA","2033-SHEN-CC","2033-SHEN-BB"]** @param indexName 索引名称* @param field字段名称* @param value字段值* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/ List searchField(String indexName, String field, Object[] value, Class rClass) throws IOException;/*** 多查询条件查询 - 【精确匹配】** @param indexName 索引名称* @param queryInfo 查询条件* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/ List searchMultipleEq(String indexName, HashMap queryInfo, Class rClass) throws IOException;/*** 多查询条件查询 - 【模糊匹配】** @param indexName 索引名称* @param queryInfo 查询条件* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/ List searchMultipleLike(String indexName, HashMap queryInfo, Class rClass) throws IOException;} 6、ES 查询查询业务类 /** * ES 查询查询 * * @author Lance * @version 1.0 * @data 2022/03/24 23:25 */@Slf4j@Service@AllArgsConstructorpublic class EsSearchApiServiceImpl implements EsSearchApiService {private final RestHighLevelClient client;private static final String COMMA_SEPARATE = ",";/*** 查询全量（其实也是分页循环查询）** @param indices索引名称，多个时逗号分隔* @param types索引类型名称，多个时逗号分隔* @param searchSourceBuilder查询的builder* @param options请求类型* @param tClass返回对象的类* @param propertyNamingStrategy 转换策略（PropertyNamingStrategyConstant.class）* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/public List searchAll(@NonNull String indices, @NonNull String types,@NonNull SearchSourceBuilder searchSourceBuilder, @NonNull RequestOptions options, @NonNull Class tClass,Integer propertyNamingStrategy) throws IOException {// 校验索引String[] indexArray = indices.split(COMMA_SEPARATE);Objects.requireNonNull(indices, "indices must not be null");for (String index : indexArray) {Objects.requireNonNull(index, "index must not be null");}// 校验类型Objects.requireNonNull(types, "types must not be null");String[] typeArray = types.split(COMMA_SEPARATE);for (String type : typeArray) {Objects.requireNonNull(type, "type must not be null");}if (Constant.ORIGINAL.equals(propertyNamingStrategy)) {} else if (Constant.UNDERSCORE_TO_CAMEL.equals(propertyNamingStrategy)) {} else {throw new RuntimeException("propertyNamingStrategy is not found");}return searchAll(indexArray, typeArray, searchSourceBuilder, options, tClass, propertyNamingStrategy);}/*** 查询全量（其实也是分页循环查询）** @param indices索引名称数组* @param types索引类型名称数组* @param searchSourceBuilder查询的builder* @param options请求类型* @param tClass返回对象的类* @param propertyNamingStrategy 转换策略（PropertyNamingStrategyConstant.class）* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/private List searchAll(String[] indices, String[] types, SearchSourceBuilder searchSourceBuilder,RequestOptions options, Class tClass, Integer propertyNamingStrategy) throws IOException {List allList = new ArrayList<>();List list;// 设置分页尺寸int from = -1 == searchSourceBuilder.from() ? 0 : searchSourceBuilder.from(),size = -1 == searchSourceBuilder.size() ? 5000 : searchSourceBuilder.size(), page = 0;do {// 组装查询条件SearchRequest searchRequest = new SearchRequest().indices(indices);// 类型名称不为空时，才组装typeif (types != null && types.length > 0) {searchRequest.types(types);}searchRequest.source(searchSourceBuilder.from(from).size(size));SearchResponse searchResponse = client.search(searchRequest, options);SearchHits searchHits = searchResponse.getHits();SearchHit[] searchHitsHits = searchHits.getHits();long totalHits = searchHits.getTotalHits();// 查询结果list = Arrays.stream(searchHitsHits).map(searchHit -> {if (Constant.ORIGINAL.equals(propertyNamingStrategy)) {return MapObjectUtil.originalMapParseObject(searchHit.getSourceAsMap(), tClass);} else if (Constant.UNDERSCORE_TO_CAMEL.equals(propertyNamingStrategy)) {return MapObjectUtil.underscoreToCamelMapParseObject(searchHit.getSourceAsMap(), tClass);} else {throw new RuntimeException("propertyNamingStrategy is not found");}}).collect(Collectors.toList());// 将查询结果投入allListallList.addAll(list);// 当命中小于分页尺寸时直接跳出if (totalHits < size) {break;}// 翻页page++;from = (page + 1) * size;// 查询直到分页的数据为null} while (!CollectionUtils.isEmpty(list));return allList;}/*** 查询全量（其实也是分页循环查询）** @param indices索引名称，多个时逗号分隔* @param searchSourceBuilder查询的builder* @param options请求类型* @param tClass返回对象的类* @param propertyNamingStrategy 转换策略（PropertyNamingStrategyConstant.class）* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/public List searchAll(@NonNull String indices, @NonNull SearchSourceBuilder searchSourceBuilder,@NonNull RequestOptions options, @NonNull Class tClass, Integer propertyNamingStrategy) throws IOException {String[] indexArray = indices.split(COMMA_SEPARATE);Objects.requireNonNull(indices, "indices must not be null");for (String index : indexArray) {Objects.requireNonNull(index, "index must not be null");}if (Constant.ORIGINAL.equals(propertyNamingStrategy)) {} else if (Constant.UNDERSCORE_TO_CAMEL.equals(propertyNamingStrategy)) {} else {throw new RuntimeException("propertyNamingStrategy is not found");}return searchAll(indexArray, null, searchSourceBuilder, options, tClass, propertyNamingStrategy);}/*** 查询全量（其实也是分页循环查询）** @param searchAllBuilder 查询所有的builder* @param rClass返回对象的类* @param 泛型* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/@Overridepublic List searchAll(SearchAllBuilder searchAllBuilder, Class rClass) throws IOException {// 无types的查询if (Objects.isNull(searchAllBuilder.getTypes())) {return this.searchAll(searchAllBuilder.getIndices(), searchAllBuilder.getSearchSourceBuilder(),searchAllBuilder.getOptions(), rClass, searchAllBuilder.getPropertyNamingStrategy());}return this.searchAll(searchAllBuilder.getIndices(), searchAllBuilder.getTypes(),searchAllBuilder.getSearchSourceBuilder(), searchAllBuilder.getOptions(), rClass,searchAllBuilder.getPropertyNamingStrategy());}/*** 同一个字段条件查询* 例如：orderSn["2033-SHEN-AA","2033-SHEN-CC","2033-SHEN-BB"]** @param indexName 索引名称* @param field字段名称* @param value字段值* @param rClass返回对象的类* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/@Overridepublic List searchField(String indexName, String field, Object[] value, Class rClass) throws IOException {SearchSourceBuilder builder = new SearchSourceBuilder().query(QueryBuilders.termsQuery(field, value));SearchAllBuilder searchAllBuilder = new SearchAllBuilder().indices(indexName).searchSourceBuilder(builder).propertyNamingStrategy(Constant.UNDERSCORE_TO_CAMEL);List searchField = searchAll(searchAllBuilder, rClass);return searchField;}/*** 多查询条件查询 - 【精确匹配】** @param indexName 索引名称* @param queryInfo 查询条件* @param rClass返回对象的类* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/@Overridepublic List searchMultipleEq(String indexName, HashMap queryInfo, Class rClass) throws IOException {SearchSourceBuilder builder = new SearchSourceBuilder();BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();// 循环添加查询条件if (CollectionUtil.isNotEmpty(queryInfo)) {for (Map.Entry entry : queryInfo.entrySet()) {String key = entry.getKey();Object value = https://tazarkount.com/read/entry.getValue();queryBuilder.must(QueryBuilders.termQuery(key, value));}}builder.query(queryBuilder);SearchAllBuilder searchAllBuilder = new SearchAllBuilder().indices(indexName).searchSourceBuilder(builder).propertyNamingStrategy(Constant.UNDERSCORE_TO_CAMEL);List searchField = searchAll(searchAllBuilder, rClass);return searchField;}/*** 多查询条件查询 - 【模糊匹配】** @param indexName 索引名称* @param queryInfo 查询条件* @param rClass返回对象的类* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/@Overridepublic List searchMultipleLike(String indexName, HashMap queryInfo, Class rClass) throws IOException {SearchSourceBuilder builder = new SearchSourceBuilder();BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();// 循环添加查询条件if (CollectionUtil.isNotEmpty(queryInfo)) {for (Map.Entry entry : queryInfo.entrySet()) {String key = entry.getKey();String value = https://tazarkount.com/read/String.valueOf(entry.getValue());queryBuilder.must(QueryBuilders.wildcardQuery(key, value));}}builder.query(queryBuilder);SearchAllBuilder searchAllBuilder = new SearchAllBuilder().indices(indexName).searchSourceBuilder(builder).propertyNamingStrategy(Constant.UNDERSCORE_TO_CAMEL);List searchField = searchAll(searchAllBuilder, rClass);return searchField;}/*** 指定字段查询** @param indexName 索引名称* @param field字段名称* @param value字段值* @param rClass返回对象的类* @return 返回查询出来的全量数据* @throws IOException 抛出io异常*/@Overridepublic T searchField(String indexName, String field, Object value, Class rClass) throws IOException {SearchSourceBuilder builder = new SearchSourceBuilder().query(QueryBuilders.termQuery(field, value));SearchAllBuilder searchAllBuilder = new SearchAllBuilder().indices(indexName).searchSourceBuilder(builder).propertyNamingStrategy(Constant.UNDERSCORE_TO_CAMEL);List searchField = searchAll(searchAllBuilder, rClass);return searchField.get(Constant.ZERO);}} 7、map对象转换工具类 /** * map对象转换工具类 * * @author Lance * @version 1.0 * @data 2022/03/24 23:25 */@Componentpublic class MapObjectUtil {/*** 将下划线key驼峰化的map转为原生object对象** @param map* @param beanClass* @param * @return*/public static T underscoreToCamelMapParseObject(Map map, Class beanClass) {if (map == null) {return null;}try {// 下划线转驼峰SerializeConfig config = new SerializeConfig();config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;return JSON.parseObject(JSON.toJSONString(map, config), beanClass);} catch (Exception e) {throw new RuntimeException(e);}}/*** 将es原始key的map转为原生object对象** @param map* @param beanClass* @param * @return*/public static T originalMapParseObject(Map map, Class beanClass) {if (map == null) {return null;}try {return JSON.parseObject(JSON.toJSONString(map), beanClass);} catch (Exception e) {throw new RuntimeException(e);}}/*** 将对象转为map** @param obj* @return*/public static Map objectParseMap(Object obj) {if (obj == null) {return null;}return ((JSONObject) JSON.toJSON(obj)).getInnerMap();}} 8、测试调用 @GetMapping("mall/tracking/number")@ApiOperation("多条件查询")@Overridepublic List findByMallIdAndTrackingNumber(Long mallId, String trackingNumber) {// Vo 里面的字段一定要和 ES 中的字段一一对应// 可以将 ES 中的json 在线生成javaBeanList orderList = new ArrayList<>();// 精确匹配查询条件HashMap queryInfo = new HashMap<>();queryInfo.put(LambdaUtils.getfieldName(PddElasticOrderVo::getMallId), mallId);queryInfo.put(LambdaUtils.getfieldName(PddElasticOrderVo::getTrackingNumber), trackingNumber);try {List searchField = searchApiService.searchMultipleEq(EsConfig.ES_ORDER_INDEX, queryInfo, JSONObject.class);searchField.stream().forEach(item -> {PddElasticOrderVo orderVo = item.toBean(PddElasticOrderVo.class);orderList.add(orderVo);});log.info("ES中搜索到 {} 条数据!", orderList.size());return orderList;} catch (IOException e) {log.error(e.getMessage());}return Collections.emptyList();}