ESUtil工具类

import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.plugins.Page;
import com.kingstar.basebusiness.exception.ExceptionUtil;
import com.kingstar.mnReport.pojo.MnIndicatorPojo;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;
//ESUtil
@Component
public class ESUtil{
    private static Log logger = LogFactory.getLog(EsUtil.class);
    public static final String INDEX_PREFIX = "index_pre";

    private static String ELASTICSEARCH_IP;
    private static String ELASTICSEARCH_PORT;
    private static RestHighLevelClient client = null;

    @Value("${spring.elasticsearch.ip}")
    public void setElasticsearchIp(String elasticsearchIp) {
        ELASTICSEARCH_IP = elasticsearchIp;
    }

    @Value("${spring.elasticsearch.port}")
    public void setElasticsearchPort(String elasticsearchPort) {
        ELASTICSEARCH_PORT = elasticsearchPort;
    }
    
    /***
     * 前缀模糊搜索索引名称
     */
    public static String[] getIndexByPrefix(String indexPrefix) {
        EsUtil.getClient();
        IndicesClient indicesClient = client.indices();
        String[] indexs = null;
        try {
            GetIndexRequest getIndexRequest = new GetIndexRequest();
            getIndexRequest.indices(indexPrefix + "*");//模糊搜索ES索引名称
            GetIndexResponse getIndexResponse = indicesClient.get(getIndexRequest, RequestOptions.DEFAULT);
            indexs = getIndexResponse.getIndices();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return indexs;
    }
    //获取连接
    public static RestHighLevelClient getClient() {
        try {
            if (null == EsUtil.client) {
                EsUtil.client = new RestHighLevelClient(RestClient.builder(new HttpHost(ELASTICSEARCH_IP, Integer.valueOf(ELASTICSEARCH_PORT), "http")));
            }
            return client;
        } catch (Exception e) {
            logger.error("EsUtil getClient has some error [{}]", e);
            throw new RuntimeException(e);
        }
    }
    //关闭连接
   public static void closeClient() {
        try {
            if (null != EsUtil.client) {
                EsUtil.client.close();
                EsUtil.client = null;
            }
        } catch (Exception e) {
            logger.error("EsUtil closeClient has some error ", e);
        }
    }
    
     //判断索引是否存在
    public static boolean isIndexExist(String index) {
        try {
            GetIndexRequest request = new GetIndexRequest();
            request.indices(index);
            request.local(false);
            request.humanReadable(true);
            IndicesClient indicesClient = client.indices();
            return indicesClient.exists(request,RequestOptions.DEFAULT);
        } catch (Exception e) {
            logger.error("*******isIndexExist Exception:index="+index,e);
            return false;
        }
    }
    
    /**
     * 根据条件分页查询
     *
     * @param code      机器或应用编码 必填
     * @param indicator 指标          可选
     * @param startTime 开始时间        可选
     * @param endTime   结束时间    可选
     * @param page      分页信息    必填
     */
    public static void query(String code, List<String> indicator, Date startTime, Date endTime, Page<MnIndicatorPojo> page) throws IOException {
        EsUtil.getClient();
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(getIndexName(startTime, endTime));
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder must = QueryBuilders.boolQuery();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.termQuery("code", code));
        if (startTime != null && endTime != null) {
            builder.must(QueryBuilders.rangeQuery("collectTime").gte(startTime).lte(endTime));
        }
        if (indicator != null && !indicator.isEmpty()) {
            builder.must(QueryBuilders.termsQuery("item", indicator));
        }
        must.must(builder);
        searchSourceBuilder.query(must);
        searchSourceBuilder.from((page.getCurrent() - 1) * page.getSize());
        searchSourceBuilder.size(page.getSize());
        searchSourceBuilder.sort("collectTime", SortOrder.DESC);
        searchRequest.source(searchSourceBuilder);
        SearchResponse res = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = res.getHits();
        long totalHits = hits.getTotalHits();
        page.setTotal((int) totalHits);
        SearchHit[] hits1 = hits.getHits();
        ArrayList<MnIndicatorPojo> mnIndicatorPojos = new ArrayList<>(hits1.length);
        for (SearchHit searchHit : hits1) {
            String sourceAsString = searchHit.getSourceAsString();
            MnIndicatorPojo mnIndicatorPojo = JSON.parseObject(sourceAsString, MnIndicatorPojo.class);
            mnIndicatorPojos.add(mnIndicatorPojo);
        }
        page.setRecords(mnIndicatorPojos);
    }
/***
     * 根据时间段，删除索引数据,不删除索引
     */
    public static BulkByScrollResponse deleteIndexByTime(Set<String> indexNames, Long startTime, Long endTime) throws IOException {
        EsUtil.getClient();
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
        deleteByQueryRequest.indices(indexNames.toArray(new String[indexNames.size()]));
        BoolQueryBuilder mustBuilder = QueryBuilders.boolQuery();
        if (!CommonUtil.isEmpty(startTime) && !CommonUtil.isEmpty(endTime)) {
            mustBuilder.must(QueryBuilders.rangeQuery("@timestamp").from(startTime, true).to(endTime, false));
        }
        BulkByScrollResponse bulkByScrollResponse = null;
        deleteByQueryRequest.setQuery(mustBuilder);
        bulkByScrollResponse = EsUtil.client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        return bulkByScrollResponse;
    }

