ElasticSearch：SpringBoot中连接单机与集群ES的通用连接配置(集群连接时使用ssl认证)

小鱼儿 2022-10-05 09:53 600阅读 0赞

# 1. 背景 #

1.  在做项目的过程，我们通常在本地开发时，需要连接单机的es进行测试，但是正式上线后，需要连接ES集群，并需要ssl认证，这种情况下，既要满足本地开发的需要，又要满足上线后集群的连接配置，故写下本文，用来记录
2.  我比较希望在SpringBoot启动的时候，就建立es连接，故使用了 @PostConstruct 注解，详见下文。

# 2. SpringBoot连接单机或集群ES #

## 2.1 首先定义外部文件elasticsearch.properties ##

我将该文件存放在src/main/resources/config/elasticsearch.properties下【该文件当然也可以写在不同环境的yml文件中】

#ES配置文件
    #是否是单机
    elasticsearch.isSingleton=true
    #ca位置
    elasticsearch.capath=src/main/resources/ca/xxxx
    #这个参数暂时保留
    elasticsearch.keystorepass=
    #账号密码
    elasticsearch.username=username
    elasticsearch.password=password
    # 三个es节点信息
    elasticsearch.node1Ip=11.14.11.72
    elasticsearch.node1Port=9200
    elasticsearch.node1Scheme=http
    elasticsearch.node2Ip=22.94.239.72
    elasticsearch.node2Port=9200
    elasticsearch.node2Scheme=http
    elasticsearch.node3Ip=33.94.239.72
    elasticsearch.node3Port=9200
    elasticsearch.node3Scheme=http

## 2.2 配置一个bean读取上述配置 ##

import lombok.Data;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.PropertySource;
    import org.springframework.stereotype.Component;
    
    
    @Data
    @Component
    @ConfigurationProperties(prefix = "elasticsearch")
    @PropertySource(value = { "classpath:config/elasticsearch.properties"}, encoding = "utf-8")
    public class ESConfig { 
    
        private Boolean isSingleton;
    
        private String capath;
    
        private String keystorepass;
    
        private String username;
    
        private String password;
    
        private String node1Ip;
    
        private Integer node1Port;
    
        private String node1Scheme;
    
        private String node2Ip;
    
        private Integer node2Port;
    
        private String node2Scheme;
    
        private String node3Ip;
    
        private Integer node3Port;
    
        private String node3Scheme;
    
    }

## 2.3 ES连接配置类 ##

