ES操作通用类_es updateresponse

EsTemplate

@Slf4j
@Component
public class EsTemplate {
 
    @Autowired
    private RestHighLevelClient client;
 
    private static final String TIMESTAMP = "timestamp";
 
    /**
     * 创建索引(默认分片数为5和副本数为1)
     *
     * @param indexName
     * @param esMapTypeClass
     * @throws IOException
     */
    public boolean createIndex(String indexName, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        if (checkIndexExists(indexName)) {
            log.error("\"index={}\"索引已经存在！", indexName);
            return false;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        XContentBuilder builder = buildXContentBuilder(indexName, esMapTypeClass);
        request.mapping(esMapTypeClass.getSimpleName(), builder);
 
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        // 指示是否所有节点都已确认请求
        boolean acknowledged = response.isAcknowledged();
        // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            log.info("创建索引成功！索引名称为{}", indexName);
            return true;
        }
 
        log.error("创建索引失败");
        return false;
 
    }
 
    /**
     * 创建索引
     *
     * @param index
     * @throws IOException
     */
    public boolean createIndex(String index) throws IOException {
        if (checkIndexExists(index)) {
            log.error("\"index={}\"索引已存在！", index);
            return false;
        }
        CreateIndexRequest request = new CreateIndexRequest(index);
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        boolean acknowledged = response.isAcknowledged();
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            log.info("创建索引成功！索引名称为{}", index);
            return true;
        } else {
            log.error("创建索引失败");
            return false;
        }
    }
 
    /**
     * 创建索引(传入参数：分片数、副本数)
     *
     * @param indexName
     * @param shards
     * @param replicas
     * @throws IOException
     */
    public boolean createIndex(String indexName, int shards, int replicas) throws IOException {
        if (checkIndexExists(indexName)) {
            log.error("\"index={}\"索引已存在！", indexName);
            return false;
        }
        Settings.Builder builder = Settings.builder().put("index.number_of_shards", shards).put("index.number_of_replicas", replicas);
        CreateIndexRequest request = new CreateIndexRequest(indexName).settings(builder);
//        request.mapping(TYPE, generateBuilder());
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            log.info("创建索引成功！索引名称为{}", indexName);
            return true;
        } else {
            log.error("创建索引失败");
            return false;
        }
    }
 
    /**
     * 删除索引
     *
     * @param indexName
     * @throws IOException
     */
    public Boolean deleteIndex(String indexName) throws IOException {
        try {
            AcknowledgedResponse response = client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
            if (response.isAcknowledged()) {
                log.info("{} 索引删除成功！", indexName);
                return true;
            }
        } catch (ElasticsearchException ex) {
            if (ex.status() == RestStatus.NOT_FOUND) {
                log.error("{} 索引名不存在", indexName);
                return false;
            }
            log.error("删除失败！");
        }
        return false;
    }
 
 
    /**
     * 判断索引是否存在
     *
     * @param indexName
     * @return
     * @throws IOException
     */
    public boolean checkIndexExists(String indexName) {
        GetIndexRequest request = new GetIndexRequest().indices(indexName);
        try {
            return client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("判断索引是否存在，操作异常！");
        }
        return false;
    }
 
 
    /**
     * 开启索引
     *
     * @param indexName
     * @throws IOException
     */
    public void openIndex(String indexName) throws IOException {
        if (!checkIndexExists(indexName)) {
            log.error("索引不存在！");
            return;
        }
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        OpenIndexResponse response = client.indices().open(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            log.info("{} 索引开启成功！", indexName);
        }
    }
 
    /**
     * 关闭索引
     *
     * @param indexName
     * @throws IOException
     */
    public void closeIndex(String indexName) throws IOException {
        if (!checkIndexExists(indexName)) {
            log.error("索引不存在！");
            return;
        }
        CloseIndexRequest request = new CloseIndexRequest(indexName);
        AcknowledgedResponse response = client.indices().close(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged()) {
            log.info("{} 索引已关闭！", indexName);
        }
    }
 
 
    /**
     * 查找文档
     *
     * @param index
     * @param type
     * @param id
     * @return
     * @throws IOException
     */
    public Map<String, Object> getDocument(String index, String type, String id) throws IOException {
        Map<String, Object> resultMap = new HashMap<>();
        GetRequest request = new GetRequest(index, type, id);
        // 实时(否)
        request.realtime(false);
        // 检索之前执行刷新(是)
        request.refresh(true);
 
        GetResponse response = null;
        try {
            response = client.get(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("文档未找到，请检查参数！");
            }
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("查找失败！");
        }
 
        if (Objects.nonNull(response)) {
            if (response.isExists()) { // 文档存在
                resultMap = response.getSourceAsMap();
            } else {
                // 处理未找到文档的方案。 请注意，虽然返回的响应具有404状态代码，但仍返回有效的GetResponse而不是抛出异常。
                // 此时此类响应不持有任何源文档，并且其isExists方法返回false。
                log.error("文档不存在！");
            }
        }
        return resultMap;
    }
 
    /**
     * 判断文档是否存在
     * @param
     * @return true:存在  false:不存在
     * @author hucm
     * @since 2021/5/31
     */
    public boolean existDoc(String index, String type, String id) throws IOException {
        GetRequest request = new GetRequest(index, type,id);
        // 不需要获取source内容
        request.fetchSourceContext(new FetchSourceContext(false));
        return client.exists(request, RequestOptions.DEFAULT);
    }
 
    /**
     * 删除文档
     *
     * @param indexName
     * @param typeName
     * @param docId
     * @throws IOException
     */
    public void deleteDocument(String indexName, String typeName, String docId) throws IOException {
        DeleteRequest request = new DeleteRequest(indexName, typeName, docId);
        DeleteResponse response = null;
        try {
            response = client.delete(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("删除失败!");
        }
        if (Objects.nonNull(response)) {
            if (response.getResult() == DocWriteResponse.Result.NOT_FOUND) {
                log.error("不存在该文档，请检查参数！");
            }
            log.info("文档已删除！");
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("失败原因：{}", reason);
                }
            }
        }
    }
 
    /**
     * 通过一个脚本语句(如："ctx._source.posttime=\"2018-09-18\"")更新文档
     * 针对单个字段更新或object类型字段更新
     * @param index
     * @param type
     * @param id
     * @param script
     */
    public void updateDocByScript(String index, String type, String id, String script) throws IOException {
        Script inline = new Script(script);
        UpdateRequest request = new UpdateRequest(index, type, id).script(inline);
        try {
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                //log.info("文档更新成功！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                log.error("\"index={},type={},id={}\"的文档已被删除，无法更新！", response.getIndex(), response.getType(), response.getId());
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                log.error("操作没有被执行！");
            }
 
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("未处理原因：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("不存在这个文档，请检查参数！");
            } else if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突异常！");
            }
            log.error("更新失败！：{}，脚本：{}，id：{}",e.getMessage(),script,id);
        }
    }
 
    /**
     * 通过一个脚本语句(针对于object和nested数据类型字段更新)更新文档
     * @param index
     * @param type
     * @param id
     * @param script
     * @param params
     * @throws IOException
     */
    public void updateDocByScript(String index, String type, String id, String script,Map<String, Object> params) throws IOException {
        Script inline = new Script(ScriptType.INLINE,Script.DEFAULT_SCRIPT_LANG,script,params);
        UpdateRequest request = new UpdateRequest(index, type, id).script(inline);
        try {
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                //log.info("文档更新成功！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                log.error("\"index={},type={},id={}\"的文档已被删除，无法更新！", response.getIndex(), response.getType(), response.getId());
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                log.error("操作没有被执行！");
            }
 
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("未处理原因：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("不存在这个文档，请检查参数！");
            } else if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突异常！");
            }
            log.error("更新失败！:{}，脚本：{}，id：{}",e.getMessage(),script,id);
        }
    }
 
    /**
     * 批量增加文档
     *
     * @param indexName
     * @param typeName
     * @param esMappingTypes 添加的文档列表
     * @throws IOException
     */
    public List<String> bulkAdd(String indexName, String typeName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (CollectionUtils.isNotEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                IndexRequest request = new IndexRequest(indexName, typeName, mappingType.generateDocId()).source(JSON.toJSONString(mappingType), XContentType.JSON);
                bulkRequest.add(request);
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        // 刷新策略
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
 
        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量增加操作失败！");
            return new ArrayList<>();
        }
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        // 全部操作成功
        if (!bulkResponse.hasFailures()) {
//            log.info("批量增加操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档增加失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("增加失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("新增失败的货品id：{}",failureList);
                    return failureList;
                } else {
//                    log.info("\"index={}, type={}, id={}\"的文档增加成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }
            }
        }
        return new ArrayList<>();
    }
 
    /**
     * 根据条件删除es数据
     *
     * @param indexName           索引名称
     * @param bigBoolQueryBuilder
     * @return 删除条数
     * @author shixiaorui
     * @date 2020/8/31  18:27
     */
    public Long delByEsQuery(String indexName, BoolQueryBuilder bigBoolQueryBuilder) {
        DeleteByQueryRequest request =
                new DeleteByQueryRequest(indexName);
        request.setQuery(bigBoolQueryBuilder);
        try {
            BulkByScrollResponse bulkResponse =
                    client.deleteByQuery(request, RequestOptions.DEFAULT);
            return bulkResponse.getDeleted();
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("删除失败!:{}", e.getMessage());
        } catch (IOException ioExp) {
            log.error("删除失败:{}", ioExp.getMessage());
        }
        return 0L;
 
    }
 
    /**
     * 批量更新文档
     *
     * @param indexName
     * @param typeName
     * @param esMappingTypes
     * @throws IOException
     */
    public List<String> bulkUpdate(String indexName, String typeName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (CollectionUtils.isNotEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                UpdateRequest request=new UpdateRequest(indexName, typeName, mappingType.generateDocId())
                        .doc(JSON.toJSONString(mappingType), XContentType.JSON);
                // true，表明如果文档不存在，则新更新的文档内容作为新的内容插入文档，
                 //这个和scriptedUpsert的区别是：更新文档的两种不同方式，有的使用doc方法更新有的使用脚本更新
                request.docAsUpsert(true);
 
                // 是否将文档内容作为结果返回，默认是禁止的
                request.fetchSource(false);
                bulkRequest.add(request);
            }
        }
 
 
 
        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量更新操作失败！");
            return new ArrayList<>();
        }
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (!bulkResponse.hasFailures()) {
//            log.info("批量更新操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档更新失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("更新失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("更新失败的货品id: {}", failureList);
                    return failureList;
                }/* else {
                    log.info("\"index={}, type={}, id={}\"的文档更新成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }*/
            }
        }
        return failureList;
    }
 
    /**
     * 批量删除文档
     *
     * @param indexName
     * @param typeName
     * @param docIds
     * @throws IOException
     */
    public boolean bulkDelete(String indexName, String typeName, List<String> docIds) throws IOException {
 
        BulkRequest bulkRequest = new BulkRequest();
        if (CollectionUtils.isNotEmpty(docIds)) {
            for (String docId : docIds) {
                DeleteRequest request = new DeleteRequest(indexName, typeName, docId);
                bulkRequest.add(request);
            }
        }
        if (bulkRequest.numberOfActions() == 0) {
            log.error("操作失败，请检查参数！");
            return false;
        }
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (!bulkResponse.hasFailures()) {
            log.info("批量删除操作成功！");
            return true;
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档删除失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("删除失败详情: {}", failure.getMessage());
                    return false;
                }
            }
            return false;
        }
    }
 
    /**
     * 批量查找文档
     *
     * @param indexName
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> multiGet(String indexName, String typeName, List<String> docIds) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        MultiGetRequest request = new MultiGetRequest();
        if (CollectionUtils.isNotEmpty(docIds)) {
            for (String docId : docIds) {
                request.add(new MultiGetRequest.Item(indexName, typeName, docId));
            }
        }
        request.realtime(false);
        request.refresh(true);
        MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
        List<Map<String, Object>> list = parseMGetResponse(response);
        if (!list.isEmpty()) {
            resultList.addAll(list);
        }
        return resultList;
    }
 
    private List<Map<String, Object>> parseMGetResponse(MultiGetResponse response) {
        List<Map<String, Object>> list = new ArrayList<>();
        MultiGetItemResponse[] responses = response.getResponses();
        for (MultiGetItemResponse item : responses) {
            GetResponse getResponse = item.getResponse();
            if (Objects.nonNull(getResponse)) {
                if (!getResponse.isExists()) {
                    log.error("\"index={}, type={}, id={}\"的文档查找失败，请检查参数！", getResponse.getIndex(), getResponse.getType(), getResponse.getId());
                } else {
                    list.add(getResponse.getSourceAsMap());
                }
            } else {
                MultiGetResponse.Failure failure = item.getFailure();
                ElasticsearchException e = (ElasticsearchException) failure.getFailure();
                if (e.status() == RestStatus.NOT_FOUND) {
                    log.error("\"index={}, type={}, id={}\"的文档不存在！", failure.getIndex(), failure.getType(), failure.getId());
                } else if (e.status() == RestStatus.CONFLICT) {
                    log.error("\"index={}, type={}, id={}\"的文档版本冲突！", failure.getIndex(), failure.getType(), failure.getId());
                }
            }
        }
        return list;
    }
 
    /**
     * 根据条件搜索日志内容(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @param size       返回记录数，可以为空，默认最大返回10条。该值必须小于10000，如果超过10000请使用  {@link #queryAllByConditions}
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> queryByConditions(String indexName, String typeName, String level, String messageKey, Long startTime, Long endTime, Integer size) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        if (StringUtils.isBlank(level) && StringUtils.isBlank(messageKey)) {
            log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return resultList;
        }
 
        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.timeout(TimeValue.timeValueMinutes(2L));
        searchBuilder.query(query);
        searchBuilder.sort(order);
        if (Objects.nonNull(size)) {
            searchBuilder.size(size);
        }
 
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        request.source(searchBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            log.error("部分分片副本处理失败！");
            for (ShardSearchFailure failure : response.getShardFailures()) {
                String reason = failure.reason();
                log.error("分片处理失败原因：{}", reason);
            }
        }
        List<Map<String, Object>> list = parseSearchResponse(response);
        if (!list.isEmpty()) {
            resultList.addAll(list);
        }
        return resultList;
    }
 
    private QueryBuilder generateQuery(String level, String messageKey, Long startTime, Long endTime) {
        // term query(检索level)
        TermQueryBuilder levelQuery = null;
        if (StringUtils.isNotBlank(level)) {
            levelQuery = QueryBuilders.termQuery("level", level.toLowerCase());
        }
        // match query(检索message)
        MatchQueryBuilder messageQuery = null;
        if (StringUtils.isNotBlank(messageKey)) {
            messageQuery = QueryBuilders.matchQuery("name", messageKey);
        }
        // range query(检索timestamp)
        RangeQueryBuilder timeQuery = QueryBuilders.rangeQuery(TIMESTAMP);
        timeQuery.format("epoch_millis");
        if (Objects.isNull(startTime)) {
            if (Objects.isNull(endTime)) {
                timeQuery = null;
            } else {
                timeQuery.lte(endTime);
            }
        } else {
            if (Objects.isNull(endTime)) {
                timeQuery.gte(startTime);
            } else {
                timeQuery.gte(startTime).lte(endTime);
            }
        }
        // 将上述三个query组合
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        if (Objects.nonNull(levelQuery)) {
            boolQuery.must(levelQuery);
        }
        if (Objects.nonNull(messageQuery)) {
            boolQuery.must(messageQuery);
        }
        if (Objects.nonNull(timeQuery)) {
            boolQuery.must(timeQuery);
        }
        return boolQuery;
    }
 
    private List<Map<String, Object>> parseSearchResponse(SearchResponse response) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        SearchHit[] hits = response.getHits().getHits();
        for (SearchHit hit : hits) {
            resultList.add(hit.getSourceAsMap());
        }
        return resultList;
    }
 
    /**
     * 查询所有文档id
     *
     * @return
     */
    public List<String> queryAllIdByIndexName(String indexName, String typeName) throws IOException {
        List<String> resultList = new ArrayList<>();
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.size(500);
 
        // 初始化 scroll 上下文
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        request.source(searchBuilder).scroll(scroll);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        String scrollId = response.getScrollId();
        SearchHit[] searchHits = response.getHits().getHits();
        // 把第一次scroll的数据添加到结果List中
        for (SearchHit searchHit : searchHits) {
            resultList.add(searchHit.getId());
        }
        // 通过传递scrollId循环取出所有相关文档
        while (searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = response.getScrollId();
            searchHits = response.getHits().getHits();
            // 循环添加剩下的数据
            for (SearchHit searchHit : searchHits) {
                resultList.add(searchHit.getId());
            }
        }
        // 清理 scroll 上下文
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        return resultList;
    }
 
    /**
     * 根据条件，搜索全部符合的记录(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @return
     */
    public List<Map<String, Object>> queryAllByConditions(String indexName, String typeName, String level, String messageKey, Long startTime, Long endTime) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        if (StringUtils.isBlank(level) && StringUtils.isBlank(messageKey)) {
            log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return resultList;
        }
 
        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.query(query).sort(order);
        searchBuilder.size(500);
 
        // 初始化 scroll 上下文
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        request.source(searchBuilder).scroll(scroll);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        String scrollId = response.getScrollId();
        SearchHit[] searchHits = response.getHits().getHits();
        // 把第一次scroll的数据添加到结果List中
        for (SearchHit searchHit : searchHits) {
            resultList.add(searchHit.getSourceAsMap());
        }
        // 通过传递scrollId循环取出所有相关文档
        while (searchHits != null && searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = response.getScrollId();
            searchHits = response.getHits().getHits();
            // 循环添加剩下的数据
            for (SearchHit searchHit : searchHits) {
                resultList.add(searchHit.getSourceAsMap());
            }
        }
        // 清理 scroll 上下文
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        return resultList;
    }
 
    /**
     * 根据条件做分页查询(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @param pageNum    当前页码，可以为空(默认设为1)
     * @param pageSize   页记录数，可以为空(默认设为10)
     * @return
     * @throws IOException
     */
    public SearchPageBean<Map<String, Object>> queryPageByConditions(String indexName, String typeName, String level, String messageKey, Long startTime, Long endTime, Integer pageNum, Integer pageSize) throws IOException {
        if (StringUtils.isBlank(level) && StringUtils.isBlank(messageKey)) {
            log.error("参数level(日志级别)、messageKey(日志信息关键字)不能同时为空！");
            return null;
        }
 
        if (Objects.isNull(pageNum)) {
            pageNum = 1;
        }
        if (Objects.isNull(pageSize)) {
            pageSize = 10;
        }
        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.timeout(TimeValue.timeValueMinutes(2L));
        searchBuilder.query(query);
        searchBuilder.sort(order);
        searchBuilder.from(pageNum - 1).size(pageSize);
 
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        request.source(searchBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        int totalRecord = (int) hits.getTotalHits();
        List<Map<String, Object>> results = new ArrayList<>();
        for (SearchHit hit : hits.getHits()) {
            results.add(hit.getSourceAsMap());
        }
 
        SearchPageBean<Map<String, Object>> page = new SearchPageBean<>();
        page.setPageNo(pageNum);
        page.setPageSize(pageSize);
        page.setTotalPages(totalRecord);
        page.setData(results);
        return page;
    }
 
 
    /**
     * 通用查询
     * 复杂查询可以自定义实现RestHighLevelClient
     *
     * @param indexName           索引
     * @param eSMappingType       返回的对象类型
     * @param searchSourceBuilder 查询条件
     * @param <T>                 泛型
     * @return pageResult
     * @author zhangxh
     * @since 2019/07/04
     */
    public <T> PageResult<T> search(String indexName, Class<T> eSMappingType, SearchSourceBuilder searchSourceBuilder) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new ServiceException("indexName 不能为空");
        }
 
        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);
 
        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            PageResult<T> pageResult = new PageResult<>();
            pageResult.setRecordCount((int) hits.getTotalHits());
            List<T> list = new ArrayList<>();
            for (SearchHit hit : hitsArray) {
                list.add(JsonHelper.parseBean(hit.getSourceAsString(), eSMappingType));
            }
            pageResult.setResult(list);
