java优雅的使用elasticsearch api

本文给出一种优雅的拼装elasticsearch查询的方式，可能会使得使用elasticsearch的方式变得优雅起来，使得代码结构很清晰易读。

建立elasticsearch连接部分请参看另一篇博客：http://www.cnblogs.com/kangoroo/p/7127003.html

1、ESDao -- 类似DB中的dao层

封装了增删改查ES的各种方法

package com.dqa.sentinel.client.es;

import com.xiaoju.dqa.sentinel.client.es.entity.ESDocument;
import com.xiaoju.dqa.sentinel.client.es.entity.SearchDocument;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetRequestBuilder;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import static org.elasticsearch.index.query.QueryBuilders.termsQuery;


@Component
public class ESDao {

    @Autowired
    private ESClient esClient;

    public void createCache(ESDocument doc) throws Exception {
        IndexResponse response = esCache.getClient().prepareIndex(doc.getIndexName(), doc.getTypeName(), doc.getId())
                .setSource(doc.getSource().getBytes(Charset.forName("UTF-8"))).setRefresh(true).get();
        if (!response.isCreated()) {
            throw new Exception();
        }
    }

    public Map<String, Object> get(ESDocument doc) throws Exception {
        try {
            GetResponse response = esClient.getClient().prepareGet(doc.getIndexName(), doc.getTypeName(), doc.getId()).get();
            return response.getSourceAsMap();
        } catch (Exception ex) {
            throw new Exception();
        }
    }

    public void create(ESDocument doc) throws Exception {
        IndexResponse response = esClient.getClient().prepareIndex(doc.getIndexName(), doc.getTypeName(), doc.getId())
                .setSource(doc.getSource().getBytes(Charset.forName("UTF-8"))).setRefresh(true).get();
        if (!response.isCreated()) {
            throw new Exception();
        }
    }

    public void mutiGet(List<ESDocument> docList) throws Exception {
        MultiGetRequestBuilder multiGetRequestBuilder = esClient.getClient().prepareMultiGet();

        for (ESDocument doc : docList) {
            multiGetRequestBuilder.add(doc.getIndexName(), doc.getTypeName(), doc.getId());
        }
        MultiGetResponse multiGetItemResponses = multiGetRequestBuilder.get();

        for (MultiGetItemResponse itemResponse : multiGetItemResponses) {
            GetResponse response = itemResponse.getResponse();
            if (response.isExists()) {
                String json = response.getSourceAsString();
                System.out.println(json);
            }
        }
    }

    public void update(ESDocument doc) throws Exception {
        try {
            UpdateRequest request = new UpdateRequest().index(doc.getIndexName())
                    .type(doc.getTypeName()).id(doc.getId())
                    .doc(doc.getSource().getBytes(Charset.forName("UTF-8"))).refresh(true);
            UpdateResponse response = esClient.getClient().update(request).get();
            long docVersion = response.getVersion();
            if (!response.isCreated()) {
                throw new Exception();
            }
        } catch (Exception e) {
            throw new Exception();
        }
    }

    public void deleteById(ESDocument doc) throws Exception{
        try {
            DeleteResponse response = esClient.getClient().prepareDelete(doc.getIndexName(),
                    doc.getTypeName(), doc.getId())
                    .execute()
                    .actionGet();
            long docVersion = response.getVersion();
            if (!response.isFound()) {
                throw new Exception();
            }
        } catch (Exception e) {
            throw new Exception();
        }

    }

    public SearchHit[] searchClient(SearchDocument searchDocument) {
        BoolQueryBuilder query = QueryBuilders.boolQuery();
        // term
        for (Map<String, String> termMap : searchDocument.getTermList()) {
            TermQueryBuilder termQueryBuilder = termQuery(termMap.get("field"), termMap.get("value"));
            query.must(termQueryBuilder);
        }
        // terms
        for (Map<String, Object> termsMap : searchDocument.getTermsList()) {
            if (termsMap.size() != 0) {
                TermsQueryBuilder termsQueryBuilder = termsQuery(termsMap.get("field").toString(), (Collection<?>) termsMap.get("terms"));
                query.must(termsQueryBuilder);
            }
        }
        // range
        for (Map rangeMap : searchDocument.getRangeList()) {
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(rangeMap.get("field").toString()).from(rangeMap.get("from")).to(rangeMap.get("to")).includeLower(true).includeUpper(false);
            query.must(rangeQueryBuilder);
        }
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(query);
        SearchRequestBuilder searchRequestBuilder = esClient.getClient()
                .prepareSearch(searchDocument.getIndexName()).setTypes(searchDocument.getTypeName());
        SearchResponse response = searchRequestBuilder
                .setQuery(query)
                .setFrom(0).setSize(10000)
                .setExplain(true).execute().actionGet();
        SearchHits hits = response.getHits();
        return hits.getHits();
    }

    

 

2、查询字段的传入我们是通过Document类完成的

有三个Document，BaseDocument，ESDocument，SearchDocument，后两个继承前一个。

BaseDocument：

public class BaseDocument {
    private String indexName;
    private String typeName;

    public String getIndexName() {
        return this.indexName;
    }
    public void setIndexName(String indexName) {
        this.indexName = indexName;
    }
    public String getTypeName() {
        return this.typeName;
    }
    public void setTypeName(String typeName) {
        this.typeName = typeName;
    }

}

ESDocument：

public class ESDocument extends BaseDocument {
    private String id;
    private String source;
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getSource() {
        return source;
    }
    public void setSource(String source) {
        this.source = source;
    }

}

SearchDocument

public class SearchDocument extends BaseDocument {
    private List<Map<String, String>> termList;
    private List<Map<String, String>> rangeList;
    private List<Map<String, Object>> termsList;
    public SearchDocument() {
        super();
        termList = new ArrayList<>();
        rangeList = new ArrayList<>();
        termsList = new ArrayList<>();
    }
    public List<Map<String, String>> getTermList() {
        return this.termList;
    }
    public void setTermList(List<Map<String, String>> termList) {
        this.termList = termList;
    }
    public List<Map<String, String>> getRangeList() {
        return this.rangeList;
    }
    public void setRangeList(List<Map<String, String>> rangeList) {
        this.rangeList = rangeList;
    }

    public void setTermsList(List<Map<String, Object>> termsList) {
        this.termsList = termsList;
    }
    public List<Map<String, Object>> getTermsList() {
        return termsList;
    }
}

 

3、使用的时候，拼装Document，并通过ESDao查询即可

例如：

　　　　　SearchDocument searchDocument = new SearchDocument();
        // term查询, field:topicName
        Map<String, String> topicMap = new HashMap<>();
        topicMap.put("field", "topicName");
        topicMap.put("value", topicName.toLowerCase());
        List<Map<String, String>> termList = new ArrayList<>();
        termList.add(topicMap);

        // range查询, field:timestamp
        Map<String, String> timeRange =  new HashMap<>();
        timeRange.put("field", "timestamp");
        timeRange.put("from", from);
        timeRange.put("to", to);
        List<Map<String, String>> rangeList = new ArrayList<>();
        rangeList.add(timeRange);

        searchDocument.setTermList(termList);
        searchDocument.setRangeList(rangeList);
        searchDocument.setIndexName("audit");
        searchDocument.setTypeName("kafka");
        SearchHit[] searchHits = esDao.searchCache(searchDocument);
        String hourStatus = "0";
        for (SearchHit hit : searchHits) {
            Map<String, Object> eachMap = hit.getSource();
　　　　}