java 操作 Es 的 CURD 操作

samlai

2020-01-30

java 操作 Es 的 CURD 操作

学习参考文章
maven 引入
参考内容
Java 代码 CURD

学习参考文章

https://blog.csdn.net/u014646662/category_8747782.html

maven 引入


<properties>
        <es.version>6.4.2</es.version>
</properties>

<dependencies>

    <!-- https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch -->
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>${es.version}</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.elasticsearch.client/transport -->
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>transport</artifactId>
        <version>${es.version}</version>
        <exclusions>
            <exclusion>
                <groupId>org.elasticsearch</groupId>
                <artifactId>elasticsearch</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
</dependencies>

参考内容

参考文章 : https://www.jianshu.com/p/a584848da515
验证数据 : http://localhost:5601/app/kibana#/dev_tools

Java 代码 CURD

package com.demo;


import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Before;
import org.junit.Test;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * 参考文章:
 *  https://www.jianshu.com/p/a584848da515
 *
 * es 服务:
 *  http://localhost:9200/
 *  来查看 :  cluster.name
 *
 * kibana 服务:
 *  http://localhost:5601/app/kibana#/dev_tools/console?_g=()
 *  来验证数据是否正确
 */

@Slf4j
public class CurdTest {


    TransportClient transportClient = null;

    private final static String hostName = "localhost";
    private final static Integer port = 9300;
    private final static Integer poolSize = 5;

    private final static String index = "db-1";
    private final static String type = "tab-1";




    @Before
    public void init() {
        try {
            // 配置信息
            Settings esSetting = Settings.builder()
                    .put("cluster.name", "docker-cluster") //集群名字
//                    .put("client.transport.sniff", true)//增加嗅探机制，找到ES集群
                    .put("thread_pool.search.size", poolSize)//增加线程池个数，暂时设为5
                    .build();
            //配置信息Settings自定义
            transportClient = new PreBuiltTransportClient(esSetting);
            TransportAddress transportAddress = new TransportAddress(InetAddress.getByName(hostName), port);
            transportClient.addTransportAddresses(transportAddress);
        } catch (Exception e) {
            log.error("elasticsearch TransportClient create error!!", e);
        }
    }






    /**
     *  新增操作
     */
    @Test
    public void oneInsert() {


        String data = "{\n" +
                "  \"cn-name\":\"java 赖\",\n" +
                "  \"en-name\":\"java es lai\",\n" +
                "  \"age\":28,\n" +
                "  \"content\":\"code 操作....\"\n" +
                "}";

        //这个版本需要把 json 字符串转为 map 对象  不然会报错

        Map map = JSON.parseObject(data, Map.class);

        IndexResponse response = transportClient.prepareIndex(index, type).setSource(map).get();
        log.info("addData response status:{},id:{}", response.status().getStatus(), response.getId());


    }


    /**
     *  批量的插入的操作
     */
    @Test
    public void batchInsert() {
        BulkRequestBuilder bulkRequest = transportClient.prepareBulk();
                for (int i = 60; i < 100; i++) {
                    Map map = Maps.newHashMap();
                    map.put("cn-name", "00" + i);
                    map.put("en-name", "java samlai" + i);
                    map.put("age", 10 + i);
                    map.put("content", " 内容  -->" + i);
                    IndexRequest request = transportClient.prepareIndex(index, type, String.valueOf(i)).setSource(map).request();
                    bulkRequest.add(request);
                }
        
                bulkRequest.execute().actionGet();
    }

    /**
     *  删除操作
     */
    @Test
    public void del() {
        String id = "dpcY7G8BVtcun_3JK7HX";
        DeleteResponse response = transportClient.prepareDelete(index, type, id).execute().actionGet();
        log.info("deleteDataById response status:{},id:{}", response.status().getStatus(), response.getId());
    }


    /**
     *  删除 index 操作
     *
     */
    @Test
    public void delIndex() {
        if (!isIndexExist(transportClient, index)) {
            log.info("Index is not exits!");
        }
        DeleteIndexResponse dResponse = transportClient.admin().indices().prepareDelete(index).execute().actionGet();
        if (dResponse.isAcknowledged()) {
            log.info("delete index " + index + "  successfully!");
        } else {
            log.info("Fail to delete index " + index);
        }
//        return dResponse.isAcknowledged();
        log.info("del Index  :   " + dResponse.isAcknowledged());
    }


    /**
     * 修改操作
     */
    @Test
    public void update() throws ExecutionException, InterruptedException {


        String data = "{\n" +
//                "  \"cn-name\":\"java 赖  update\"\n" +
                "  \"cn-name\":\"java 赖  update\",\n" +
                "  \"en-name\":\"java es lai  ipdate\",\n" +
                "  \"age\":13,\n" +
                "  \"content\":\"update content\"\n" +
                "}";


        String id = "d5dG7G8BVtcun_3JnrGp";

        //这个版本需要把 json 字符串转为 map 对象  不然会报错

        Map map = JSON.parseObject(data, Map.class);

        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index(index).type(type).id(id).doc(map);
        transportClient.update(updateRequest);

        UpdateResponse result = transportClient.update(updateRequest).get();

        // 默认情况下，不更改任何内容的更新会检测到它们不会更改任何内容，并返回“结果”：“noop”

        log.info("");

        log.info(" result :  " + JSON.toJSONString(result));

        log.info(" update data  ：  " + JSON.toJSONString(result.getResult()));

        log.info("");

    }