//            log.info("es查询语句：{}" , searchSourceBuilder.toString());
            return pageResult;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return new PageResult<>();
    }
 
 
    public <T> PageResult<T> searchMore(String indexName, Class<T> eSMappingType, SearchSourceBuilder searchSourceBuilder) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new ServiceException("indexName 不能为空");
        }
 
        searchSourceBuilder.from(0).size(10000);
 
        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);
 
        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            PageResult<T> pageResult = new PageResult<>();
            pageResult.setRecordCount((int) hits.getTotalHits());
            List<T> list = new ArrayList<>();
            for (SearchHit hit : hitsArray) {
                list.add(JsonHelper.parseBean(hit.getSourceAsString(), eSMappingType));
            }
            pageResult.setResult(list);
//            log.info("es查询语句：{}" , searchSourceBuilder.toString());
            return pageResult;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return new PageResult<>();
    }
 
 
    private XContentBuilder generateBuilder() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        builder.startObject("properties");
        builder.startObject("message");
        builder.field("type", "text");
        // 为message字段，设置分词器为 ik_smart(最粗粒度)
        builder.field("analyzer", "ik_smart");
        builder.endObject();
        builder.startObject(TIMESTAMP);
        builder.field("type", "date");
        // 设置 日志时间的格式为  毫秒数的long类型
        builder.field("format", "epoch_millis");
        builder.endObject();
        builder.endObject();
        builder.endObject();
        return builder;
    }
 
 
 
 
    /**
     * 根据IESMappingType 的实现类,生成创建mappingType的 XContentBuilder
     *
     * @param esMapTypeClass
     * @return
     * @throws IOException
     */
    public XContentBuilder buildXContentBuilder(String index, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
 
        // 处理对象属性类型
        dealObj(builder, esMapTypeClass);
        builder.endObject();
        return builder;
    }
 
    /**
     * 根据对象的es属性注解判断是否进行属性处理
     *
     * @param builder
     * @param esMapTypeClass
     * @throws IOException
     */
    private void dealObj(XContentBuilder builder, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        // 子属性放到该索引下
        builder.startObject("properties");
        for (Field field : esMapTypeClass.getDeclaredFields()) {
            ESField fieldProp = field.getAnnotation(ESField.class);
            if (fieldProp != null) {
                dealField(builder, field, fieldProp);
            }
        }
 
        //判断对象是否有 @Suggestable 注解,有则添加 suggest 字段
        EnableSuggest suggest = esMapTypeClass.getAnnotation(EnableSuggest.class);
        if (suggest != null) {
            String analyzer = suggest.analyzerType().name();
            String suggestName = suggest.suggestName();
            builder.startObject(suggestName)
                    .field("type", "completion")
                    .field("index_analyzer", analyzer)
                    .field("search_analyzer", analyzer)
                    .field("payloads", "true")
                    .field("preserve_position_increments", false)
                    .field("preserve_separators", false)
                    .endObject();
        }
        builder.endObject();
    }
 
 
 
 
    /**
     * 处理对象的属性类型
     *
     * @param builder
     * @param field     属性
     * @param fieldProp 属性注解信息
     * @throws IOException
     */
    private void dealField(XContentBuilder builder, Field field, ESField fieldProp) throws IOException {
        try {
            if (List.class.isAssignableFrom(field.getType()) || field.getType().isArray()) {
                //list 类型的  嵌套类型和对象数组
                builder.startObject(field.getName());// 这里如果是startArray就会有问题.
                if (fieldProp.fieldType() == ESFieldType.NESTED) {
                    //嵌套类型(要查询对象信息)
                    builder.field("type", ESFieldType.NESTED.getTypeValue());
                }else{
                    //对象数组(只是保存对象信息)
                    builder.field("type", ESFieldType.OBJECT.getTypeValue());
                }
 
                String className = "";
                Type fc = field.getGenericType();
                if (fc instanceof ParameterizedType) {
                    ParameterizedType pt = (ParameterizedType) fc;
                    className = pt.getActualTypeArguments()[0].toString().replace("class ", "");
                } else if (field.getType().isArray()) {
                    className = field.getGenericType().toString().replace("class [L", "")
                            .replace("/", ".").replace(";", "");
                }
 
                Class clazz = Class.forName(className);
 
                if (IESMappingType.class.isAssignableFrom(clazz) || clazz.getAnnotation(ESDocObject.class) != null) {
                    dealObj(builder, clazz);
                } else if (clazz.isPrimitive() || isSimpleType(clazz)) {
                    builder.field("type", ESFieldType.STRING.getTypeValue())
                            .field("index", ESAnalyzer.not_analyzed.name()).field("store", fieldProp.isStore());
                }
                builder.endObject();
            } else if (Map.class.isAssignableFrom(field.getType())) {
                System.out.println("Map:" + field.getName());
            } else {
                // 处理简单对象
                if (isSimpleType(field.getType())) {
                    dealSimpleObjField(builder, field.getName(), fieldProp);
                    return;
                }
 
                // 如果是复杂的组合类型，继承于ESMapTypeI,则进行递归处理
                String className = field.getGenericType().toString().replace("class ", "");
                Class complexClazz = Class.forName(className);
                if (IESMappingType.class.isAssignableFrom(complexClazz)) {
                    builder.startObject(field.getName());
                    if (fieldProp.fieldType() == ESFieldType.NESTED) {
                        builder.field("type", ESFieldType.NESTED.getTypeValue());
                    }
                    dealObj(builder, complexClazz);
                    builder.endObject();
                }
            }
        } catch (Exception e) {
            log.error("创建mapping出错...", e);
        }
    }
 
    /**
     * 判断是否是简单的对象.
     *
     * @param cls
     * @return
     */
    private static boolean isSimpleType(Class cls) {
        if (cls == String.class || cls == Integer.class || cls == BigDecimal.class || cls == Date.class || cls == int.class || cls == long.class || cls == Long.class) {
            return true;
        } else {
            return false;
        }
    }
 
    /**
     * 处理对象类型的域值,处理已经是最简单对象的field
     */
    public void dealSimpleObjField(XContentBuilder mapbuilder, String fieldName, ESField eSMapType) throws IOException {
 
        mapbuilder.startObject(fieldName).field("store", eSMapType.isStore()).field("type", eSMapType.fieldType().getTypeValue());
        ESAnalyzer esAnalyzer = eSMapType.analyzerType();
        if (esAnalyzer != ESAnalyzer.not_analyzed) {
            if (esAnalyzer == ESAnalyzer.analyzed) {
                mapbuilder.field("index", "true");
            } else {
                mapbuilder.field("index", "true").
                        field("search_analyzer", esAnalyzer.name())
                        .field("analyzer", esAnalyzer.name());
            }
 
        } else if (esAnalyzer == ESAnalyzer.not_analyzed) {
            mapbuilder.field("index", "false");
        }
        mapbuilder.endObject();
 
    }
 
    /**
     * es聚合分组求和
     *
     * @param indexName 索引名称
     * @param groupKey  分组的key
     * @param sumKey    求和的key
     * @return Map
     * @author shixiaorui
     * @date 2020/11/12  10:29
     */
    public  Map<String, Long> groupKeySum(String indexName, String eSMappingTypeName, SearchSourceBuilder searchSourceBuilder, String groupKey, String sumKey) {
        try {
            SearchRequest searchRequest = new SearchRequest()
                    .indices(indexName)
                    .types(eSMappingTypeName)
                    .source(searchSourceBuilder);
 
            String aggName = groupKey + "_key";//分组字段别名
            String aggField = sumKey + "_key";//求和字段别名
 
 
            //分组groupKey,求和sumKey
            TermsAggregationBuilder field = AggregationBuilders.terms(aggName).field(groupKey);
            field.subAggregation(AggregationBuilders.sum(aggField).field(sumKey));
            searchSourceBuilder.aggregation(field);
 
            searchRequest.source(searchSourceBuilder);
            //执行查询
            SearchResponse response = client.search(searchRequest);
 
            //获取搜索的文档结果
            Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();
            Terms gradeTerms = (Terms) aggMap.get(aggName);
 
            Map<String, Long> returnMap = new HashMap<>();
            Map<String, Long> valueResult = new LinkedHashMap<>();
            if(null!=gradeTerms.getBuckets()&&gradeTerms.getBuckets().size()>0){
                for (Terms.Bucket bucket : gradeTerms.getBuckets()) {
                    double num = ((Sum) bucket.getAggregations().get(aggField)).getValue();
                    returnMap.put(bucket.getKeyAsString(), new Double(num).longValue());
                }
 
                //排序
                returnMap.entrySet().stream()
                        .sorted(Map.Entry
                                .comparingByValue(Comparator.reverseOrder()))
                        .forEachOrdered(b->valueResult.put(b.getKey(), b.getValue()));
                return valueResult;
            }
            return new HashMap<>();
 
        } catch (Exception ex) {
            ex.printStackTrace();
            return null;
        }
    }
 
 
 
    /**
     * 查询，包含下钻聚合数据
     *
     * @param indexName           索引
     * @param eSMappingType       返回的对象类型
     * @param searchSourceBuilder 查询条件
     * @return pageResult
     * @author hucm
     * @since 2021/08/17
     */
    public IndexSearchResponse searchGroupBy(String indexName, Class eSMappingType, SearchSourceBuilder searchSourceBuilder, MallEsSearchFormDTO form) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new ServiceException("indexName 不能为空");
        }
 
        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);
 
        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            // 封装返回结果
            IndexSearchResponse response = new IndexSearchResponse();
            // 放入源数据
            response.setOriginHits(hits);
 
            List<String> sources = new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
                sources.add(hit.getSourceAsString());
            }
            response.putReslutData(sources);
            // 聚合结果
            response.putResultAggs(agg2Map(search.getAggregations()));
            // 总数据
            response.setPageNum(form.getPageNo());
            response.setPageSize(form.getPageSize());
            response.setTotalCount((int) hits.getTotalHits());
            return response;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return null;
    }
 
    /**
     * 处理一个聚合对象数据{@link Aggregations}
     *
     * @param aggregations
     * @return Map{key:一个聚合的名称,value:聚合后的桶数据}
     */
    public Map<String, Object> agg2Map(Aggregations aggregations) {
        Map<String, Object> resultMap = new HashMap<>();
        if (aggregations == null) {
            return resultMap;
        }
        for (Aggregation aggregation : aggregations) {
 
            resultMap.put(aggregation.getName(), dealOneAggregation(aggregation));
        }
        return resultMap;
    }
 
    /**
     * 处理一个聚合对象
     *
     * @param aggregation {@link Aggregation}
     * @return 如果只有一个桶数据, 则返回对象, 如果有多个桶数据, 则返回数组
     */
    public Object dealOneAggregation(Aggregation aggregation) {
        if (aggregation instanceof ParsedStringTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedStringTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedDoubleTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedDoubleTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedLongTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedLongTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof UnmappedTerms) {
            Collection<? extends Terms.Bucket> buckets = ((UnmappedTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedNested) {
            Aggregations aggregations = ((ParsedNested) aggregation).getAggregations();
            return agg2list(aggregations);
        } else if (aggregation instanceof ParsedAvg) {
            return aggregation.getName();
        } else {
            throw new IllegalArgumentException("未知聚合类型,不可处理");
        }
    }
 
    /**
     * 处理一个聚合下边的一个或多个桶数据
     *
     * @param buckets
     * @return 如果有一个桶, 那么就返回一个对象, 而不是列表 如果有多个桶,则返回列表 如果没有值,则返回一个空对象
     */
    public Object dealBunkets(Collection<? extends Terms.Bucket> buckets) {
        List<Object> list = new ArrayList<>();
        for (Terms.Bucket bucket : buckets) {
            list.add(dealOneBunket(bucket));
        }
        return list.isEmpty() ? new Object() : list.size() == 1 ? list.get(0) : list;
    }
 
    /**
     * 处理一个bunkey
     *
     * @param bucket
     * @return 如果没有子查询返回bunkey中的值[String]
     * 如果有子查询,返回一个对应的map对象[HashMap]--{key:bunketKey,value:子查询返回的map}
     */
    public Object dealOneBunket(Terms.Bucket bucket) {
        Object params = null;
        if (bucket.getAggregations().iterator().hasNext()) {
            params = agg2list(bucket.getAggregations());
        }
        if (params == null) {
            return bucket.getKey();// 没有子查询
        } else if (params instanceof List) {
            List<Object> resultList = (List) params;
            if (resultList.size() == 1) {
                return resultList.get(0);
            } else if (resultList.size() > 1) {
                Map<String, Object> resultMap = new HashMap<>();
                resultMap.put(bucket.getKey().toString(), resultList);
                return resultMap;
            } else {
                return resultList;
            }
        } else {
            // 没有子查询
            Map<String, Object> resultMap = new HashMap<>();
            resultMap.put(bucket.getKey().toString(), params);
            return resultMap;
        }
    }
 
    /**
     * 处理子聚合的方法
     *
     * @param aggregations
     * @return
     * @author of1081
     */
    public Object agg2list(Aggregations aggregations) {
        List<Object> resultList = new ArrayList<>();
        for (Aggregation aggregation : aggregations) {
            if (!(aggregation instanceof InternalAvg)) {
                resultList.add(dealOneAggregation(aggregation));
            }
        }
        return resultList.isEmpty() ? new Object() : resultList.size() == 1 ? resultList.get(0) : resultList;
    }
 
}