ElastaticSearch es客户端RestHighLevelClient

RestHighLevelClient

es的客户端有 RestClient、 TransportClient 、 RestHighLevelClient，
RestClient、 TransportClient不适合高版本的es，官方推荐使用RestHighLevelClient.

maven配置pom.xml

   <properties>
        <es.version>6.2.1</es.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>${es.version}</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>${es.version}</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>${es.version}</version>
        </dependency>

RestHighLevelClient

import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


public class RestHighLevelUtil {
    private RestHighLevelUtil() {
    }

    /**
     * 连接 RestHighLevelClient 客户端
     * 
     */
    public static RestHighLevelClient getEsClient() {
        //RestHighLevelClient的配置，最好写到properties等文件配置中，方便配置。
        String address = "10.202.116.33:9200";
        String schema = "http";
        int connectTimeout = 5000;
        int socketTimeout = 5000;
        int connectionRequestTimeout = 5000;
        int maxConnectNum = 100;
        int maxConnectPerRoute = 100;
        //参数含义，见以下方法
        return getRestHighLevelClient(address, schema, connectTimeout, socketTimeout
                , connectionRequestTimeout, maxConnectNum, maxConnectPerRoute);

    }

    /**
     * 连接 RestHighLevelClient 客户端
     *
     * @param address      集群地址，如果有多个用“,”隔开
     * @param schema      协议
     * @param connectTimeout    连接超时时间
     * @param socketTimeout     Socket 连接超时时间
     * @param connectionRequestTimeout  获取连接的超时时间
     * @param maxConnectNum             最大连接数
     * @param maxConnectPerRoute        最大连接数
     * @return
     */
    public static RestHighLevelClient getRestHighLevelClient(String address, String schema, int connectTimeout, int socketTimeout
            , int connectionRequestTimeout, int maxConnectNum, int maxConnectPerRoute) {
        // 拆分地址
        List<HttpHost> hostLists = new ArrayList<>();
        String[] hostList = address.split(",");
        for (String addr : hostList) {
            String host = addr.split(":")[0];
            String port = addr.split(":")[1];
            hostLists.add(new HttpHost(host, Integer.parseInt(port), schema));
        }
        // 转换成 HttpHost 数组
        HttpHost[] httpHost = hostLists.toArray(new HttpHost[]{});
        // 构建连接对象
        RestClientBuilder builder = RestClient.builder(httpHost);
        // 异步连接延时配置
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(connectTimeout);
            requestConfigBuilder.setSocketTimeout(socketTimeout);
            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeout);
            return requestConfigBuilder;
        });
        // 异步连接数配置
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(maxConnectNum);
            httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
            return httpClientBuilder;
        });
        return new RestHighLevelClient(builder);
    }

    /**
     * 获取es查询结果
     *
     * @param index
     * @param searchSourceBuilder
     * @return
     */
    public static SearchResponse getSearchResponse(String index, SearchSourceBuilder searchSourceBuilder) {
        RestHighLevelClient restHighLevelClient = getEsClient();
        // 创建查询请求对象，将查询对象配置到其中
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(searchSourceBuilder);
        // 执行查询，然后处理响应结果
        SearchResponse searchResponse = null;
        try {
            //注意，如果是高版本的client， search方法需要加上更一个参数 RequestOptions.DEFAULT
            searchResponse = restHighLevelClient.search(searchRequest);
        } catch (IOException e) {
            log.error("getSearchResponse exception.", e);
        }
        return searchResponse;
    }

    /**
     * 获取es查询结果中的Hits
     *
     * @param searchResponse
     * @return
     */
    public static SearchHits getSearchHits(SearchResponse searchResponse) {
        if (searchResponse == null || searchResponse.getHits() == null) {
            return null;
        }
        SearchHit[] searchHitArray = searchResponse.getHits().getHits();
        if (searchHitArray != null && searchHitArray.length > 0) {
            return searchResponse.getHits();
        }
        return null;
    }

    /**
     * 获取es查询结果中的Hits
     *
     * @param index
     * @param searchSourceBuilder
     * @return
     */
    public static SearchHits getSearchHits(String index, SearchSourceBuilder searchSourceBuilder) {
        SearchResponse searchResponse = getSearchResponse(index, searchSourceBuilder);
        return getSearchHits(searchResponse);
    }

}

