elasticsearch的基本api

条件查询、排序、去重

SearchSourceBuilder builder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        if (ObjectUtil.isNotEmpty(apiDispatchRecordDto.getApiName())) {
            boolQueryBuilder.must(QueryBuilders.wildcardQuery("apiName.keyword",  "*" + apiDispatchRecordDto.getApiName()+ "*"));
        }

        if (ObjectUtil.isNotEmpty(apiDispatchRecordDto.getAppApiKey())) {
            boolQueryBuilder.must(QueryBuilders.wildcardQuery("appApiKey", apiDispatchRecordDto.getAppApiKey()));
        }

        if (ObjectUtil.isNotEmpty(apiDispatchRecordDto.getAppName())) {
            boolQueryBuilder.must(QueryBuilders.wildcardQuery("appName.keyword", "*"+apiDispatchRecordDto.getAppApiKey()+"*"));
        }
        if (ObjectUtil.isNotEmpty(apiDispatchRecordDto.getStart())&&ObjectUtil.isNotEmpty(apiDispatchRecordDto.getEnd())) {
            boolQueryBuilder.must(QueryBuilders.rangeQuery("dispatchTime").gte(apiDispatchRecordDto.getStart()).lte(apiDispatchRecordDto.getEnd()));
        }
        builder.query(boolQueryBuilder);
        //去重
        builder.collapse(new CollapseBuilder("apiName.keyword"));
        //排序
        builder.sort("dispatchTime", SortOrder.DESC);
        Map map = elasticsearchUtil.searchListData(KafkaConstant.OPEN_API_DISPATCH_RECORD_TOPIC, builder, pageSize, pageNum-1,
                null, null, null);
        List list = (List) map.get("data");

查询符合条件的数据量

		SearchSourceBuilder builder = new SearchSourceBuilder();
        CountRequest countRequest = new CountRequest();
        //构造查询条件
        builder.query(QueryBuilders.termQuery("apiResponseCode",code));
        countRequest.indices(KafkaConstant.OPEN_API_DISPATCH_RECORD_TOPIC).query(QueryBuilders.termQuery("apiResponseCode",code));
        return (int)restHighLevelClient.count(countRequest, RequestOptions.DEFAULT).getCount();

添加数据

        ApiDispatchRecordEsBo apiDispatchRecordEsBo = new ApiDispatchRecordEsBo();
        apiDispatchRecordEsBo.setApiId("1506101690120949761");
        apiDispatchRecordEsBo.setApiName("物联网设备注册");
        apiDispatchRecordEsBo.setApiResponseCode(404);
        apiDispatchRecordEsBo.setAppApiKey("d0d8544326304087848934583a271fe3");
        apiDispatchRecordEsBo.setApiResponseTime(444l);
        apiDispatchRecordEsBo.setDispatchTime(new Date(1651373682));
        JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(apiDispatchRecordEsBo));
        System.out.println(elasticsearchUtil.addData(jsonObject, "open_api_dispatch_record"));

工具类

package com.netintech.es.util;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONObject;
import com.netintech.common.core.utils.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

@Component
public class  ElasticsearchUtil {
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    
    public boolean createIndex(String index) throws IOException {
        if (isIndexExist(index)) {
            return false;
        }
        //1.创建索引请求
        CreateIndexRequest request = new CreateIndexRequest(index);
        //2.执行客户端请求
        CreateIndexResponse response = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        return response.isAcknowledged();
    }

    
    public boolean deleteIndex(String index) throws IOException {
        if (!isIndexExist(index)) {
            return false;
        }
        //删除索引请求
        DeleteIndexRequest request = new DeleteIndexRequest(index);
        //执行客户端请求
        AcknowledgedResponse delete = restHighLevelClient.indices().delete(request, RequestOptions.DEFAULT);


        return delete.isAcknowledged();
    }

    public static void main(String[] args) {

    }


    
    public boolean isIndexExist(String index) throws IOException {

        GetIndexRequest request = new GetIndexRequest(index);

        boolean exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);