    /**
     * 根据 id 来查找 es 的数据内容
     */
    @Test
    public void queryId() {

        String id = "d5dG7G8BVtcun_3JnrGp";

        String fields = "";


        GetRequestBuilder getRequestBuilder = transportClient.prepareGet(index, type, id);

        if (StringUtils.isNotEmpty(fields)) {
            getRequestBuilder.setFetchSource(fields.split(","), null);
        }

        GetResponse getResponse = getRequestBuilder.execute().actionGet();

        log.info("");
        log.info(" reponse  data  :  " + JSON.toJSONString(getResponse));
        log.info(" data  :  " + getResponse.getSource());
        log.info("");
    }








    /**
     * 使用分词查询,并分页
     *
     * @param index          索引名称
     * @param type           类型名称,可传入多个type逗号分隔
     * @param startPage      当前页
     * @param pageSize       每页显示条数
     * @param query          查询条件
     * @param fields         需要显示的字段，逗号分隔（缺省为全部字段）
     * @param sortField      排序字段
     * @param highlightField 高亮字段
     * @return
     */
    @Test
    public void queryPage() {

        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index);

        String fields = "content,en-name";
        String sortField = "age";
        String highlightField = "content";
//        String highlightField = "cn-name";
        QueryBuilder query = QueryBuilders.boolQuery();


        int startPage = 1;
        int pageSize = 10;

        if (StringUtils.isNotEmpty(type)) {
            searchRequestBuilder.setTypes(type.split(","));
        }
        searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH);

        // 需要显示的字段，逗号分隔（缺省为全部字段）
        if (StringUtils.isNotEmpty(fields)) {
            searchRequestBuilder.setFetchSource(fields.split(","), null);
        }

        //排序字段
        if (StringUtils.isNotEmpty(sortField)) {
            searchRequestBuilder.addSort(sortField, SortOrder.DESC);
        }

        // 高亮（xxx=111,aaa=222）
        if (StringUtils.isNotEmpty(highlightField)) {

            HighlightBuilder highlightBuilder = new HighlightBuilder();

            highlightBuilder.preTags("<span style='color:red' >");//设置前缀
            highlightBuilder.postTags("</span>");//设置后缀

            // 设置高亮字段
            highlightBuilder.field(highlightField);
            searchRequestBuilder.highlighter(highlightBuilder);
        }

        //searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
        searchRequestBuilder.setQuery(query);

        // 分页应用
        searchRequestBuilder.setFrom(startPage).setSize(pageSize);

        // 设置是否按查询匹配度排序
        searchRequestBuilder.setExplain(true);

        //打印的内容 可以在 Elasticsearch head 和 Kibana  上执行查询
        log.info("\n{}", searchRequestBuilder);

        // 执行搜索,返回搜索响应信息
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();

        long totalHits = searchResponse.getHits().totalHits;
        long length = searchResponse.getHits().getHits().length;

        log.info("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);

        if (searchResponse.status().getStatus() == 200) {
            // 解析对象
            List<Map<String, Object>> sourceList = setSearchResponse(searchResponse, highlightField);

            log.info(" ");
            log.info(" startPage  :  " + startPage);
            log.info(" pageSize  :  " + pageSize);
            log.info(" totalHits  :  " + totalHits);
            log.info(" sourceList  :  " + JSON.toJSONString(sourceList));
            log.info(" ");

        }

    }


    /**
     * 判断索引是否存在
     *
     * @param index
     * @return
     */
    public static boolean isIndexExist(TransportClient transportClient, String index) {

        IndicesExistsResponse inExistsResponse = transportClient.admin().indices().exists(new IndicesExistsRequest(index)).actionGet();

        if (inExistsResponse.isExists()) {
            log.info("Index [" + index + "] is exist!");
        } else {
            log.info("Index [" + index + "] is not exist!");
        }
        return inExistsResponse.isExists();
    }


    /**
     * @Description: 判断inde下指定type是否存在
     */
    public boolean isTypeExist(TransportClient transportClient, String index, String type) {
        return isIndexExist(transportClient, index)
                ? transportClient.admin().indices().prepareTypesExists(index).setTypes(type).execute().actionGet().isExists()
                : false;
    }




    /**
     * 高亮结果集 特殊处理
     *
     * @param searchResponse
     * @param highlightField
     */
    private static List<Map<String, Object>> setSearchResponse(SearchResponse searchResponse, String highlightField) {
        List<Map<String, Object>> sourceList = new ArrayList<Map<String, Object>>();
        StringBuffer stringBuffer = new StringBuffer();

        for (SearchHit searchHit : searchResponse.getHits().getHits()) {
            searchHit.getSourceAsMap().put("id", searchHit.getId());

            if (StringUtils.isNotEmpty(highlightField)) {

                log.info("遍历 高亮结果集，覆盖 正常结果集" + searchHit.getSourceAsMap());

                HighlightField field = searchHit.getHighlightFields().get(highlightField);

                if (field != null) {

                    Text[] text = field.getFragments();

                    if (text != null) {
                        for (Text str : text) {
                            stringBuffer.append(str.string());
                        }
                        //遍历 高亮结果集，覆盖 正常结果集
                        searchHit.getSourceAsMap().put(highlightField, stringBuffer.toString());
                    }
                }

            }
            sourceList.add(searchHit.getSourceAsMap());
        }
        return sourceList;
    }

}

xxx-job 分布式定时任务系统

Java 操作 Es 查询操作