新建es索引：

PUT /user_info
{
  "settings": {
      "index": {
        "number_of_shards": "5",
        "number_of_replicas": "1"
      }
    }
  ,"mappings": {  
    "user_info":{  
       "properties": {
          "name": {
            "type": "keyword"
          },
          "address": {
            "type": "text",
            "analyzer": "ik_max_word"
          },
          "createTime": {
            "type": "long"
          },
	  "age": {
            "type": "integer"
          },
          "province": {
            "type": "keyword"
          },
          "weight": {
            "type": "double"
          },
          "birthDay": {
            "type": "date"
          }
        }

      }
    }
}

UserInfo对象 ：

public class UserInfo {
    private String address;
    private String name ;
    private Integer age;
    private String  birthDay;
    private String  province;
    private double  weight;
    private Long createTime;
    //忽略getter、setter
}

RestHighLevelClient 代码示例：

    /**
     * 增加文档信息
     */
    public void addDocument() {
        try {
            // 创建索引请求对象
            IndexRequest indexRequest = new IndexRequest("user_info", "user_info","1");
            // 创建员工信息
            UserInfo userInfo = new UserInfo();
            userInfo.setAge(30);
            userInfo.setName("chen");
            userInfo.setAddress("深圳市福田区");
            userInfo.setBirthDay("1993-01-10");
            userInfo.setCreateTime(System.currentTimeMillis());
            userInfo.setWeight(75.00);
            userInfo.setProvince("北京市");
            // 将对象转换为 byte 数组
            byte[] json = JSON.toJSONBytes(userInfo);
            // 设置文档内容
            indexRequest.source(json, XContentType.JSON);
            // 执行增加文档
            RestHighLevelClient restHighLevelClient = RestHighLevelUtil.getEsClient();
            IndexResponse response = restHighLevelClient.index(indexRequest);
            log.info("创建状态：{}", response.status());
        } catch (Exception e) {
            log.error("addDocument", e);
        }
    }

    /**
     * 获取文档信息
     */
    public void getDocument() {
        try {
            // 获取请求对象
            GetRequest getRequest = new GetRequest("user_info", "user_info", "1");
            // 获取文档信息
            RestHighLevelClient restHighLevelClient = RestHighLevelUtil.getEsClient();
            GetResponse getResponse = restHighLevelClient.get(getRequest);
            // 将 JSON 转换成对象
            if (getResponse.isExists()) {
                UserInfo userInfo = JSON.parseObject(getResponse.getSourceAsBytes(), UserInfo.class);
                log.info("员工信息：{}", userInfo);
            }
        } catch (IOException e) {
            log.error("getDocument.", e);
        }
    }

    /**
     * 更新文档信息
     */
    public void updateDocument() {
        try {
            // 创建索引请求对象
            UpdateRequest updateRequest = new UpdateRequest("user_info", "user_info", "1");
            // 设置员工更新信息
            UserInfo userInfo = new UserInfo();
            userInfo.setName("chen");
            userInfo.setAddress("北京市海淀区");
            userInfo.setProvince("北京市");
            // 将对象转换为 byte 数组
            byte[] json = JSON.toJSONBytes(userInfo);
            // 设置更新文档内容
            updateRequest.doc(json, XContentType.JSON);
            // 执行更新文档
            RestHighLevelClient restHighLevelClient = RestHighLevelUtil.getEsClient();
            UpdateResponse response = restHighLevelClient.update(updateRequest);
            log.info("创建状态：{}", response.status());
        } catch (Exception e) {
            log.error("", e);
        }
    }

    /**
     * 删除文档信息
     */
    public void deleteDocument() {
        try {
            // 创建删除请求对象
            DeleteRequest deleteRequest = new DeleteRequest("user_info", "user_info", "1");
            // 执行删除文档
            RestHighLevelClient restHighLevelClient = RestHighLevelUtil.getEsClient();
            DeleteResponse response = restHighLevelClient.delete(deleteRequest);
            log.info("删除状态：{}", response.status());
        } catch (IOException e) {
            log.error("", e);
        }
    }

    /**
     * 精确查询（查询条件不会进行分词，但是查询内容可能会分词，导致查询不到）
     */
    public void termQuery() {
        String index = "user_info";
        String esField = "name";
        String value = "lin";
        // 构建查询条件（注意：termQuery 支持多种格式查询，如 boolean、int、double、string 等，这里使用的是 string 的查询）
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.termQuery(esField, value));
        log.info("termQuery:{}", searchSourceBuilder.toString());
        // 创建查询请求对象，将查询对象配置到其中
        SearchHits hits = RestHighLevelUtil.getSearchHits(index, searchSourceBuilder);
        List<UserInfo> userInfoList = getUserList(hits);
        log.info(userInfoList.toString());

    }



    /**
     * 多个内容在一个字段中进行查询
     */
    public void termsQuery() {
        String index = "user_info";
        String filterName = "name" ;
        String[] values = new String[]{"lin", "zhang"};
        // 构建查询条件（注意：termsQuery 支持多种格式查询，如 boolean、int、double、string 等，这里使用的是 string 的查询）
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.termsQuery(filterName, values));
        log.info("termsQuery:{}", searchSourceBuilder.toString());
        SearchHits hits = RestHighLevelUtil.getSearchHits( index, searchSourceBuilder);
        List<UserInfo> userInfoList = getUserList(hits);
        log.info(userInfoList.toString());
    }


    public List<UserInfo> getUserList(SearchHits hits) {
        if (hits == null) {
            return Collections.emptyList();
        }
        List<UserInfo> userInfoList = new ArrayList<>();
        for (SearchHit hit : hits) {
            UserInfo userInfo = JSON.parseObject(hit.getSourceAsString(), UserInfo.class);
            userInfoList.add(userInfo);
        }
        return userInfoList;
    }


    /**
     * 批量插入数据。
     * 
     */
    public void bulkTest() {
        BulkRequest bulkRequest = new BulkRequest();
        List<IndexRequest> requests = generateRequests();
        for (IndexRequest indexRequest : requests) {
            bulkRequest.add(indexRequest);
        }

        RestHighLevelClient restHighLevelClient = RestHighLevelUtil.getEsClient();
        try {
            restHighLevelClient.bulk(bulkRequest);
        } catch (IOException e) {
            log.error("batchAddTest exception.", e);
        }
    }

    public List<IndexRequest> generateRequests(){
        List<IndexRequest> requests = new ArrayList<>();
        requests.add(generateNewsRequest("wu", 21, 60.00, "2000-01-27", "深圳市", "罗湖区"));
        requests.add(generateNewsRequest("xu", 36, 10.00, "2008-02-27", "深圳市", "福田区"));
        requests.add(generateNewsRequest("wang", 46, 20.00, "2007-03-27", "深圳市", "南山区"));
        requests.add(generateNewsRequest("gu", 26, 30.00, "2012-04-27", "深圳市", "龙华区"));
        return requests;
    }

    public IndexRequest generateNewsRequest(String name, int age, Double weight, String birthDay, String province, String address) {
        IndexRequest indexRequest = new IndexRequest("user_info", "user_info");
        UserInfo userInfo = new UserInfo();
        userInfo.setName(name);
        userInfo.setAge(age);
        userInfo.setWeight(weight);
        userInfo.setBirthDay(birthDay);
        userInfo.setProvince(province);
        userInfo.setAddress(address);
        String source = JSON.toJSONString(userInfo);
        indexRequest.source(source, XContentType.JSON);
        return indexRequest;
    }

参考资料：

https://blog.csdn.net/qq_38241225/article/details/114834170
https://blog.csdn.net/JKjiang123/article/details/106431222