        return exists;
    }

    

    public String bulkPutIndex(List> list, String index) {
        BulkRequest request = new BulkRequest();
        for (int j = 0; j < list.size(); j++) {
            Map item = list.get(j);
            request.add(new IndexRequest(index).source(item));
        }
        try {
            BulkResponse bulk = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            if (bulk.status().getStatus() == 200) {
                return "success";
            } else {

                return "error";
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "";
    }

    
    public String addData(JSONObject jsonObject, String index, String id) throws IOException {

        //创建请求
        IndexRequest request = new IndexRequest(index);
        //规则 put /test_index/_doc/1
        request.id(id);
        request.timeout(TimeValue.timeValueSeconds(1));
        //将数据放入请求 json
        IndexRequest source = request.source(jsonObject, XContentType.JSON);
        //客户端发送请求
        IndexResponse response = restHighLevelClient.index(request, RequestOptions.DEFAULT);
        return response.getId();
    }


    
    public String addData(JSONObject jsonObject, String index) throws IOException {

        return addData(jsonObject, index, UUID.randomUUID().toString().replaceAll("-", "").toUpperCase());
    }

    
    public void deleteDataById(String index, String id) throws IOException {
        //删除请求
        DeleteRequest request = new DeleteRequest(index, id);
        //执行客户端请求
        DeleteResponse delete = restHighLevelClient.delete(request, RequestOptions.DEFAULT);
    }

    
    public void deleteByQuery(String index, QueryBuilder queryBuilder) throws IOException {
        DeleteByQueryRequest request = new DeleteByQueryRequest(index);
        request.setQuery(queryBuilder);
        //request.setBatchSize(1000);
        //执行客户端请求
        restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
    }


    
    public void deleteByData(String index) throws IOException {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
        AcknowledgedResponse deleteResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
    }

    
    public void updateDataById(JSONObject object, String index, String id) throws IOException {
        //更新请求
        UpdateRequest update = new UpdateRequest(index, id);

        //保证数据实时更新
        //update.setRefreshPolicy("wait_for");

        update.timeout("1s");
        update.doc(object, XContentType.JSON);
        //执行更新请求
        UpdateResponse update1 = restHighLevelClient.update(update, RequestOptions.DEFAULT);
    }


    
    public int selectAllCount(String index) throws IOException {
        int result = 0;
        if (isIndexExist(index) == false) {
            createIndex(index);
        }

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        CountRequest countRequest = new CountRequest();
        //构造查询条件
        // searchSourceBuilder.query(QueryBuilders.termQuery("fieldName", 1));
        countRequest.indices(index).source(searchSourceBuilder);
        CountResponse countResponse = null;
        try {
            countResponse = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
            result = (int) countResponse.getCount();

        } catch (IOException e) {
        }
        return result;
    }


    
    public void updateDataByIdNoRealTime(JSONObject object, String index, String id) throws IOException {
        //更新请求
        UpdateRequest update = new UpdateRequest(index, id);

        //保证数据实时更新
        update.setRefreshPolicy("wait_for");

        update.timeout("1s");
        update.doc(object, XContentType.JSON);
        //执行更新请求
        UpdateResponse update1 = restHighLevelClient.update(update, RequestOptions.DEFAULT);
    }


    
    public Map searchDataById(String index, String id, String fields) throws IOException {
        GetRequest request = new GetRequest(index, id);
        if (StringUtils.isNotEmpty(fields)) {
            //只查询特定字段。如果需要查询所有字段则不设置该项。
            request.fetchSourceContext(new FetchSourceContext(true, fields.split(","), Strings.EMPTY_ARRAY));
        }
        GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
        Map map = response.getSource();
        if(map!=null){

            //为返回的数据添加id
            map.put("id", response.getId());
        }
        return map;
    }

    
    public boolean existsById(String index, String id) throws IOException {
        GetRequest request = new GetRequest(index, id);
        //不获取返回的_source的上下文
        request.fetchSourceContext(new FetchSourceContext(false));
        request.storedFields("_none_");
        return restHighLevelClient.exists(request, RequestOptions.DEFAULT);
    }

    
    public RestClient getLowLevelClient() {
        return restHighLevelClient.getLowLevelClient();
    }


    
    public Map setSearchResponses(SearchResponse searchResponse, String highlightField, Integer total) {
        //解析结果
        ArrayList> list = new ArrayList<>();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map high = hit.getHighlightFields();
            HighlightField title = high.get(highlightField);

            hit.getSourceAsMap().put("id", hit.getId());

            Map sourceAsMap = hit.getSourceAsMap();//原来的结果
            //解析高亮字段,将原来的字段换为高亮字段
            if (title != null) {
                Text[] texts = title.fragments();
                String nTitle = "";
                for (Text text : texts) {
                    nTitle += text;
                }
                //替换
                sourceAsMap.put(highlightField, nTitle);
            }
            list.add(sourceAsMap);
        }
        Map map = new HashMap<>();
        map.put("data", list);
        map.put("total", total);
        return map;
    }


    
    public Map setSearchResponse(SearchResponse searchResponse, String highlightField, Integer total) {
        //解析结果
        ArrayList> list = new ArrayList<>();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map result = new HashMap<>();
            Map high = hit.getHighlightFields();
            Integer indexs = highlightField.indexOf(",");
            if (indexs > 0) {
                for (String str : highlightField.split(",")) {
                    HighlightField title = high.get(str);

                    hit.getSourceAsMap().put("id", hit.getId());

                    Map sourceAsMap = hit.getSourceAsMap();//原来的结果
                    //解析高亮字段,将原来的字段换为高亮字段
                    if (title != null) {
                        Text[] texts = title.fragments();
                        String nTitle = "";
                        for (Text text : texts) {
                            nTitle += text;
                        }
                        //替换
                        sourceAsMap.put(str, nTitle);


                    }
                    result = sourceAsMap;
                }
                list.add(result);
            } else {
                HighlightField title = high.get(highlightField);

                hit.getSourceAsMap().put("id", hit.getId());

                Map sourceAsMap = hit.getSourceAsMap();//原来的结果
                //解析高亮字段,将原来的字段换为高亮字段
                if (title != null) {
                    Text[] texts = title.fragments();
                    String nTitle = "";
                    for (Text text : texts) {
                        nTitle += text;
                    }
                    //替换
                    sourceAsMap.put(highlightField, nTitle);

                }

                list.add(sourceAsMap);
            }


        }
        Map map = new HashMap<>();
        map.put("data", list);
        map.put("total", total);
        return map;
    }

    public Map setSearchResponse(SearchResponse searchResponse, Integer total) {
        Map map = new HashMap<>();
        List> list = new ArrayList<>();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map sourceAsMap = hit.getSourceAsMap();
            sourceAsMap.put("id",hit.getId());
            list.add(sourceAsMap);
        }
        map.put("data",list);
        map.put("total", total);
        return map;
    }

    
    public Map searchListData(String index,
                                              SearchSourceBuilder query,
                                              Integer size,
                                              Integer from,
                                              String fields,
                                              String sortField,
                                              String highlightField) throws IOException {
        SearchRequest request = new SearchRequest(index);
        SearchSourceBuilder builder = query;
        if (StringUtils.isNotEmpty(fields)) {
            //只查询特定字段。如果需要查询所有字段则不设置该项。
            builder.fetchSource(new FetchSourceContext(true, fields.split(","), Strings.EMPTY_ARRAY));
        }
        from = from <= 0 ? 0 : from * size;
        //设置确定结果要从哪个索引开始搜索的from选项，默认为0
        builder.from(from);
        builder.size(size);
        if (StringUtils.isNotEmpty(sortField)) {
            //排序字段，注意如果proposal_no是text类型会默认带有keyword性质，需要拼接.keyword
            builder.sort(sortField, SortOrder.DESC);
        }
        if (ObjectUtil.isNotNull(highlightField)){
            Integer indexs = highlightField.indexOf(",");
            HighlightBuilder highlight = new HighlightBuilder();
            if (indexs > 0) {
                for (String str : highlightField.split(",")) {
                    highlight.field(str);
                }

            } else {
                highlight.field(highlightField);


            }

            highlight.preTags("");
            highlight.postTags("");
            highlight.highlighterType("unified");
            highlight.requireFieldMatch(true);//多次段高亮需要设置为false
            //下面这两项,如果你要高亮如文字内容等有很多字的字段,必须配置,不然会导致高亮不全,文章内容缺失等
            highlight.fragmentSize(800000); //最大高亮分片数
            highlight.numOfFragments(0); //从第一个分片获取高亮片段
            //不返回源数据。只有条数之类的数据。
            //builder.fetchSource(false);
            builder.highlighter(highlight);
        }

        request.source(builder);
        //SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);

        RequestOptions.Builder option = RequestOptions.DEFAULT.toBuilder();
        option.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(500 * 1024 * 1024));
        SearchResponse response = restHighLevelClient.search(request, option.build());

        if (response.status().getStatus() == 200) {
            // 解析对象
            if (ObjectUtil.isNotNull(highlightField)){
                return setSearchResponse(response, highlightField, (int) response.getHits().getTotalHits().value);
            }
            return setSearchResponse(response, (int) response.getHits().getTotalHits().value);
        }
        return null;
    }


    
    public List> searchPage(String index, String keyword, int pageNo, int pageSize) throws IOException {
//        指定要查询的索引
        SearchRequest request = new SearchRequest(index);

//        设置分页条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        pageNo = pageNo == 0 ? 0 : pageNo;
        pageSize = pageSize == 0 ? 10 : pageSize;
        searchSourceBuilder.from(pageNo);
        searchSourceBuilder.size(pageSize);

//        输入的关键字匹配的字段
        TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("opeInterface.keyword", keyword);
        searchSourceBuilder.query(termQueryBuilder);


//        执行
        request.source(searchSourceBuilder);
        SearchResponse search = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        ArrayList> list = new ArrayList<>();
        search.getHits().forEach(hit -> {
            Map map = hit.getSourceAsMap();
            list.add(map);

        });

        return list;

    }


}

elasticsearch的基本api

Java相关栏目本月热门文章