ElasticSearch 7.x 集成RestHighLevelClient

Java High Level REST Client是目前官方推荐使用的客户端，Elasticsearch 在7.0中对TransportClient进行弃用，并在8.0中完全删除，替而代之的是Java High Level REST Client。High Level REST Client与 Elasticsearch 具有相同的发布周期。故我们能够使用最新版的 Elasticsearch

参考：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.x/java-rest-high-getting-started-maven.html

Maven依赖

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.1.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>7.1.0</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-client</artifactId>
    <version>7.1.0</version>
</dependency>

elasticsearch必须要单独依赖，否则会出现如下错误:

An attempt was made to call the method org.elasticsearch.action.search.SearchRequest.isCcsMinimizeRoundtrips()Z but it does not exist.

参考：https://github.com/elastic/elasticsearch/issues/41189

集成RestHighLevelClient

为了简化EsClient的创建流程和多es集群的支持，编写一个构造器

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author: yezhou
 * @Date: 2019/5/24 21:04
 * @Version 1.0
 */

@Component
public class EsClientBuilder {
    private int connectTimeoutMillis = 1000;
    private int socketTimeoutMillis = 30000;
    private int connectionRequestTimeoutMillis = 500;
    private int maxConnectPerRoute = 10;
    private int maxConnectTotal = 30;

    private final List<HttpHost> httpHosts;

    private EsClientBuilder(List<HttpHost> httpHosts) {
        this.httpHosts = httpHosts;
    }

    public EsClientBuilder setConnectTimeoutMillis(int connectTimeoutMillis) {
        this.connectTimeoutMillis = connectTimeoutMillis;
        return this;
    }

    public EsClientBuilder setSocketTimeoutMillis(int socketTimeoutMillis) {
        this.socketTimeoutMillis = socketTimeoutMillis;
        return this;
    }

    public EsClientBuilder setConnectionRequestTimeoutMillis(int connectionRequestTimeoutMillis) {
        this.connectionRequestTimeoutMillis = connectionRequestTimeoutMillis;
        return this;
    }

    public EsClientBuilder setMaxConnectPerRoute(int maxConnectPerRoute) {
        this.maxConnectPerRoute = maxConnectPerRoute;
        return this;
    }

    public EsClientBuilder setMaxConnectTotal(int maxConnectTotal) {
        this.maxConnectTotal = maxConnectTotal;
        return this;
    }

    public static EsClientBuilder build(List<HttpHost> httpHosts) {
        return new EsClientBuilder(httpHosts);
    }

    public RestHighLevelClient create() {

        HttpHost[] httpHostArr = httpHosts.toArray(new HttpHost[0]);
        RestClientBuilder builder = RestClient.builder(httpHostArr);

        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(connectTimeoutMillis);
            requestConfigBuilder.setSocketTimeout(socketTimeoutMillis);
            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeoutMillis);
            return requestConfigBuilder;
        });

        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(maxConnectTotal);
            httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
            return httpClientBuilder;
        });

        return new RestHighLevelClient(builder);
    }
}

配置文件application.yml

elasticsearch:
  nodes: 192.168.1.10:9200
  schema: http
  max-connect-total: 50
  max-connect-per-route: 10
  connection-request-timeout-millis: 500
  socket-timeout-millis: 30000
  connect-timeout-millis: 1000

配置类Configuration

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: yezhou
 * @Date: 2019/5/24 21:06
 * @Version 1.0
 */

@Configuration
public class ESConfig {
    @Value("${elasticsearch.nodes}")
    private List<String> nodes;

    @Value("${elasticsearch.schema}")
    private String schema;

    @Value("${elasticsearch.max-connect-total}")
    private Integer maxConnectTotal;

    @Value("${elasticsearch.max-connect-per-route}")
    private Integer maxConnectPerRoute;

    @Value("${elasticsearch.connection-request-timeout-millis}")
    private Integer connectionRequestTimeoutMillis;

    @Value("${elasticsearch.socket-timeout-millis}")
    private Integer socketTimeoutMillis;

    @Value("${elasticsearch.connect-timeout-millis}")
    private Integer connectTimeoutMillis;

    @Bean
    public RestHighLevelClient getRestHighLevelClient() {
        List<HttpHost> httpHosts = new ArrayList<>();
        for (String node : nodes) {
            try {
                String[] parts = StringUtils.split(node, ":");
                Assert.notNull(parts,"Must defined");
                Assert.state(parts.length == 2, "Must be defined as 'host:port'");
                httpHosts.add(new HttpHost(parts[0], Integer.parseInt(parts[1]), schema));
            } catch (RuntimeException ex) {
                throw new IllegalStateException(
                        "Invalid ES nodes " + "property '" + node + "'", ex);
            }
        }

        return EsClientBuilder.build(httpHosts)
                .setConnectionRequestTimeoutMillis(connectionRequestTimeoutMillis)
                .setConnectTimeoutMillis(connectTimeoutMillis)
                .setSocketTimeoutMillis(socketTimeoutMillis)
                .setMaxConnectTotal(maxConnectTotal)
                .setMaxConnectPerRoute(maxConnectPerRoute)
                .create();
    }
}

控制器Controller

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;

/**
 * @Author: yezhou
 * @Date: 2019/5/24 20:13
 * @Version 1.0
 */

@Slf4j
@RestController
public class ElasticSearchController {

    @Resource
    private RestHighLevelClient client;

    @GetMapping("/search")
    @ResponseBody
    public String search() throws IOException {
//        RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(new HttpHost("192.168.165.239", 9200, "http")));

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        boolQueryBuilder.must(QueryBuilders.termQuery("loglvl.keyword", "ERROR"));
//        boolQueryBuilder.must(QueryBuilders.wildcardQuery("loglvl.keyword", "INFO"));
//        boolQueryBuilder.must(QueryBuilders.rangeQuery("field").gt("value"));
//        boolQueryBuilder.must(QueryBuilders.termsQuery("loglvl.keyword", "INFO"));

//        boolQueryBuilder.must(QueryBuilders.matchQuery("loglvl", "ERROR"));

        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(boolQueryBuilder);
        sourceBuilder.sort("@timestamp", SortOrder.DESC);
        sourceBuilder.from(0);
        sourceBuilder.size(10);

        RequestOptions requestOptions = RequestOptions.DEFAULT;

        SearchRequest searchRequest = new SearchRequest("thaipay");
        //searchRequest.searchType(SearchType.QUERY_THEN_FETCH);
        searchRequest.source(sourceBuilder);
        log.info(searchRequest.toString());
        SearchResponse searchResponse = client.search(searchRequest, requestOptions);
        return JSON.toJSONString(searchResponse);
    }

}

如果是cluster，application.yml的nodes设置多个ip:host逗号隔开即可。

ps: 多个集群，在application.yml再定义一组配置，在ESConfig中创建两个name，不同的RestHighLevelClient Bean