java ES RestHighLevelClient客户端“单机“、集群配置

本文介绍，SpringBoot整合ES时RestHighLevelClient的配置,包括单机版本，以及集群版本如何配置（不涉及如何使用RestHighLevelClient）

Spring Boot版本：2.2.5 ES版本：6.8.6

spring:
  elasticsearch:
    rest:
      ip: 10.218.223.73
      port: 9200
      scheme: http

@Configuration
public class EsConfig {

   @Value("${spring.elasticsearch.rest.ip}")
   private String ip;

   @Value("${spring.elasticsearch.rest.port}")
   private int port;

   @Value("${spring.elasticsearch.rest.scheme}")
   private String scheme;
   
   @Bean
   public RestHighLevelClient restHighLevelClient() {
     //RestClient.builder()可接收一个或多个HttpHost
      return new RestHighLevelClient(RestClient.builder(
            new HttpHost(ip, port, scheme)
      ));
   }
}

其实无非就是创建一个HttpHost，并传入ip、port、scheme。这样就完成了单机版本的配置。

集群就是要为每台机器创建一个HttpHost。HttpHost提供了一个create方法，接收一个url，它会根据url，拆解出sheme、ip、host，然后创建一个HttpHost。

基于这个方法，我们可以这么配置我们的url：

spring:
  elasticsearch:
    rest:
      uris: http://10.218.223.73:9200,http://10.218.223.132:9200

@Value("${spring.elasticsearch.rest.uris}")
private String[] uris;

@Bean
public RestHighLevelClient restHighLevelClient() {
   // 创建多个HttpHost
   HttpHost[] httpHosts = Arrays.stream(uris).map(HttpHost::create).toArray(HttpHost[]::new);

   return new RestHighLevelClient(RestClient.builder(httpHosts));
}

如果你对java8熟悉的话，上面的代码对你来说应该很简单。如果实在看不懂，上面的方法其实等价于：

public RestHighLevelClient restHighLevelClient() {
   HttpHost[] httpHosts = new HttpHost[uris.length];
   for (int i = 0; i < uris.length; i++) {
      httpHosts[i] = HttpHost.create(uris[i]);
   }

   return new RestHighLevelClient(RestClient.builder(httpHosts));
}

不过还是强烈建议您去学一下java8，相信你会爱上它的。

ES集群之间的节点能够互相发现的，所以在集群模式下，我们与单机一样，只配置一个HttpHost也是可行的。
但如果碰巧你配置的那台机器挂掉了，就会导致服务不可用。因此还是建议将所有机器都配上。
此外虽然我一直说单机，但对于ES来说，它单台机器就是集群模式，我说单机只是为了方便理解。

正常情况下，集群版本的配置，就可以满足我们的需求了。但现在，我们系统中使用了俩套ES集群，一套用来记录与业务相关的数据，一套用来记录与日志相关的数据，并且，俩套集群有各自的配置，如最大连接数、连接超时时间等。

那么我们就需要重新规划我们配置文件的结构了。

spring:
  elasticsearch:
    mappers:
      # 业务ES的配置
      order:
        scheme: 'http'
        nodes:
          - { ip: '10.218.223.132', port: 9300, httpPort: 9200 }
          - { ip: '10.218.223.73', port: 9300, httpPort: 9200 }
        pool: 50
        connectTimeout: 5000
        socketTimeout: 40000
        connectionRequestTimeout: 1000
        maxConnectNum: 100
      # 日志ES的配置
      log:
        scheme: 'http'
        nodes:
          - { ip: '10.218.223.132', port: 9300, httpPort: 9200 }
          - { ip: '10.218.223.73', port: 9300, httpPort: 9200 }
        pool: 30
        connectTimeout: 5000
        socketTimeout: 40000
        connectionRequestTimeout: 1000
        maxConnectNum: 100
        maxConnectPerRoute: 100

mappers对应的是一个Map，key表示ES集群的作用，value表示这个集群的具体配置。
所以从配置文件，我们可以看出，我们有俩套ES集群，一套是order服务使用的，一套是专门用来记录日志的。

定义与配置文件结构匹配的类：

@Data
@Configuration
@ConfigurationProperties(prefix = "spring.elasticsearch")
public class EsMapperConfig {

    private Map mappers;

    
    public EsServerConfig getEsServerConfig(String serverType) {
        return mappers.get(serverType);
    }
}

@Data
public class EsServerConfig {
    
    private String scheme = "http";
    
    private int connectTimeout = 5000;
    
    private int socketTimeout = 40000;
    
    private int connectionRequestTimeout = 1000;
    
    private int maxConnectNum = 100;
    
    private int maxConnectPerRoute = 100;
    
    private List nodes = new ArrayList<>();
}

@Data
public class EsServerBean {
    
    private String ip;
    
    private int port;
    
    private int httpPort;
}

@Configuration
public class EsConfig {
    // 进阶版本
    @Resource
    private EsMapperConfig esMapperConfig;
    
    // order服务使用的ES
    @Bean
    public RestHighLevelClient orderHighLevelClient() {
       EsServerConfig esServerConfig = esMapperConfig.getEsServerConfig("order");
    
       return createRestClient(esServerConfig);
    }
    
    // 日志使用的ES
    @Bean
    public RestHighLevelClient logHighLevelClient() {
       EsServerConfig esServerConfig = esMapperConfig.getEsServerConfig("log");
    
       return createRestClient(esServerConfig);
    }
    
    
    private RestHighLevelClient createRestClient(EsServerConfig esServerConfig){
       HttpHost[] httpHosts = esServerConfig.getNodes().stream()
             .map(node -> new HttpHost(node.getIp(), node.getHttpPort(), esServerConfig.getScheme()))
             .toArray(HttpHost[]::new);
    
       RestClientBuilder builder = RestClient.builder(httpHosts);
    
       // 异步连接延时配置
       builder.setRequestConfigCallback(requestConfigBuilder -> {
          requestConfigBuilder.setConnectTimeout(esServerConfig.getConnectTimeout());
          requestConfigBuilder.setSocketTimeout(esServerConfig.getSocketTimeout());
          requestConfigBuilder.setConnectionRequestTimeout(esServerConfig.getConnectionRequestTimeout());
          return requestConfigBuilder;
       });
    
       // 异步连接数配置
       builder.setHttpClientConfigCallback(httpClientBuilder -> {
          httpClientBuilder.setMaxConnTotal(esServerConfig.getMaxConnectNum());
          httpClientBuilder.setMaxConnPerRoute(esServerConfig.getMaxConnectPerRoute());
          return httpClientBuilder;
       });
    
       return new RestHighLevelClient(builder);
    }
}