    /***
     * 删除索引
     */
    public static AcknowledgedResponse deleteIndex(Set<String> indexNames) throws IOException {
        EsUtil.getClient();
        DeleteIndexRequest deleteRequest = new DeleteIndexRequest();
        deleteRequest.indices(indexNames.toArray(new String[indexNames.size()]));
        AcknowledgedResponse acknowledgedResponse = EsUtil.client.indices().delete(deleteRequest, RequestOptions.DEFAULT);
        logger.info("delete index response:" + JSON.toJSONString(acknowledgedResponse));

        return acknowledgedResponse;
    }
 /***
     * 根据时间段和应用code，删除索引数据
     */
    public static BulkByScrollResponse deleteIndicatorIndex(List<String> codes, String indexName, Date startTime, Date endTime) throws IOException {
        EsUtil.getClient();
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
        deleteByQueryRequest.indices(indexName);
//        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder mustBuilder = QueryBuilders.boolQuery();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        if (codes != null) {
            for (String code : codes) {
                builder.should(QueryBuilders.termQuery("equipNo", code));
            }
        }
        mustBuilder.must(builder);

        if (startTime != null) {
            mustBuilder.must(QueryBuilders.rangeQuery("collectTime").gte(startTime));
        }
        if (endTime != null) {
            mustBuilder.must(QueryBuilders.rangeQuery("collectTime").lte(endTime));
        }

        deleteByQueryRequest.setQuery(mustBuilder);
        BulkByScrollResponse bulkByScrollResponse = EsUtil.client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        return bulkByScrollResponse;
    }
 /**
     * 获取索引名称
     *
     * @param startTime
     * @param endTime
     * @return
     */
    private static String[] getIndexName(Date startTime, Date endTime) throws IOException {
        if (startTime == null || endTime == null) {
            String[] indices = new String[1];
            indices[0] = INDEX_PREFIX + "*";
            return indices;
        }
        ExceptionUtil.isTrue(endTime.getTime() - startTime.getTime() >= 0, "时间错误");
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(startTime);
        calendar.set(Calendar.HOUR_OF_DAY, 0);
        calendar.set(Calendar.MINUTE, 0);
        calendar.set(Calendar.SECOND, 0);
        calendar.set(Calendar.MILLISECOND, 0);
        ArrayList<String> strings = new ArrayList<>();
        Date date = calendar.getTime();
        while (true) {
            String dateStr = DateUtil.dateToString(date, DateUtil.YYYYMMDD);
            String index = INDEX_PREFIX + dateStr;
            ExceptionUtil.isTrue(isIndexExist(index), DateUtil.dateToString(date, DateUtil.YYYY_MM_DD) + "的数据不存在");
            strings.add(index);
            calendar.add(Calendar.DAY_OF_MONTH, 1);
            date = calendar.getTime();
            if (date.getTime() > endTime.getTime()) {
                break;
            }
        }
        String[] result = new String[strings.size()];
        return strings.toArray(result);
    }
}