JestClient操作ES

2020-05-31

字数统计: 2k字 | 阅读时长: 11分

使用JestClient操作ElasticSearch，具体代码如下

InitElasticSearchConfig.java

package com.mdl.monitor.init;

import com.google.gson.GsonBuilder;

import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;


/**
 * 初始化es
 */
public class InitElasticSearchConfig {

    private JestClient client ;

    public JestClient getClient() {
        return client;
    }

    public InitElasticSearchConfig(String esUrl){
        client = getClientConfig(esUrl) ;
    }

    public JestClient getClientConfig(String esUrl){
        JestClientFactory factory = new JestClientFactory();
        factory.setHttpClientConfig(new HttpClientConfig
                .Builder(esUrl)
                .gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss").create())
                .multiThreaded(true)
                .readTimeout(10000)
                .build());
        JestClient client = factory.getObject();
        return client ;
    }

}

ElasticSearchDao.java

package com.mdl.monitor.repositorys.elasticsearch;

import java.util.List;

import com.google.gson.JsonObject;

import io.searchbox.client.JestResult;
import io.searchbox.core.SearchResult.Hit;
import io.searchbox.core.SuggestResult;

/**
 * ES操作 抽象方法 基本包含所有基本操作
 */
public interface ElasticSearchDao {

    /**
     * 删除索引
     * @param type ：当前删除document名称
     * @return
     */
    public JestResult deleteIndex(String type) ;

    //清除缓存
    public JestResult clearCache() ;

    /**
     * 关闭索引
     * @param type ：文档表示的对象类别
     * @return
     */
    public JestResult closeIndex(String type) ;

    //优化索引
    public JestResult optimizeIndex() ;

    //刷新索引
    public JestResult flushIndex();

    //判断索引是否存在
    public JestResult indicesExists();

    //查看节点信息
    public JestResult nodesInfo();

    //查看集群健康信息
    public JestResult health();

    //节点状态
    public JestResult nodesStats();

    /**
     * 更新Document
     * @param index ：文档在哪存放
     * @param type ： 文档表示的对象类别
     * @param id ：文档唯一标识
     */
    public JestResult updateDocument(String script , String index,String type,String id);

    /**
     * 删除Document
     * @param index ：文档在哪存放
     * @param type ： 文档表示的对象类别
     * @param id ：文档唯一标识
     * @return
     */
    public JestResult deleteDocument(String index,String type,String id) ;

/**
     * 根据条件删除
     * @param index
     * @param type
     * @param params
     */
    public JestResult deleteDocumentByQuery(String index, String type, String params);


    /**
     * 获取Document
     * @param o ：返回对象
     * @param index ：文档在哪存放
     * @param type ： 文档表示的对象类别
     * @param id ：文档唯一标识
     * @return
     */
    public <T> JestResult getDocument(T o , String index , String type , String id) ;

    //Suggestion
    public List<SuggestResult.Suggestion> suggest() ;

    /**
     * 查询全部
     * @param index ：文档在哪存放
     * @return
     */
    public <T> List<Hit<T,Void>> searchAll(String index , T o);

    /**
     * 搜索
     * @param keyWord ：搜索关键字
     * @return 
     */
    public <T> List<Hit<T,Void>> createSearch(String keyWord , String type , T o , String... fields) ;

    //bulkIndex操作
    public <T> void bulkIndex(String index , String type , T o) ;

    /**
     * 创建索引
     * @param o ：返回对象
     * @param index ：文档在哪存放
     * @param type ： 文档表示的对象类别
     * @return
     */
    public <T> JestResult createIndex(T o , String index , String type);

    /**
     * 搜索事件流图表数据
     * @param param
     * @return
     */
    public JsonObject searchEvent(String param); 

}

ElasticSearchDaoImpl.java

package com.mdl.monitor.repositorys.elasticsearch.impl;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.mdl.monitor.init.InitElasticSearchConfig;
import com.mdl.monitor.repositorys.elasticsearch.ElasticSearchDao;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

