Elasticsearch6中RestHighLevelClient和RestLowLevelClient使用

项目依赖

 <spring.version>4.3.2.RELEASE</spring.version>

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
     <version>6.2.3</version>
</dependency>
 
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
     <version>6.2.3</version>
</dependency>
 <!-- json转换 -->
<dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-databind</artifactId>
</dependency>
 <dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-core</artifactId>
</dependency>
<dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-annotations</artifactId>
</dependency>
 <dependency>
      <groupId>com.alibaba</groupId>
       <artifactId>fastjson</artifactId>
</dependency>

加载配置

package com.chy.els.config;
 
 
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowire;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.io.IOException;
 
/**
* @Title: ElasticsearchConfig
* @Description: Elasticsearch 配置类*/
@Configuration
public class ElasticsearchConfig {
    private String host = "127.0.0.1";
    private int port = 9200;
    private String schema = "http";
    private int connectTimeOut = 1000;
    private int socketTimeOut = 30000;
    private int connectionRequestTimeOut = 500;
    private int maxConnectNum = 100;
    private int maxConnectPerRoute = 100;
    private HttpHost httpHost;
    private boolean uniqueConnectTimeConfig = true;
    private boolean uniqueConnectNumConfig = true;
    private RestClientBuilder builder;
    private RestHighLevelClient client;
 
    @Bean(autowire = Autowire.BY_NAME, name = "restHighLevelClient")
    public RestHighLevelClient client() {
        httpHost= new HttpHost(host, port, schema);
        builder = RestClient.builder(httpHost);
        if (uniqueConnectTimeConfig) {
            setConnectTimeOutConfig();
        }
        if (uniqueConnectNumConfig) {
            setMutiConnectConfig();
        }
        client = new RestHighLevelClient(builder);
        return client;
    }
 
    /**
     * 异步httpclient的连接延时配置
     */
    public void setConnectTimeOutConfig() {
        builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                requestConfigBuilder.setConnectTimeout(connectTimeOut);
                requestConfigBuilder.setSocketTimeout(socketTimeOut);
                requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
                return requestConfigBuilder;
            }
        });
    }
 
 
    /**
     * 异步httpclient的连接数配置
     */
    public void setMutiConnectConfig() {
        builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                httpClientBuilder.setMaxConnTotal(maxConnectNum);
                httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
                return httpClientBuilder;
            }
        });
    }
 
    /**
     * 关闭连接
     */
    public void close() {
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
 
}

工具类

package com.chy.els.util;
 
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
 
