ElasticSearch 海量数据查询

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;
    }
}

public interface Constant {
    
    Integer ORIGINAL = 1;

    
    Integer UNDERSCORE_TO_CAMEL = 2;
}

public class LambdaUtils {

    public static  String getfieldName(TypeFunction typeFunction) {
        return TypeFunction.getLambdaColumnName(typeFunction);
    }

}

@FunctionalInterface
public interface TypeFunction extends Serializable, Function {

    
    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);
        }
    }

}

public interface EsSearchApiService {

    
     List searchAll(SearchAllBuilder searchAllBuilder, Class rClass) throws IOException;
   
     T searchField(String indexName, String field, Object value, Class rClass) throws IOException;

    
     List searchField(String indexName, String field, Object[] value, Class rClass) throws IOException;

    
     List searchMultipleEq(String indexName, HashMap queryInfo, Class rClass) throws IOException;

    
     List searchMultipleLike(String indexName, HashMap queryInfo, Class rClass) throws IOException;


}

@Slf4j
@Service
@AllArgsConstructor
public class EsSearchApiServiceImpl implements EsSearchApiService {

    private final RestHighLevelClient client;

    private static final String COMMA_SEPARATE = ",";

    
    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);
    }

    
    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);
            // 类型名称不为空时，才组装type
            if (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());
            // 将查询结果投入allList
            allList.addAll(list);

            // 当命中小于分页尺寸时直接跳出
            if (totalHits < size) {
                break;
            }
            // 翻页
            page++;
            from = (page + 1) * size;
            // 查询直到分页的数据为null
        } while (!CollectionUtils.isEmpty(list));

        return allList;
    }

    
    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);
    }

    
    @Override
    public  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());
    }
   
    @Override
    public  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;
    }

    
    @Override
    public  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 = 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;
    }

    
    @Override
    public  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 = 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;

    }

    
    @Override
    public  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);
    }
}

@Component
public class MapObjectUtil {
    
    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);
        }

    }

    
    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);
        }

    }

    
    public static Map objectParseMap(Object obj) {
        if (obj == null) {
            return null;
        }

        return ((JSONObject) JSON.toJSON(obj)).getInnerMap();
    }
}

    @GetMapping("mall/tracking/number")
    @ApiOperation("多条件查询")
    @Override
    public List findByMallIdAndTrackingNumber(Long mallId, String trackingNumber) {
        // Vo 里面的字段一定要和 ES 中的字段一一对应
        // 可以将 ES 中的json 在线生成javaBean
        List 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();
    }