# 通过RestHighLevelClient接入集群 在使用Elasticsearch进行数据查询和管理时，如果直接调用API，会因为API的复杂性和多样性导致开发效率低下。CSS服务的Elasticsearch集群支持通过High Level REST Client进行数据查询和管理。High Level REST Client对Elasticsearch的API进行了封装，用户只需要构造对应的结构即可对Elasticsearch集群进行访问，简化了开发流程。REST Client的详细使用指导请参见[Java High Level REST Client](https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high.html)。 #### 前提条件 - Elasticsearch集群处于可用状态。 - 运行Java代码的服务器与Elasticsearch集群之间网络互通。 - 根据集群的网络配置方式，获取集群的访问地址，操作指导请参见[获取集群访问地址](https://support.huaweicloud.com/usermanual-css/css_01_0190.html#ZH-CN_TOPIC_0000001975823337__section855085010198)。 - 确认服务器已安装Java，要求JDK版本为1.8及以上，JDK1.8官网下载地址：[Java Downloads](https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html)。 - 确认High Level REST Client版本。CSS服务支持使用高于Elasticsearch集群版本的Java客户端连接Elasticsearch集群，但是为确保更好的兼容性，建议使用与Elasticsearch集群同版本的Java客户端连接Elasticsearch集群。 当使用比Elasticsearch集群更高版本的High Level REST Client，并且某些请求存在兼容性问题时，可以使用"RestHighLevelClient.getLowLevelClient()"方式直接获取Low Level REST Client，实现自定义的Elasticsearch请求。操作指导请参见[通过RestLowLevelClient接入集群](https://support.huaweicloud.com/usermanual-css/css_01_0066.html)。 #### 引入依赖 在运行Java代码的服务器中引入Java依赖，有如下两种方式： - Maven方式： 使用时需要将7.10.2替换为实际Java客户端的版本号。 ``` org.elasticsearch.client elasticsearch-rest-high-level-client 7.10.2 org.elasticsearch elasticsearch 7.10.2 ``` - Gradle方式： 使用时需要将7.10.2替换为实际Java客户端的版本号。 ``` compile group: 'org.elasticsearch.client', name: 'elasticsearch-rest-high-level-client', version: '7.10.2' ``` #### 接入集群 当使用Java客户端库访问不同安全类型的Elasticsearch集群时，示例代码会有差异，请根据实际业务场景选择参考文档。 表1接入集群的场景介绍 | Elasticsearch集群安全类型 | 是否加载安全证书 | 参考文档 | |:---|:---|:---| | 非安全模式 | - | [RestHighLevelClient连接非安全集群] | | 安全模式+HTTP协议 安全模式+HTTPS协议 | 否 | [RestHighLevelClient连接安全集群（无证书）] | | 安全模式+HTTPS协议 | 是 | [RestHighLevelClient连接安全集群（加载证书）] | #### RestHighLevelClient连接非安全集群 使用Elasticsearch RestHighLevelClient连接非安全模式的Elasticsearch集群，并查询索引"test"是否存在。代码示例如下： ``` import org.apache.http.HttpHost; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.client.indices.GetIndexRequest; import java.io.IOException; import java.util.Arrays; import java.util.List; /** * Rest High Level Client客户端连接非安全集群 */ public class Main { public static void main(String[] args) throws IOException { List host = Arrays.asList("{集群访问地址}"); RestClientBuilder builder = RestClient.builder(constructHttpHosts(host, 9200, "http")); final RestHighLevelClient client = new RestHighLevelClient(builder); GetIndexRequest indexRequest = new GetIndexRequest("test"); boolean exists = client.indices().exists(indexRequest, RequestOptions.DEFAULT); System.out.println(exists); client.close(); } /** * constructHttpHosts函数转换host集群节点IP列表。 */ public static HttpHost[] constructHttpHosts(List host, int port, String protocol) { return host.stream().map(p -> new HttpHost(p, port, protocol)).toArray(HttpHost[]::new); } } ``` 该示例代码为判断集群是否存在test索引，当返回"true"（索引存在）或"false"（索引不存在）时，表示正常返回查询结果，集群连接成功。 #### RestHighLevelClient连接安全集群（无证书） 使用Elasticsearch RestHighLevelClient在不加载安全证书的情况下连接安全模式的Elasticsearch集群（支持HTTP协议或HTTPS协议），查询索引"test"是否存在。代码示例如下： ``` 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.nio.conn.ssl.SSLIOSessionStrategy; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest; import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.client.indices.GetIndexRequest; import java.io.IOException; import java.security.KeyManagementException; import java.security.NoSuchAlgorithmException; import java.security.SecureRandom; import java.security.cert.CertificateException; import java.security.cert.X509Certificate; import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManager; import javax.net.ssl.X509TrustManager; /** * Rest High Level Client客户端连接安全集群（不使用证书） */ public class Main { private static final Logger logger = LogManager.getLogger(Main.class); /** * 创建客户端的类，定义create函数用于创建客户端 */ public static RestHighLevelClient create(List host, int port, String protocol, int connectTimeout, int connectionRequestTimeout, int socketTimeout, String username, String password) throws IOException { RestClientBuilder builder = RestClient.builder(constructHttpHosts(host, port, protocol)) .setRequestConfigCallback(requestConfig -> requestConfig.setConnectTimeout(connectTimeout) .setConnectionRequestTimeout(connectionRequestTimeout) .setSocketTimeout(socketTimeout)) .setHttpClientConfigCallback(httpClientBuilder -> { // enable user authentication final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password)); httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); // set keepalive httpClientBuilder.setKeepAliveStrategy(((httpResponse, httpContext) -> TimeUnit.MINUTES.toMinutes(10))); // enable SSL / TLS SSLContext sc = null; try { sc = SSLContext.getInstance("SSL"); sc.init(null, trustAllCerts, new SecureRandom()); } catch (KeyManagementException | NoSuchAlgorithmException e) { e.printStackTrace(); } SSLIOSessionStrategy sslStrategy = new SSLIOSessionStrategy(sc, new NoopHostnameVerifier()); httpClientBuilder.setSSLStrategy(sslStrategy); return httpClientBuilder; }); final RestHighLevelClient client = new RestHighLevelClient(builder); logger.info("es rest client build success {} ", client); ClusterHealthRequest request = new ClusterHealthRequest(); ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT); System.out.println("es rest client health response {} " + response); return client; } /** * constructHttpHosts函数转换host集群节点IP列表 */ public static HttpHost[] constructHttpHosts(List host, int port, String protocol) { return host.stream().map(p -> new HttpHost(p, port, protocol)).toArray(HttpHost[]::new); } /** * trustAllCerts忽略证书配置 */ public static TrustManager[] trustAllCerts = new TrustManager[] { new X509TrustManager() { @Override public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public X509Certificate[] getAcceptedIssuers() { return null; } } }; /** * main函数参考如下，调用上面的create函数创建客户端，查询“test”索引是否存在 */ public static void main(String[] args) throws IOException { RestHighLevelClient client = create(Arrays.asList("{host}"), 9200, "https", 1000, 1000, 1000, "username", "password"); GetIndexRequest indexRequest = new GetIndexRequest("test"); boolean exists = client.indices().exists(indexRequest, RequestOptions.DEFAULT); System.out.println(exists); client.close(); } } ``` 表2函数中的变量说明 | 参数 | 描述 | |:---|:---| | host | 集群访问地址，当存在多个IP地址时，中间用","隔开。 | | port | 集群的连接端口，默认是"9200"。 | | protocol | 连接协议，"http"或者"https"，根据集群实际情况填写。 | | connectTimeout | 连接超时时间（毫秒）。 | | connectionRequestTimeout | socket连接请求超时时间（毫秒）。 | | socketTimeout | socket请求超时时间（毫秒）。 | | username | 访问集群的用户名。 | | password | 用户名对应的密码。 | 该示例代码为判断集群是否存在test索引，当返回"true"（索引存在）或"false"（索引不存在）时，表示正常返回查询结果，集群连接成功。 #### RestHighLevelClient连接安全集群（加载证书） 使用Elasticsearch RestHighLevelClient在加载安全证书的情况下连接安全模式+HTTPS协议的Elasticsearch集群，查询索引"test"是否存在。代码示例如下： ![](https://support.huaweicloud.com/usermanual-css/public_sys-resources/caution_3.0-zh-cn.png) 连接安全模式+HTTPS协议的Elasticsearch集群需要先提前准备好安全证书，操作指导请参见[获取并上传安全证书]。 ``` 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.nio.conn.ssl.SSLIOSessionStrategy; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest; import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.client.indices.GetIndexRequest; import java.io.File; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.security.KeyStore; import java.security.SecureRandom; import java.security.cert.CertificateException; import java.security.cert.X509Certificate; import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManager; import javax.net.ssl.TrustManagerFactory; import javax.net.ssl.X509TrustManager; /** * Rest High Level Client客户端连接安全集群（使用https证书） */ public class Main { public static RestHighLevelClient create(List host, int port, String protocol, int connectTimeout, int connectionRequestTimeout, int socketTimeout, String username, String password, String certFilePath, String certPassword) throws IOException { RestClientBuilder builder = RestClient.builder(constructHttpHosts(host, port, protocol)) .setRequestConfigCallback(requestConfig -> requestConfig.setConnectTimeout(connectTimeout) .setConnectionRequestTimeout(connectionRequestTimeout) .setSocketTimeout(socketTimeout)) .setHttpClientConfigCallback(httpClientBuilder -> { // enable user authentication final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password)); httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); // set keepalive httpClientBuilder.setKeepAliveStrategy(((httpResponse, httpContext) -> TimeUnit.MINUTES.toMinutes(10))); // enable SSL / TLS SSLContext sc = null; try { TrustManager[] tm = {new MyX509TrustManager(certFilePath, certPassword)}; sc = SSLContext.getInstance("SSL", "SunJSSE"); //也可以使用SSLContext sslContext = SSLContext.getInstance("TLSv1.2"); sc.init(null, tm, new SecureRandom()); } catch (Exception e) { e.printStackTrace(); } SSLIOSessionStrategy sslStrategy = new SSLIOSessionStrategy(sc, new NoopHostnameVerifier()); httpClientBuilder.setSSLStrategy(sslStrategy); return httpClientBuilder; }); final RestHighLevelClient client = new RestHighLevelClient(builder); logger.info("es rest client build success {} ", client); ClusterHealthRequest request = new ClusterHealthRequest(); ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT); logger.info("es rest client health response {} ", response); return client; } /** * constructHttpHosts函数转换host集群节点IP列表。 */ public static HttpHost[] constructHttpHosts(List host, int port, String protocol) { return host.stream().map(p -> new HttpHost(p, port, protocol)).toArray(HttpHost[]::new); } private static final Logger logger = LogManager.getLogger(Main.class); public static class MyX509TrustManager implements X509TrustManager { X509TrustManager sunJSSEX509TrustManager; MyX509TrustManager(String certFilePath, String certPassword) throws Exception { File file = new File(certFilePath); if (!file.isFile()) { throw new Exception("Wrong Certification Path"); } System.out.println("Loading KeyStore " + file + "..."); InputStream in = new FileInputStream(file); KeyStore ks = KeyStore.getInstance("JKS"); ks.load(in, certPassword.toCharArray()); TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509", "SunJSSE"); tmf.init(ks); TrustManager[] tms = tmf.getTrustManagers(); for (TrustManager tm : tms) { if (tm instanceof X509TrustManager) { sunJSSEX509TrustManager = (X509TrustManager) tm; return; } } throw new Exception("Couldn't initialize"); } @Override public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } } /** * main函数参考如下，调用上面的create函数创建客户端，查询“test”索引是否存在。 */ public static void main(String[] args) throws IOException { RestHighLevelClient client = create(Arrays.asList({host}), 9200, "https", 1000, 1000, 1000, "username", "password", "certFilePath", "certPassword"); GetIndexRequest indexRequest = new GetIndexRequest("test"); boolean exists = client.indices().exists(indexRequest, RequestOptions.DEFAULT); System.out.println(exists); client.close(); } } ``` 表3函数中的参数说明 | 参数 | 描述 | |:---|:---| | host | 集群访问地址，当存在多个IP地址时，中间用","隔开。 | | port | 集群的连接端口，默认是"9200"。 | | protocol | 连接协议，此处填写"https"。 | | connectTimeout | 连接超时时间（毫秒）。 | | connectionRequestTimeout | socket连接请求超时时间（毫秒）。 | | socketTimeout | socket请求超时时间（毫秒）。 | | username | 访问集群的用户名。 | | password | 用户名对应的密码。 | | certFilePath | 安全证书路径。 | | certPassword | 安全证书密码。 | 该示例代码为判断集群是否存在test索引，当返回"true"（索引存在）或"false"（索引不存在）时，表示正常返回查询结果，集群连接成功。 #### 获取并上传安全证书 当接入安全模式+HTTPS协议的Elasticsearch集群时，如需加载安全证书则可以参考如下步骤获取安全证书，并上传至客户端。 1. 获取安全证书（CloudSearchService.cer）。 1. 登录[云搜索服务管理控制台](https://console.huaweicloud.com/elasticsearch/)。 2. 在左侧导航栏，选择"集群管理 \> Elasticsearch"。 3. 在集群列表，单击目标集群名称，进入集群详情页。 4. 选择"概览"页签，在"网络信息"下方，单击"HTTPS访问"的"下载证书"获取安全证书。 图1下载安全证书 ![](https://support.huaweicloud.com/usermanual-css/images/zh-cn_image_0000002412660693.png) 2. 转换安全证书（CloudSearchService.cer）。将下载的安全证书上传到客户端机器上，使用keytool工具将".cer"证书转换成Java可以读取的".jks"证书格式。 - 在Linux系统中，执行如下命令转换证书。 ``` keytool -import -alias newname -keystore ./truststore.jks -file ./CloudSearchService.cer ``` - 在Windows系统中，执行如下命令转换证书。 ``` keytool -import -alias newname -keystore .\truststore.jks -file .\CloudSearchService.cer ``` 其中，*newname*是由用户自定义的证书名称。 该命令执行后，会提示设置证书密码，并确认密码。请保存该密码，后续接入集群会使用。