import lombok.extern.slf4j.Slf4j;
    import org.apache.http.HttpHost;
    import org.apache.http.auth.AuthScope;
    import org.apache.http.auth.UsernamePasswordCredentials;
    import org.apache.http.client.CredentialsProvider;
    import org.apache.http.conn.ssl.NoopHostnameVerifier;
    import org.apache.http.impl.client.BasicCredentialsProvider;
    import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
    import org.apache.http.ssl.SSLContextBuilder;
    import org.apache.http.ssl.SSLContexts;
    import org.elasticsearch.client.RestClient;
    import org.elasticsearch.client.RestClientBuilder;
    import org.elasticsearch.client.RestHighLevelClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.net.ssl.SSLContext;
    import java.io.InputStream;
    import java.nio.file.Files;
    import java.nio.file.Path;
    import java.nio.file.Paths;
    import java.security.KeyStore;
    
    
    @Slf4j
    @Component
    public class ESClient { 
    
        public RestHighLevelClient restHighLevelClient;
    
        @Autowired
        private ESConfig esConfig;
    
        @PostConstruct
        private void init() { 
            //初始化RestHighLevelClient
            this.initRestHighLevelClient();
        }
    
        private ESClient() { 
        }
    
        private void initRestHighLevelClient() { 
            try { 
                if (restHighLevelClient == null) { 
                    if (!esConfig.getIsSingleton()) { 
                        //正式环境 使用ssl认证 集群连接
                        log.info("******************ES集群连接开始****************");
                        restHighLevelClient = getClusterHighLevelClient();
                        log.info("******************ES集群连接成功****************");
                        return;
                    }
                    //单机连接
                    log.info("******************单机ES连接开始****************");
                    restHighLevelClient = getSingleHighLevelClient();
                    log.info("******************单机ES连接成功****************");
                }
            } catch (Exception e) { 
                log.error("es连接出现异常:{}", e.toString());
            }
        }
    
        /** * 集群连接 * * @return * @throws Exception */
        private RestHighLevelClient getClusterHighLevelClient() throws Exception { 
            Path trustStorePath = Paths.get(esConfig.getCapath());
            KeyStore truststore = KeyStore.getInstance("pkcs12");
    
            String keyStorePass = esConfig.getKeystorepass();
            InputStream is = Files.newInputStream(trustStorePath);
            truststore.load(is, keyStorePass.toCharArray());
    
            SSLContextBuilder sslBuilder = SSLContexts.custom().loadTrustMaterial(truststore, null);
            final SSLContext sslContext = sslBuilder.build();
    
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(esConfig.getUsername(), esConfig.getPassword()));
            log.info("连接ES集群: {}:{},{}:{},{}:{}", esConfig.getNode1Ip(), esConfig.getNode1Port(), esConfig.getNode2Ip(), esConfig.getNode2Port(), esConfig.getNode3Ip(), esConfig.getNode3Port());
            RestClientBuilder rclientBuilder = RestClient.builder(
                    new HttpHost(esConfig.getNode1Ip(), esConfig.getNode1Port(), esConfig.getNode1Scheme()),
                    new HttpHost(esConfig.getNode2Ip(), esConfig.getNode2Port(), esConfig.getNode2Scheme()),
                    new HttpHost(esConfig.getNode3Ip(), esConfig.getNode3Port(), esConfig.getNode3Scheme()))
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() { 
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { 
                            return httpClientBuilder.setSSLContext(sslContext)
                                    .setSSLHostnameVerifier(new NoopHostnameVerifier())
                                    .setDefaultCredentialsProvider(credentialsProvider);
                        }
                    });
            restHighLevelClient = new RestHighLevelClient(rclientBuilder);
            return restHighLevelClient;
        }
    
        /** * 单机连接 * * @return * @throws Exception */
        private RestHighLevelClient getSingleHighLevelClient() { 
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(esConfig.getUsername(), esConfig.getPassword()));
            log.info("连接单机ES: {}:{}", esConfig.getNode1Ip(), esConfig.getNode1Port());
            RestClientBuilder rclientBuilder = RestClient.builder(new HttpHost(esConfig.getNode1Ip(), esConfig.getNode1Port(), esConfig.getNode1Scheme()))
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() { 
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { 
                            return httpClientBuilder.setSSLHostnameVerifier(new NoopHostnameVerifier())
                                    .setDefaultCredentialsProvider(credentialsProvider);
                        }
                    });
            restHighLevelClient = new RestHighLevelClient(rclientBuilder);
            return restHighLevelClient;
        }
    
    }

# 3. 测试 #

import com.alibaba.fastjson.JSONObject;
    import com.qs.myspringboottest.config.dbconnected.ESClient;
    import org.elasticsearch.action.search.SearchRequest;
    import org.elasticsearch.action.search.SearchResponse;
    import org.elasticsearch.client.RequestOptions;
    import org.elasticsearch.client.RestHighLevelClient;
    import org.elasticsearch.index.query.QueryBuilders;
    import org.elasticsearch.search.builder.SearchSourceBuilder;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.io.IOException;
    
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class ServiceTest1 { 
    
        @Autowired
        private ESClient esClient;
    
        @Test
        public void myTest2() throws IOException { 
           //获取esclient
            RestHighLevelClient restHighLevelClient = esClient.restHighLevelClient;
            SearchRequest searchRequest = new SearchRequest("indexTest");
            
            SearchSourceBuilder builder = new SearchSourceBuilder();
            builder.query(QueryBuilders.matchAllQuery());
            
            searchRequest.source(builder);
    
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(JSONObject.toJSONString(searchResponse));
        }
    }

然后在SpringBoot启动过程中就会建立ES连接

2021-06-08 14:37:05.496  INFO 22736 --- [           main] c.q.m.config.dbconnected.ESClient        : ******************单机ES连接开始****************
    2021-06-08 14:37:05.499  INFO 22736 --- [           main] c.q.m.config.dbconnected.ESClient        : 连接单机ES: 11.14.11.72:9200
    2021-06-08 14:37:08.312  INFO 22736 --- [           main] c.q.m.config.dbconnected.ESClient        : ******************单机ES连接结束****************