/**
* @Title: ElasticsearchUtil
* @Description: 工具类*/
@Component
public class ElasticsearchUtil {
 
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchUtil.class);
 
    @Resource(name="restHighLevelClient")
    private RestHighLevelClient restHighLevelClient;
 
    private static RestHighLevelClient client;
 
    private static ObjectMapper mapper = new ObjectMapper();
 
    /**
     * @PostContruct是spring框架的注解
     * spring容器初始化的时候执行该方法
     */
    @PostConstruct
    public void init() {
        client = this.restHighLevelClient;
    }
 
    /**
     * 创建索引
     *
     * @param index
     * @return
     */
    public static boolean createIndex(String index) {
        //index名必须全小写，否则报错
        CreateIndexRequest request = new CreateIndexRequest(index);
        try {
            CreateIndexResponse indexResponse = client.indices().create(request);
            if (indexResponse.isAcknowledged()) {
                LOGGER.info("创建索引成功");
            } else {
                LOGGER.info("创建索引失败");
            }
            return indexResponse.isAcknowledged();
        } catch (IOException e) {
            e.printStackTrace();
        }
 
        return false;
    }
 
    /**
     * 插入数据
     * @param index
     * @param type
     * @param object
     * @return
     */
    public static String addData(String index,String type,String id,JSONObject object) {
        IndexRequest indexRequest = new IndexRequest(index, type, id);
        try {
            indexRequest.source(mapper.writeValueAsString(object), XContentType.JSON);
            IndexResponse indexResponse = client.index(indexRequest);
            return indexResponse.getId();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
 
    /**
     * 检查索引
     * @param index
     * @return
     * @throws IOException
     */
    public static boolean checkIndexExist(String index) {
        try {
            Response response = client.getLowLevelClient().performRequest("HEAD", index);
            boolean exist = response.getStatusLine().getReasonPhrase().equals("OK");
            return exist;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
 
 
    /**
     * 获取低水平客户端
     * @return
     */
    public static RestClient getLowLevelClient() {
        return client.getLowLevelClient();
    }
}

测试代码

/**
* @Title: EsController
* @Description:*/
@RestController
@RequestMapping("/es")
public class EsController {
 
    private static final Logger logger = LoggerFactory.getLogger(EsController.class);
 
    /**
     * 测试索引
     */
    private String indexName="resthighindex";
 
    /**
     * 类型
     */
    private String esType="normal";
 
    /**
     * 首页
     * @return
     */
    @RequestMapping("/index")
    public String index(){
        return "index";
    }
 
 
    /**
     *  http://localhost:8080/es/createIndex
     * 创建索引
     * @param request
     * @param response
     * @return
     */
    @RequestMapping("/createIndex")
    @ResponseBody
    public String createIndex(HttpServletRequest request, HttpServletResponse response) {
        if (!ElasticsearchUtil.checkIndexExist(indexName)) {
            if (ElasticsearchUtil.createIndex(indexName)) {
                return "索引创建成功";
            } else {
                return "索引已经失败";
            }
        } else {
            return "索引已经存在";
        }
    }
 
    /**
     * 插入记录
     * http://localhost:8080/es/addData
     * @return
     */
    @RequestMapping("/addData")
    @ResponseBody
    public String addData() {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("id", DateUtil.formatDate(new Date()));
        jsonObject.put("age", 29);
        jsonObject.put("name", "liming");
        jsonObject.put("date", new Date());
        String id=ElasticsearchUtil.addData(indexName, esType, jsonObject.getString("id"), jsonObject);
        if(StringUtil.isNotBlank(id)){
            return "插入成功";
        }
        else{
            return "插入失败";
        }
    }
 
    /**
     * 查询所有
     * @return
     */
    @RequestMapping("/queryAll")
    @ResponseBody
    public String queryAll() {
        try {
            HttpEntity entity = new NStringEntity(
                    "{ \"query\": { \"match_all\": {}}}",
                    ContentType.APPLICATION_JSON);
            String endPoint = "/" + indexName + "/" + esType + "/_search";
            Response response = ElasticsearchUtil.getLowLevelClient().performRequest("POST", endPoint, Collections.<String, String>emptyMap(), entity);
            return EntityUtils.toString(response.getEntity());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "查询数据出错";
    }
 
    /**
     * 根据条件查询
     * @return
     */
    @RequestMapping("/queryByMatch")
    @ResponseBody
    public String queryByMatch(){
        try {
            String endPoint = "/" + indexName + "/" + esType + "/_search";
 
            IndexRequest indexRequest = new IndexRequest();
            XContentBuilder builder;
            try {
                builder = JsonXContent.contentBuilder()
                        .startObject()
                        .startObject("query")
                        .startObject("match")
                        .field("name.keyword", "zjj")
                        .endObject()
                        .endObject()
                        .endObject();
                indexRequest.source(builder);
            } catch (IOException e) {
                e.printStackTrace();
            }
 
            String source = indexRequest.source().utf8ToString();
 
            logger.info("source---->"+source);
 
            HttpEntity entity = new NStringEntity(source, ContentType.APPLICATION_JSON);
 
            Response response = ElasticsearchUtil.getLowLevelClient().performRequest("POST", endPoint, Collections.<String, String>emptyMap(), entity);
            return EntityUtils.toString(response.getEntity());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "查询数据出错";
    }
 
    /**
     * 复合查询
     * @return
     */
    @RequestMapping("/queryByCompound")
    @ResponseBody
    public String queryByCompound(){
        try {
            String endPoint = "/" + indexName + "/" + esType + "/_search";
 
            IndexRequest indexRequest = new IndexRequest();
            XContentBuilder builder;
            try {
                /**
                 * 查询名字等于 liming
                 * 并且年龄在30和35之间
                 */
                builder = JsonXContent.contentBuilder()
                        .startObject()
                        .startObject("query")
                            .startObject("bool")
                                  .startObject("must")
                                           .startObject("match")
                                                .field("name.keyword", "liming")
                                           .endObject()
                                  .endObject()
                                  .startObject("filter")
                                           .startObject("range")
                                                 .startObject("age")
                                                     .field("gte", "30")
                                                     .field("lte", "35")
                                                 .endObject()
                                         .endObject()
                                  .endObject()
                           .endObject()
                        .endObject()
                        .endObject();
                indexRequest.source(builder);
            } catch (IOException e) {
                e.printStackTrace();
            }
 
            String source = indexRequest.source().utf8ToString();
 
            logger.info("source---->"+source);
 
            HttpEntity entity = new NStringEntity(source, ContentType.APPLICATION_JSON);
 
            Response response = ElasticsearchUtil.getLowLevelClient().performRequest("POST", endPoint, Collections.<String, String>emptyMap(), entity);
            return EntityUtils.toString(response.getEntity());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "查询数据出错";
    }
 
    /**
     * 删除查询的数据
     * @return
     */
    @RequestMapping("/delByQuery")
    @ResponseBody
    public String delByQuery() {
 
        String deleteText = "chy";
 
        String endPoint = "/" + indexName + "/" + esType + "/_delete_by_query";
 
        /**
         * 删除条件
         */
        IndexRequest indexRequest = new IndexRequest();
        XContentBuilder builder;
        try {
            builder = JsonXContent.contentBuilder()
                    .startObject()
                    .startObject("query")
                    .startObject("term")
                    //name中包含deleteText
                    .field("name.keyword", deleteText)
                    .endObject()
                    .endObject()
                    .endObject();
            indexRequest.source(builder);
        } catch (IOException e) {
            e.printStackTrace();
        }
 
        String source = indexRequest.source().utf8ToString();
 
        HttpEntity entity = new NStringEntity(source, ContentType.APPLICATION_JSON);
        try {
            Response response = ElasticsearchUtil.getLowLevelClient().performRequest("POST", endPoint, Collections.<String, String>emptyMap(), entity);
            return EntityUtils.toString(response.getEntity());
        } catch (IOException e) {
            e.printStackTrace();
        }
 
        return "删除错误";
    }
 
    /**
     * 演示聚合统计
     * @return
     */
    @RequestMapping("/aggregation")
    @ResponseBody
    public String aggregation(){
        try {
            String endPoint = "/" + indexName + "/" + esType + "/_search";
 
            IndexRequest indexRequest = new IndexRequest();
            XContentBuilder builder;
            try {
                builder = JsonXContent.contentBuilder()
                        .startObject()
                        .startObject("aggs")
                                .startObject("名称分组结果")
                                          .startObject("terms")
                                                   .field("field", "name.keyword")
                                                   .startArray("order")
                                                         .startObject()
                                                               .field("_count", "asc")
                                                         .endObject()
                                                   .endArray()
                                          .endObject()
                               .endObject()
                        .endObject()
                        .endObject();
                indexRequest.source(builder);
            } catch (IOException e) {
                e.printStackTrace();
            }
 
            String source = indexRequest.source().utf8ToString();
 
            logger.info("source---->"+source);
 
            HttpEntity entity = new NStringEntity(source, ContentType.APPLICATION_JSON);
 
            Response response = ElasticsearchUtil.getLowLevelClient().performRequest("POST", endPoint, Collections.<String, String>emptyMap(), entity);
            return EntityUtils.toString(response.getEntity());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "查询数据出错";
    }
 
}