import io.searchbox.client.JestResult;
import io.searchbox.cluster.Health;
import io.searchbox.cluster.NodesInfo;
import io.searchbox.cluster.NodesStats;
import io.searchbox.core.Bulk;
import io.searchbox.core.Delete;
import io.searchbox.core.Get;
import io.searchbox.core.Index;
import io.searchbox.core.Search;
import io.searchbox.core.SearchResult;
import io.searchbox.core.SearchResult.Hit;
import io.searchbox.core.Suggest;
import io.searchbox.core.SuggestResult;
import io.searchbox.core.SuggestResult.Suggestion;
import io.searchbox.core.Update;
import io.searchbox.indices.ClearCache;
import io.searchbox.indices.CloseIndex;
import io.searchbox.indices.DeleteIndex;
import io.searchbox.indices.Flush;
import io.searchbox.indices.IndicesExists;
import io.searchbox.indices.Optimize;

/**
 * es操作实现类
 */

@Service
public class ElasticSearchDaoImpl implements ElasticSearchDao{

    static protected final Log log = LogFactory.getLog(ElasticSearchDaoImpl.class.getName());

    @Autowired
    private InitElasticSearchConfig esConfig ;

    @Override
    public JestResult deleteIndex(String type) {
        DeleteIndex deleteIndex = new DeleteIndex.Builder(type).build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(deleteIndex);
            log.info("deleteIndex == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult clearCache() {
        ClearCache closeIndex = new ClearCache.Builder().build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(closeIndex);
            log.info("clearCache == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult closeIndex(String type) {
        CloseIndex closeIndex = new CloseIndex.Builder(type).build(); 
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(closeIndex);
            log.info("closeIndex == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult optimizeIndex() {
        Optimize optimize = new Optimize.Builder().build(); 
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(optimize);
            log.info("optimizeIndex == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult flushIndex() {
        Flush flush = new Flush.Builder().build(); 
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(flush);
            log.info("flushIndex == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult indicesExists() {
        IndicesExists indicesExists = new IndicesExists.Builder("article").build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(indicesExists);
            log.info("indicesExists == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult nodesInfo() {
        NodesInfo nodesInfo = new NodesInfo.Builder().build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(nodesInfo);
            log.info("nodesInfo == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult health() {
        Health health = new Health.Builder().build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(health);
            log.info("health == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult nodesStats() {
        NodesStats nodesStats = new NodesStats.Builder().build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(nodesStats);
            log.info("nodesStats == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult updateDocument(String script , String index, String type, String id) {
        /*String script = "{" +
                "    \"doc\" : {" +
                "        \"title\" : \""+article.getTitle()+"\"," +
                "        \"content\" : \""+article.getContent()+"\"," +
                "        \"author\" : \""+article.getAuthor()+"\"," +
                "        \"source\" : \""+article.getSource()+"\"," +
                "        \"url\" : \""+article.getUrl()+"\"," +
                "        \"pubdate\" : \""+new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").format(article.getPubdate())+"\"" +
                "    }" +
                "}";*/
        Update update = new Update.Builder(script).index(index).type(type).id(id).build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(update);
            log.info("updateDocument == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result ;
    }

    @Override
    public JestResult deleteDocument(String index, String type, String id) {
        Delete delete = new Delete.Builder(id).index(index).type(type).build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(delete);
            log.info("deleteDocument == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

@Override
    public JestResult deleteDocumentByQuery(String index, String type, String params) {

        DeleteByQuery db = new DeleteByQuery.Builder(params)
                .addIndex(index)
                .addType(type)
                .build();

        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(db);
            log.info("deleteDocument == " + result.getJsonString());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }


    @Override
    public <T> JestResult getDocument(T object , String index, String type, String id) {
        Get get = new Get.Builder(index, id).type(type).build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(get);     
            T o = (T) result.getSourceAsObject(object.getClass());
            for (Method method : o.getClass().getMethods()) {
                log.info("getDocument == " + method.getName());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;

    }

    @Override
    public List<Suggestion> suggest() {
        String suggestionName = "my-suggestion";
        Suggest suggest = new Suggest.Builder("{" +
                "  \"" + suggestionName + "\" : {" +
                "    \"text\" : \"the amsterdma meetpu\"," +
                "    \"term\" : {" +
                "      \"field\" : \"body\"" +
                "    }" +
                "  }" +
                "}").build();
        SuggestResult suggestResult = null ;
        List<SuggestResult.Suggestion> suggestionList = null ;
        try {
            suggestResult = esConfig.getClient().execute(suggest);
            log.info("suggestResult.isSucceeded() == " + suggestResult.isSucceeded());
            suggestionList = suggestResult.getSuggestions(suggestionName);
            log.info("suggestionList.size() == " + suggestionList.size());
            for(SuggestResult.Suggestion suggestion:suggestionList){
                System.out.println(suggestion.text);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return suggestionList ;
    }

    @Override
    public <T> List<Hit<T, Void>> searchAll(String index , T o) {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        Search search = new Search.Builder(searchSourceBuilder.toString())
                .addIndex(index)
                .build();
        SearchResult result = null ;
        List<?> hits = null ;
        try {
            result = esConfig.getClient().execute(search);
            System.out.println("本次查询共查到："+result.getTotal()+"个关键字！");
            log.info("本次查询共查到："+result.getTotal()+"个关键字！");
            hits = result.getHits(o.getClass());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return (List<Hit<T, Void>>) hits ;
    }

    @Override
    public <T> List<Hit<T, Void>> createSearch(String keyWord , String type , T o , String... fields) {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.queryStringQuery(keyWord));
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        for(String field : fields){
            highlightBuilder.field(field);//高亮field
        }
        highlightBuilder.preTags("<em>").postTags("</em>");//高亮标签
        highlightBuilder.fragmentSize(200);//高亮内容长度
        searchSourceBuilder.highlighter(highlightBuilder);
        Search search = new Search.Builder(searchSourceBuilder.toString()).addIndex(type).build();
        SearchResult result = null ;
        List<?> hits = null ;
        try {
            result = esConfig.getClient().execute(search);
            System.out.println("本次查询共查到："+result.getTotal()+"个结果！");

            hits = result.getHits(o.getClass());

        } catch (IOException e) {
            e.printStackTrace();
        }
        return (List<Hit<T, Void>>) hits ;
    }

    @Override
    public <T> void bulkIndex(String index, String type , T o) {
        Bulk bulk = new Bulk.Builder()
                .defaultIndex(index)
                .defaultType(type)
                .addAction(Arrays.asList(
                        new Index.Builder(o).build()
                )).build(); 
        try {
            esConfig.getClient().execute(bulk);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public <T> JestResult createIndex(T o, String index, String type) {
        Index index1 = new Index.Builder(o).index(index).type(type).build();
        JestResult jestResult = null ;
        try {
            jestResult = esConfig.getClient().execute(index1);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return jestResult;
    }

    @Override
    public JsonObject searchEvent(String param) {
        JsonObject returnData = new JsonParser().parse(param).getAsJsonObject();
        Search search = new Search.Builder(returnData.toString()).addType("event").addIndex("pi").build();

//      Gson gs = new Gson();
//      System.out.println("输入参数为：" + "\n" + gs.toJson(search));

        SearchResult result = null ;
        try {
            result = esConfig.getClient().execute(search);
//          System.out.println("\n" + gs.toJson(result.getJsonObject()) + "\n" );
//          System.out.println("本次查询共查到：" + "\n" +result.getTotal()+"个结果！");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result.getJsonObject();
    }

}

project_servlet.xml

<!-- es连接配置 -->
    <bean id="esConfig" class="com.mdl.monitor.init.InitElasticSearchConfig" >
        <constructor-arg  index="0" value="${elasticUrl}"/>
    </bean>

scroll分页

@Override
    public JsonObject searchEventHistogramByScroll(String scrollId) {
        SearchScroll scroll = new SearchScroll.Builder(scrollId, "1m").build();
        JestResult result = null ;
        try {
            result = esConfig.getClient().execute(scroll);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result.getJsonObject();
    }

    @Override
    public JsonObject searchInitEventHistogram(String param) {
        JsonObject returnData = new JsonParser().parse(param).getAsJsonObject();
        Search search = new Search.Builder(returnData.toString())
                    .addIndex("pi")
                    .addType("event")
                    .setParameter(Parameters.SCROLL, "1m")
                    .build();

        JestResult result = null;

        try {
            result = esConfig.getClient().execute(search);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result.getJsonObject();
    }

本文作者： Mr Han
本文链接： https://linbei.top/JestClient%E6%93%8D%E4%BD%9CElasticSearch/
版权声明： 本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。转载请注明出处！