ElasticSearch公共方法封装

业务场景

1、RestClientBuilder初始化（同时支持单机与集群）
2、发送ES查询请求公共方法封装（支持sql、kql、代理访问、集群访问、鉴权支持）
3、判断ES索引是否存在（/_cat/indices/${indexName}）
4、判断ES索引别名是否存在 （/_cat/aliases/${indexName}）
5、判断ES索引指定字段/属性是否存在（这里字段支持多级，如：logObj.id）
6、判断ES索引指定字段/属性的类型（字段支持多级）
7、阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）
8、创建索引别名（可用于支持sql查询，索引名中有特殊字符不能用作表名，可通过创建别名来解决）
9、索引别名创建结果解析（ 判断acknowledged）
10、KQL查询ES（Kibana语法查询ElasticSearch）
11、SQL查询ES（标准SQL语法查询ElasticSearch）
12、Java在本地通过代理访问ES（可用于解决网络不能直接的问题）
13、Java 客户端访问ES集群（同时支持单机与集群）
14、Java ES客户端鉴权（安全需求）

软件环境

ElasticSearch 7.17.23 下载地址

ElasticSearch 7.17.23 帮助文档

ElasticSearch 8.17.2 下载地址

ElasticSearch 8.17.2 帮助文档

说明：当前例子中用的7，理论上8也通用

Kibana查询效果

KQL查询ES

SQL查询ES

下面讲java代码实现

Java类方法详解

1、RestClientBuilder初始化

同时支持单机与集群

/*** RestClientBuilder 初始化** @param host 同时支持单机与集群*        单机：host和port各司其职*        集群时：port参数无效，host中包含IP和PORT，多个实例用逗号分隔**    eg：*         10.***.6.247*     或： host:10.***.6.247:9200,10.***.6.34:9200,10.***.6.120:9200* @param port* @return*/private RestClientBuilder buildClient(String host, Integer port){RestClientBuilder restClientBuilder = null;if(host.indexOf(",")==-1){// 单机 host:10.***.6.247, 只有单机会使用port参数restClientBuilder = RestClient.builder(new HttpHost( host, port, "http" ) );}else{// 集群 host:10.***.6.247:9200,10.***.6.34:9200,10.***.6.120:9200,10.***.6.9:9200,10.***.6.183:9200String[] hostArr = host.split("\\,");HttpHost[] httpHosts = new HttpHost[hostArr.length];for( int i=0; i<hostArr.length; i++ ){String[] addrs = hostArr[i].split("\\:");HttpHost httpHost = new HttpHost( addrs[0],  Integer.valueOf(addrs[1]), "http" );httpHosts[i] = httpHost;}restClientBuilder = RestClient.builder( httpHosts );}return restClientBuilder;}

2、发送ES查询请求公共方法

SQL支持
KQL支持
支持代理访问支持
鉴权支持

/*** 发送ES 查询请求** @param host* @param port* @param username* @param password* @param method* @param endpoint ES接口*      eg：*        1、创建别名： "/_aliases"*        2、判断索引是否存在： "/_cat/indices/myIndexName"*        2、判断索引别名是否存在： "/_cat/aliases/indexName"** @param jsonEntity 查询语句*      eg：*      1、为索引创建别名（可用于支持sql查询，如果使用sql查询时原索引名中有特殊字符不能用作表名，可通过创建别名来解决）*         String kqlJson = "{\"actions\" : [{ \"add\" : { \"index\" : \""+idxName+"\",\"alias\" : \""+idxAliases+"\" } }]}  ";*         String kqlJson= "";*      2、String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\", " +*                 "\"params\": ["+cStart+","+cEnd+"]," +*                 " \"fetch_size\": 65536 }";**  实际业务场景举例：*  POST _sql?format=txt* {*   "query": "SELECT tags.svc_code, sum(iif(tags.response_code.keyword='0000',1,0)) as success_count, count(metric) as total*               FROM order_service_*****              where create_time between '2025-02-27T11:00:00+0800' and '2025-02-27T13:59:00+0800'*              group by tags.svc_code having count(metric)>=50*              order by 3 desc",*   "fetch_size": 65536* }** @return* @throws IOException*/public String request(String host, Integer port,String username, String password,String method, String endpoint, String jsonEntity) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());/* *******  能直连ES的不需要；如果本地不能直连ES的则加上，IP根据实际调整**  */if("dev".equals(profile) && host.indexOf("10.***.137")==0 ) {httpClientBuilder.setProxy(new HttpHost("10.***.248.54", 8443, "http")  //设置代理服务);}else if("dev".equals(profile)  && host.indexOf("10.***.6")==0  ){httpClientBuilder.setProxy(new HttpHost( "192.***.66.30", 8443,"http")  //设置代理服务);}return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request(method, endpoint );request.setJsonEntity( jsonEntity );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);return EntityUtils.toString(entity);}

3、判断ES索引是否存在

 /***  判断索引名是否存在** @param host* @param port* @param username* @param password* @param indexName 索引名* @return* @throws IOException*/public boolean isExistsIndex( String host, Integer port,String username, String password,String indexName ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/_cat/indices/"+indexName );try{Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();entity = new BufferedHttpEntity(entity);if (!StringUtils.hasLength(EntityUtils.toString(entity))) {System.out.println("Index exists.");return true;} else {System.out.println("Index does not exist.");return false;}}catch (Exception e){//如果不存在会报404的错误，返回false创建别名return false;}finally {restClient.close();}}

4、判断ES索引别名是否存在

/*** 判断索引别名是否存在** @param host* @param port* @param username* @param password* @param indexName* @return* @throws IOException*/public boolean isExistsAliases( String host, Integer port,String username, String password,String indexName ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/_cat/aliases/"+indexName );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);if (!StringUtils.isEmpty(EntityUtils.toString(entity))) {System.out.println("Index alias exists.");return true;} else {System.out.println("Index alias does not exist.");return false;}}

5、获取ES索引指定字段/属性是否存在

这里字段支持多级，如：logObj.id

/*** 判断索引 某个字段/属性是否存在*   说明： 这里字段支持多级，如：logObj.id** @param host* @param port* @param username* @param password* @param indexName* @param property  eg：id、  logObj.id* @return* @throws IOException*/public boolean isExistsProperty( String host, Integer port,String username, String password,String indexName, String property ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/"+indexName+"/_mapping" );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);JSONObject obj = JSONObject.parseObject(EntityUtils.toString(entity));JSONObject properties = obj.getJSONObject( obj.keySet().iterator().next() ).getJSONObject("mappings" ).getJSONObject("properties" );String[] arr = property.split("\\.");for( int i=0; i<arr.length; i++ ){if(i==arr.length-1){}else{if(properties.containsKey( arr[i] )){properties = properties.getJSONObject(arr[i]).getJSONObject("properties" );}else{return false;}}}boolean bool =  properties.containsKey( arr[arr.length-1] );log.info("property:{} , isExist:{}", property, bool );return bool;}

6、获取ES索引指定字段/属性的类型

同一个索引的同一字段，不同时间的数据类型可能不一样，从而影响sql语句的写法（sql语法不一样），所以个别场景要做判断

 /*** 判断索引某个字段/属性的类型*   说明： 这里字段支持多级，如：logObj.id** @param host* @param port* @param username* @param password* @param indexName* @param property  eg：id、  logObj.id* @return* @throws IOException*/public String getIndexPropertyType( String host, Integer port,String username, String password,String indexName, String property ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/"+indexName+"/_mapping" );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);JSONObject obj = JSONObject.parseObject(EntityUtils.toString(entity));JSONObject properties = obj.getJSONObject( obj.keySet().iterator().next() ).getJSONObject("mappings" ).getJSONObject("properties" );String[] arr = property.split("\\.");for( int i=0; i<arr.length; i++ ){if(i==arr.length-1){properties = properties.getJSONObject(arr[i]);}else{if(properties.containsKey( arr[i] )){properties = properties.getJSONObject(arr[i]).getJSONObject("properties" );}else{log.error( "判断字段类型时，发现字段不存在！property:{}", property );throw new RuntimeException( "判断字段类型时，发现字段不存在！property:"+property );}}}String type = properties.getString("type" );log.info("property:{} , type:{}", property, type );return type;}

7、阻塞线程直至索引就绪

为了应对跨日时索引名短时间可能不存在的问题（不处理可能导致程序报错）

/*** 阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）** @param host* @param port* @param username* @param password* @param indexName* @throws IOException* @throws InterruptedException*/@Overridepublic void waitIndexReady( String host, Integer port,String username, String password,String indexName ) throws IOException, InterruptedException {// 循环一次是10s，6次是1分钟，60次是10分钟for( int i=0; i<60 && !this.isExistsIndex(host,   port,username,   password,indexName) ; i++ ){// 共循环10分钟Thread.sleep( 10*1000 );}}

8、创建索引别名

有时候索引名带特殊字符，是sql的关键字，所以创建别名可供sql查询用作表名

/*** 创建索引别名** @param username* @param password* @param host* @param port* @param idxName 索引名* @param idxAliases 索引别名* @throws ParseException* @return*/public void createAliases(  String username, String password,String host, int port,String idxName, String idxAliases ) throws ParseException {log.info("创建索引别名 ：{}:{}, {}", host, port, idxAliases );String method = "POST";String endpoint = "/_aliases";// 为索引创建别名，用于支持sql查询String kqlJson = "{\"actions\" : [{ \"add\" : { \"index\" : \""+idxName+"\",\"alias\" : \""+idxAliases+"\" } }]}  ";try{String body =null;body = this.request(   host,   port,username,   password,method,   endpoint,   kqlJson );boolean acknowledged = parseCreateAliasesResult( body );if(acknowledged){log.info("别名创建成功");}else{log.error("别名创建失败");log.error("别名创建失败 kqlJson:{}", kqlJson );throw new RuntimeException("索引别名创建失败！");}}catch (Exception ex){log.error("创建索引别名异常！message:{}",ex.getLocalizedMessage());log.error("创建索引别名异常 kqlJson:{}", kqlJson );ex.printStackTrace();}}

9、索引别名创建结果解析

判断创建时返回值中的acknowledged属性值

    /***  判断body 中是否包含 acknowledged** @param body* @return*/private boolean parseCreateAliasesResult(String body  ){JSONObject json = JSONObject.parseObject(body);if(json.containsKey("acknowledged") && json.getBoolean("acknowledged")){return true;}return false;}

完整代码实现

完整maven pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>person.brickman</groupId><artifactId>javaProject</artifactId><version>1.0-SNAPSHOT</version><!-- 统一管理jar包版本 --><properties><java.version>17</java.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>17</maven.compiler.source><maven.compiler.target>17</maven.compiler.target><spring-boot.version>2.7.18</spring-boot.version><spring-cloud.version>2021.0.9</spring-cloud.version><spring-cloud-starter-bootstrap.version>3.1.9</spring-cloud-starter-bootstrap.version><elasticsearch-client.version>7.17.23</elasticsearch-client.version><commons-lang3.version>3.14.0</commons-lang3.version><fastjson2.version>2.0.53</fastjson2.version><lombok.version>1.18.28</lombok.version><testng.version>6.14.3</testng.version></properties><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bootstrap</artifactId><version>${spring-cloud-starter-bootstrap.version}</version></dependency><dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId><version>${elasticsearch-client.version}</version></dependency><dependency><groupId>org.apache.httpcomponents.client5</groupId><artifactId>httpclient5</artifactId><version>5.3</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>${commons-lang3.version}</version></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>${fastjson2.version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency><!-- 测试相关 默认集成junit5 作者用testng，所以排除掉junit5  --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter</artifactId></exclusion><exclusion><groupId>org.mockito</groupId><artifactId>mockito-junit-jupiter</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.testng</groupId><artifactId>testng</artifactId><version>${testng.version}</version><scope>test</scope></dependency></dependencies><dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>${spring-boot.version}</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>${spring-cloud.version}</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement>
</project>

完整ES公共组件类

package person.brickman.es;import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpEntity;
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.entity.BufferedHttpEntity;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;import java.io.IOException;
import java.text.ParseException;/*** @Description: ES公共方法类（公共组件）*   1、RestClientBuilder初始化（同时支持单机与集群）*   2、发送ES查询请求公共方法封装（支持sql、kql、代理访问、集群访问、鉴权）*   3、判断ES索引是否存在（/_cat/indices/${indexName}）*   4、判断ES索引别名是否存在 （/_cat/aliases/${indexName}）*   5、判断ES索引指定字段/属性是否存在（这里字段支持多级，如：logObj.id）*   6、判断ES索引指定字段/属性的类型（字段支持多级）*   7、阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）*   8、创建索引别名（可用于支持sql查询，索引名中有特殊字符不能用作表名，可通过创建别名来解决）*   9、索引别名创建结果解析（ 判断acknowledged）** @Author: brickman* @CreateDate: 2025/2/20 23:46* @Version: 1.0*/
@Slf4j
@Service
public class ESRestClientService {@Value("${spring.profiles.active}")private String profile;/*** RestClientBuilder 初始化** @param host 同时支持单机与集群*        单机：host和port各司其职*        集群时：port参数无效，host中包含IP和PORT，多个实例用逗号分隔**    eg：*         10.***.6.247*     或： host:10.***.6.247:9200,10.***.6.34:9200,10.***.6.120:9200* @param port* @return*/private RestClientBuilder buildClient(String host, Integer port){RestClientBuilder restClientBuilder = null;if(host.indexOf(",")==-1){// 单机 host:10.***.6.247, 只有单机会使用port参数restClientBuilder = RestClient.builder(new HttpHost( host, port, "http" ) );}else{// 集群 host:10.***.6.247:9200,10.***.6.34:9200,10.***.6.120:9200,10.***.6.9:9200,10.***.6.183:9200String[] hostArr = host.split("\\,");HttpHost[] httpHosts = new HttpHost[hostArr.length];for( int i=0; i<hostArr.length; i++ ){String[] addrs = hostArr[i].split("\\:");HttpHost httpHost = new HttpHost( addrs[0],  Integer.valueOf(addrs[1]), "http" );httpHosts[i] = httpHost;}restClientBuilder = RestClient.builder( httpHosts );}return restClientBuilder;}/*** 发送ES 查询请求，包含sql、kql、代理访问、鉴权支持** @param host 同时支持单机与集群*             单机：host和port各司其职*             集群时：port参数无效，host中包含IP和PORT，多个实例用逗号分隔* @param port* @param username* @param password* @param method* @param endpoint ES接口*      eg：*        1、创建别名： "/_aliases"*        2、判断索引是否存在： "/_cat/indices/myIndexName"*        2、判断索引别名是否存在： "/_cat/aliases/indexName"** @param jsonEntity 查询语句*      eg：*      1、为索引创建别名（可用于支持sql查询，如果使用sql查询时原索引名中有特殊字符不能用作表名，可通过创建别名来解决）*         String kqlJson = "{\"actions\" : [{ \"add\" : { \"index\" : \""+idxName+"\",\"alias\" : \""+idxAliases+"\" } }]}  ";*         String kqlJson= "";*      2、String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\", " +*                 "\"params\": ["+cStart+","+cEnd+"]," +*                 " \"fetch_size\": 65536 }";**  实际业务场景举例：*  POST _sql?format=txt* {*   "query": "SELECT tags.svc_code, sum(iif(tags.response_code.keyword='0000',1,0)) as success_count, count(metric) as total*               FROM order_service_*****              where create_time between '2025-02-27T11:00:00+0800' and '2025-02-27T13:59:00+0800'*              group by tags.svc_code having count(metric)>=50*              order by 3 desc",*   "fetch_size": 65536* }** @return* @throws IOException*/public String request(String host, Integer port,String username, String password,String method, String endpoint, String jsonEntity) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());/* *******  本地能直连ES的不需要；如果本地不能直连ES的则加上，IP根据实际调整**  */if("dev".equals(profile) && host.indexOf("10.***.137")==0 ) {httpClientBuilder.setProxy(new HttpHost("10.***.248.54", 8443, "http")  //设置代理服务);}else if("dev".equals(profile)  && host.indexOf("10.***.6")==0  ){httpClientBuilder.setProxy(new HttpHost( "192.***.66.30", 8443,"http")  //设置代理服务);}return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request(method, endpoint );request.setJsonEntity( jsonEntity );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);return EntityUtils.toString(entity);}/***  判断索引名是否存在** @param host 同时支持单机与集群*             单机：host和port各司其职*             集群时：port参数无效，host中包含IP和PORT，多个实例用逗号分隔* @param port* @param username* @param password* @param indexName 索引名* @return* @throws IOException*/public boolean isExistsIndex( String host, Integer port,String username, String password,String indexName ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/_cat/indices/"+indexName );try{Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();entity = new BufferedHttpEntity(entity);if (!StringUtils.hasLength(EntityUtils.toString(entity))) {System.out.println("Index exists.");return true;} else {System.out.println("Index does not exist.");return false;}}catch (Exception e){//如果不存在会报404的错误，返回false创建别名return false;}finally {restClient.close();}}/*** 判断索引别名是否存在** @param host* @param port* @param username* @param password* @param indexName* @return* @throws IOException*/public boolean isExistsAliases( String host, Integer port,String username, String password,String indexName ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/_cat/aliases/"+indexName );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);if (!StringUtils.isEmpty(EntityUtils.toString(entity))) {System.out.println("Index alias exists.");return true;} else {System.out.println("Index alias does not exist.");return false;}}/*** 判断索引 某个字段/属性是否存在*   说明： 这里字段支持多级，如：logObj.id** @param host* @param port* @param username* @param password* @param indexName* @param property  eg：id、  logObj.id* @return* @throws IOException*/public boolean isExistsProperty( String host, Integer port,String username, String password,String indexName, String property ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/"+indexName+"/_mapping" );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);JSONObject obj = JSONObject.parseObject(EntityUtils.toString(entity));JSONObject properties = obj.getJSONObject( obj.keySet().iterator().next() ).getJSONObject("mappings" ).getJSONObject("properties" );String[] arr = property.split("\\.");for( int i=0; i<arr.length; i++ ){if(i==arr.length-1){}else{if(properties.containsKey( arr[i] )){properties = properties.getJSONObject(arr[i]).getJSONObject("properties" );}else{return false;}}}boolean bool =  properties.containsKey( arr[arr.length-1] );log.info("property:{} , isExist:{}", property, bool );return bool;}/*** 判断索引某个字段/属性的类型*   说明： 这里字段支持多级，如：logObj.id*   同一个索引的同一字段，不同时间的数据类型可能不一样，从而影响sql语句的写法（sql语法不一样），所以个别场景要做判断** @param host* @param port* @param username* @param password* @param indexName* @param property  eg：id、  logObj.id* @return* @throws IOException*/public String getIndexPropertyType( String host, Integer port,String username, String password,String indexName, String property ) throws IOException {RestClientBuilder restClientBuilder = buildClient( host, port );if(!StringUtils.isEmpty(username)){final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);//线程设置httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());return httpClientBuilder;}});}RestClient restClient = restClientBuilder.build();Request request = new Request("GET", "/"+indexName+"/_mapping" );Response response = restClient.performRequest(request);HttpEntity entity=response.getEntity();restClient.close();entity = new BufferedHttpEntity(entity);JSONObject obj = JSONObject.parseObject(EntityUtils.toString(entity));JSONObject properties = obj.getJSONObject( obj.keySet().iterator().next() ).getJSONObject("mappings" ).getJSONObject("properties" );String[] arr = property.split("\\.");for( int i=0; i<arr.length; i++ ){if(i==arr.length-1){properties = properties.getJSONObject(arr[i]);}else{if(properties.containsKey( arr[i] )){properties = properties.getJSONObject(arr[i]).getJSONObject("properties" );}else{log.error( "判断字段类型时，发现字段不存在！property:{}", property );throw new RuntimeException( "判断字段类型时，发现字段不存在！property:"+property );}}}String type = properties.getString("type" );log.info("property:{} , type:{}", property, type );return type;}/*** 阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）** @param host* @param port* @param username* @param password* @param indexName* @throws IOException* @throws InterruptedException*/public void waitIndexReady( String host, Integer port,String username, String password,String indexName ) throws IOException, InterruptedException {// 循环一次是10s，6次是1分钟，60次是10分钟for( int i=0; i<60 && !this.isExistsIndex(host,   port,username,   password,indexName) ; i++ ){// 共循环10分钟Thread.sleep( 10*1000 );}}/*** 创建索引别名*   有时候索引名带特殊字符，是sql的关键字，所以创建别名可供sql查询用作表名** @param username* @param password* @param host* @param port* @param idxName 索引名* @param idxAliases 索引别名* @throws ParseException* @return*/public void createAliases(  String username, String password,String host, int port,String idxName, String idxAliases ) throws ParseException {log.info("创建索引别名 ：{}:{}, {}", host, port, idxAliases );String method = "POST";String endpoint = "/_aliases";// 为索引创建别名，用于支持sql查询String kqlJson = "{\"actions\" : [{ \"add\" : { \"index\" : \""+idxName+"\",\"alias\" : \""+idxAliases+"\" } }]}  ";try{String body =null;body = this.request(   host,   port,username,   password,method,   endpoint,   kqlJson );boolean acknowledged = parseCreateAliasesResult( body );if(acknowledged){log.info("别名创建成功");}else{log.error("别名创建失败");log.error("别名创建失败 kqlJson:{}", kqlJson );throw new RuntimeException("索引别名创建失败！");}}catch (Exception ex){log.error("创建索引别名异常！message:{}",ex.getLocalizedMessage());log.error("创建索引别名异常 kqlJson:{}", kqlJson );ex.printStackTrace();}}/***  判断body 中是否包含 acknowledged** @param body* @return*/private boolean parseCreateAliasesResult(String body  ){JSONObject json = JSONObject.parseObject(body);if(json.containsKey("acknowledged") && json.getBoolean("acknowledged")){return true;}return false;}
}

单元测试方法详解

作者用的testng

1、执行检索

/*** 执行检索（这里使用sql查询近一分钟的数据）** @throws IOException* @throws ParseException*/@Test(groups = "hlog", enabled = true )public void testRequest() throws IOException, ParseException {String method = "POST";String endpoint = "/_sql?format=json";RangeTimeUtils rangeTimeUtils = new RangeTimeUtils();
//        String time = "2025-02-21 17:48:00";String time = TimeUtils.calcWholeMinute();rangeTimeUtils.calcAllByTimeAndPeriod(  time, 1);//  "yyyy-MM-dd HH:mm:ss"  --> "yyyy-MM-dd'T'HH:mm"  根据实际时间字段格式调整SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm" );Long cStart = rangeTimeUtils.getStartDate().getTime();Long cEnd = rangeTimeUtils.getStartDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );String idxAliases = "hlog_crm3_comm_undef_"+sdf_target.format(rangeTimeUtils.getEndDate());// 生产环境String sqlQuery2 = "SELECT logObj.province,server, logObj.app, logObj.node_ip, " +"SUBSTRING(logObj.uri,0,30) , count(id) as num, avg(logObj.costTime) latency ,   " +"sum(iif(logObj.code=200,1,0))   as success_count " +" FROM " + idxAliases + " "+
//                "where cTime between '"+cStart+":00+0800' and '"+cEnd+":00+0800' " +" where cTime between "+cStart+" and "+cEnd+" " +"group by logObj.province,server, logObj.app, logObj.node_ip,  SUBSTRING(logObj.uri,0,30)  "; // 将index_name、field_name和value替换为相应的索引名称、字段名和值// max fetch_size is 65536  不同的版本单次查询最大数据限制不一样，这里测试只查 5 条String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\",  \"fetch_size\": 5}";String ret =   service.request(  host,   port,username,   password,method,   endpoint,   jsonEntity );log.debug("ret:{}",ret);}

2、判断索引是否存在

/*** 判断索引是否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsIndex() throws IOException {/*  interf_dand_hig_c_trans_2024_12_03interf_dand_hig_c_trans_2025_02_20interf_dand_comm_undef_2024_12_03interf_dand_comm_undef_2025_02_20*/boolean ret =   service.isExistsIndex( host, port,username,   password,"interf_dand_comm_undef_2025_02_21");log.info("ret:{}",ret);}

3、判断索引别名是否存在

    /*** 判断索引别名是否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsAliases() throws IOException {/*  interf_dand_hig_c_trans_2025_02_21interf_dand_comm_undef_2025_02_21*/boolean ret =   service.isExistsAliases( host, port,username,   password,"interf_dand_comm_undef_2025_02_21");log.info("ret:{}",ret);}

4、判断索引字段时否存在

    /*** 判断索引字段时否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsProperty() throws IOException {// logObj.node_ip     logObj.statusboolean ret =   service.isExistsProperty( host, port,username, password,"interf_dand_comm_undef_2025_02_21", "logObj.status");log.info("ret:{}",ret);}

5、获取索引列类型

    /*** 获取索引列类型** @throws IOException*/@Test( enabled = true )public void testGetIndexPropertyType() throws IOException {/*  interf_dand_hig_c_trans_2025_02_21interf_dand_comm_undef_2025_02_21*/String ret =   service.getIndexPropertyType(   host,   port,username,   password,"interf_dand_comm_undef_2025_02_21","logObj.status"  );log.info("ret:{}",ret);}

6、阻塞线程直至索引就绪

    /*** 阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）** @throws IOException*/@Test(  enabled = true )public void testWaitIndexReady() throws IOException, ParseException, InterruptedException {RangeTimeUtils rangeTimeUtils = new RangeTimeUtils();String time = "2025-02-21 20:00:00";rangeTimeUtils.calcAllByTimeAndPeriod( time, 1 );service.waitIndexReady( host,   port,username,   password,"interf_dand_comm_undef_2025_02_21");}

7、创建索引别名

    /*** 创建索引别名** @throws IOException*/@Test(  enabled = true )public void testCreateAliases() throws IOException, ParseException {RangeTimeUtils rangeTimeUtils = new RangeTimeUtils();String time = "2025-02-21 20:00:00";rangeTimeUtils.calcAllByTimeAndPeriod( time, 1 );service.createAliases( username,   password,host,   port,"interf_dand-comm-undef.2025.02.21","interf_dand_comm_undef_2025_02_21");}

8、SQL查询ES

    /*** sql查询** @throws IOException*/@Test(  enabled = true )public void testSQLRequest() throws IOException, ParseException, InterruptedException {log.info("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口/不区分接口)服务级 ：{}:{}", host, port );RangeTimeUtils rangeTimeUtils = new RangeTimeUtils();String time = "2025-02-21 20:00:00";rangeTimeUtils.calcAllByTimeAndPeriod( time, 1 );List<Object> ret=null;String method = "POST";String endpoint = "/_sql?format=json";//  根据实际时间字段格式调整SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm");long cStart = rangeTimeUtils.getStartDate().getTime();long cEnd = rangeTimeUtils.getEndDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );// interf_dand-comm-undef.2024.01.06String idxName = ESConsts.INTERFLOG_INDEX_NAME_PREFIX + sdf_source.format(rangeTimeUtils.getStartDate());// interf_dand_comm_undef_2024_01_06String idxAliases = ESConsts.INTERFLOG_INDEX_ALIASES_PREFIX + sdf_target.format(rangeTimeUtils.getStartDate());//等待索引就绪service.waitIndexReady(host,   port,username,   password,idxName);// 判断别名是否存在，不存在则创建if(!service.isExistsAliases(  host,   port,username,   password,idxAliases)){service.createAliases(  username,   password,  host,   port,idxName, idxAliases);}/*   logObj.status.keyword（对应String） 、还是 logObj.status（对应int ） 根据类型来  */String statusFieldName = getStatusFieldName(host,   port,username,   password,idxAliases );String sqlQuery2 = "SELECT logObj.province,server, logObj.app, '' node_ip, " +"  '' uri, count(id) as num, avg(logObj.cost) latency ,   " +"sum(iif("+statusFieldName+"=200,1,0))   as success_count " +" FROM " + idxAliases + " " +// 算头不算尾"where cTime >= ? and cTime < ? " +"group by logObj.province,server, logObj.app ";// max fetch_size is 65536 ， 已验证一次查询结果不会超过这个数，故不用做滚动查询String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\", " +"\"params\": ["+cStart+","+cEnd+"]," +" \"fetch_size\": 10 }"; // 65536 -- 不同的版本单次查询最大数据限制不一样，这里测试只查10条String body =null;try{body = service.request(    host,   port,username,   password,method,   endpoint,   jsonEntity );// 组装成实际需要的业务类型集合
//            ret = ESIntfLoad3Parser.parseESresult4SvcLoad( body, IDC, host ,port, rangeTime);}catch (Exception ex){log.error("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口)服务级 指标异常！message:{}",ex.getLocalizedMessage());log.error("sqlQuery2:{}", sqlQuery2 );log.error("jsonEntity:{}", jsonEntity);ex.printStackTrace();}log.debug(" body:{}", body );}

9、KQL查询ES

   /*** KQL查询** @throws IOException*/@Test(  enabled = true )public void testKQLRequest() throws IOException, ParseException, InterruptedException {log.info("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口/不区分接口)服务级 ：{}:{}", host, port );RangeTimeUtils rangeTimeUtils = new RangeTimeUtils();String time = "2025-02-21 20:00:00";rangeTimeUtils.calcAllByTimeAndPeriod( time, 1 );List<Object> ret=null;String method = "GET";String endpoint = "/interf_dand_comm_undef_2025_02_21/_search";SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm");// 这里cStart、cEnd两者相关1分钟，在上面做了初始化long cStart = rangeTimeUtils.getStartDate().getTime();long cEnd = rangeTimeUtils.getEndDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );// interf_dand-comm-undef.2024.01.06String idxName = ESConsts.INTERFLOG_INDEX_NAME_PREFIX + sdf_source.format(rangeTimeUtils.getStartDate());// interf_dand_comm_undef_2024_01_06String idxAliases = ESConsts.INTERFLOG_INDEX_ALIASES_PREFIX + sdf_target.format(rangeTimeUtils.getStartDate());//等待索引就绪service.waitIndexReady(host,   port,username,   password,idxName);// 判断别名是否存在，不存在则创建if(!service.isExistsAliases(  host,   port,username,   password,idxAliases)){service.createAliases(  username,   password,  host,   port,idxName, idxAliases);}String kqlQuery2 = "{ " +"    \"bool\": { " +"      \"must\": [ " +"        { \"match_phrase\": { \"logObj.app\":\"order-service\" } }, " +"        { \"range\": { " +"          \"cTime\": {  " +"            \"gte\": ?, " +"            \"lt\": ? " +"          } " +"        }} " +"      ] " +"    } " +"  }"; // 将index_name、field_name和value替换为相应的索引名称、字段名和值// max fetch_size is 65536 ， 已验证一次查询结果不会超过这个数，故不用做滚动查询String jsonEntity = "{\"query\": \"" + kqlQuery2 + "\", " +"\"params\": ["+cStart+","+cEnd+"]," +" \"size\": 0,\" +\n" +" \"size\": 2 }"; // 65536 -- 不同的版本单次查询最大数据限制不一样, 这里测试只查两条String body =null;try{body = service.request(    host,   port,username,   password,method,   endpoint,   jsonEntity );// 组装成实际需要的业务类型集合
//            ret = ESIntfLoad3Parser.parseESresult4SvcLoad( body, IDC, host ,port, rangeTime);}catch (Exception ex){log.error("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口)服务级 指标异常！message:{}",ex.getLocalizedMessage());log.error("kqlQuery2:{}", kqlQuery2 );log.error("jsonEntity:{}", jsonEntity);ex.printStackTrace();}log.debug(" body:{}", body );}

完整单元测试实现

完整单元测试类

package person.brickman.es;import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.testng.AbstractTestNGSpringContextTests;
import org.testng.annotations.Test;
import person.brickman.MainApplication;
import person.brickman.constant.ESConsts;
import person.brickman.util.RangeTime;
import person.brickman.util.TimeUtils;import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.List;/***   单元测试类**   1、RestClientBuilder初始化（同时支持单机与集群），因每个单元测试方法都会调，所以没有写独立的单元测试方法*   2、发送ES查询请求公共方法封装（支持sql、kql、代理访问、集群访问、鉴权），单元测试以SQL查询举例，参数直接拼的（使用占位符示例见：9、10）*   3、判断ES索引是否存在（/_cat/indices/${indexName}）*   4、判断ES索引别名是否存在 （/_cat/aliases/${indexName}）*   5、判断ES索引指定字段/属性是否存在（这里字段支持多级，如：logObj.id）*   6、判断ES索引指定字段/属性的类型（字段支持多级）*   7、阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）*   8、创建索引别名（可用于支持sql查询，索引名中有特殊字符不能用作表名，可通过创建别名来解决）*   9、SQL查询ES*   10、KQL查询ES** @Author:         brickman* @CreateDate:     2025/2/21 22:14* @Version:        1.0*/
@Slf4j
@SpringBootTest(classes = MainApplication.class)
public class ESRestClientServiceImplTest extends AbstractTestNGSpringContextTests {@Value("${elasticsearch.dand.interf.hosts}")private String host;@Value("${elasticsearch.dand.interf.port}")private int port;@Value("${elasticsearch.dand.interf.username}")private String username;@Value("${elasticsearch.dand.interf.password}")private String password;@Autowiredprivate ESRestClientService service;/*** 执行检索（这里使用sql查询近一分钟的数据）** @throws IOException* @throws ParseException*/@Test(groups = "hlog", enabled = true )public void testRequest() throws IOException, ParseException {String method = "POST";String endpoint = "/_sql?format=json";RangeTime rangeTime = new RangeTime();
//        String time = "2025-02-21 17:48:00";String time = TimeUtils.calcWholeMinute();rangeTime.calcAllByTimeAndPeriod(  time, 1);//  "yyyy-MM-dd HH:mm:ss"  --> "yyyy-MM-dd'T'HH:mm"  根据实际时间字段格式调整SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm" );
//        String cStart = sdf2.format(rangeTime.getStartDate());
//        String cEnd = sdf2.format(rangeTime.getEndDate());Long cStart = rangeTime.getStartDate().getTime();Long cEnd = rangeTime.getStartDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );
//        // hlog-crm3-hig-c-trans-2025.02.21 --》 hlog-crm3-comm-undef.2025.02.21
//        String idxName = "hlog-crm3-hig-c-trans-"+sdf_source.format(rangeTime.getStartDate());// hlog_crm3_hig_c_trans_2025_02_21 --》 hlog_crm3_comm_undef_2025_02_21String idxAliases = "hlog_crm3_comm_undef_"+sdf_target.format(rangeTime.getEndDate());// 生产环境String sqlQuery2 = "SELECT logObj.province,server, logObj.app, logObj.node_ip, " +"SUBSTRING(logObj.uri,0,30) , count(id) as num, avg(logObj.costTime) latency ,   " +"sum(iif(logObj.code=200,1,0))   as success_count " +" FROM " + idxAliases + " "+
//                "where cTime between '"+cStart+":00+0800' and '"+cEnd+":00+0800' " +" where cTime between "+cStart+" and "+cEnd+" " +"group by logObj.province,server, logObj.app, logObj.node_ip,  SUBSTRING(logObj.uri,0,30)  "; // 将index_name、field_name和value替换为相应的索引名称、字段名和值// max fetch_size is 65536  不同的版本单次查询最大数据限制不一样，这里测试只查 5 条String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\",  \"fetch_size\": 5}";String ret =   service.request(  host,   port,username,   password,method,   endpoint,   jsonEntity );log.debug("ret:{}",ret);}/*** 判断索引是否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsIndex() throws IOException {/*  hlog_crm3_hig_c_trans_2024_12_03hlog_crm3_hig_c_trans_2025_02_20hlog_crm3_comm_undef_2024_12_03hlog_crm3_comm_undef_2025_02_20*/boolean ret =   service.isExistsIndex( host, port,username,   password,"hlog_crm3_comm_undef_2025_02_21");log.info("ret:{}",ret);}/*** 判断索引别名是否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsAliases() throws IOException {/*  hlog_crm3_hig_c_trans_2025_02_21hlog_crm3_comm_undef_2025_02_21*/boolean ret =   service.isExistsAliases( host, port,username,   password,"hlog_crm3_comm_undef_2025_02_21");log.info("ret:{}",ret);}/*** 判断索引字段时否存在* @throws IOException*/@Test(groups = "hlog", enabled = true )public void testIsExistsProperty() throws IOException {// logObj.node_ip     logObj.statusboolean ret =   service.isExistsProperty( host, port,username, password,"hlog_crm3_comm_undef_2025_02_21", "logObj.status");log.info("ret:{}",ret);}/*** 获取索引列类型** @throws IOException*/@Test( enabled = true )public void testGetIndexPropertyType() throws IOException {/*  hlog_crm3_hig_c_trans_2025_02_21hlog_crm3_comm_undef_2025_02_21*/String ret =   service.getIndexPropertyType(   host,   port,username,   password,"hlog_crm3_comm_undef_2025_02_21","logObj.status"  );log.info("ret:{}",ret);}/*** 阻塞线程直至索引就绪（为了应对跨日时索引名短时间可能不存在的问题）** @throws IOException*/@Test(  enabled = true )public void testWaitIndexReady() throws IOException, ParseException, InterruptedException {RangeTime rangeTime = new RangeTime();String time = "2025-02-21 20:00:00";rangeTime.calcAllByTimeAndPeriod( time, 1 );service.waitIndexReady( host,   port,username,   password,"hlog_crm3_comm_undef_2025_02_21");}/*** 创建索引别名** @throws IOException*/@Test(  enabled = true )public void testCreateAliases() throws IOException, ParseException {RangeTime rangeTime = new RangeTime();String time = "2025-02-21 20:00:00";rangeTime.calcAllByTimeAndPeriod( time, 1 );service.createAliases( username,   password,host,   port,"hlog-crm3-comm-undef.2025.02.21","hlog_crm3_comm_undef_2025_02_21");}/*** sql查询** @throws IOException*/@Test(  enabled = true )public void testSQLRequest() throws IOException, ParseException, InterruptedException {log.info("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口/不区分接口)服务级 ：{}:{}", host, port );RangeTime rangeTime = new RangeTime();String time = "2025-02-21 20:00:00";rangeTime.calcAllByTimeAndPeriod( time, 1 );List<Object> ret=null;String method = "POST";String endpoint = "/_sql?format=json";//  根据实际时间字段格式调整SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm");long cStart = rangeTime.getStartDate().getTime();long cEnd = rangeTime.getEndDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );// hlog-crm3-comm-undef.2024.01.06String idxName = ESConsts.INTERFLOG_INDEX_NAME_PREFIX + sdf_source.format(rangeTime.getStartDate());// hlog_crm3_comm_undef_2024_01_06String idxAliases = ESConsts.INTERFLOG_INDEX_ALIASES_PREFIX + sdf_target.format(rangeTime.getStartDate());//等待索引就绪service.waitIndexReady(host,   port,username,   password,idxName);// 判断别名是否存在，不存在则创建if(!service.isExistsAliases(  host,   port,username,   password,idxAliases)){service.createAliases(  username,   password,  host,   port,idxName, idxAliases);}/*   logObj.status.keyword（对应String） 、还是 logObj.status（对应int ） 根据类型来  */String statusFieldName = getStatusFieldName(host,   port,username,   password,idxAliases );String sqlQuery2 = "SELECT logObj.province,server, logObj.app, '' node_ip, " +"  '' uri, count(id) as num, avg(logObj.cost) latency ,   " +"sum(iif("+statusFieldName+"=200,1,0))   as success_count " +" FROM " + idxAliases + " " +// 算头不算尾"where cTime >= ? and cTime < ? " +"group by logObj.province,server, logObj.app ";// max fetch_size is 65536 ， 已验证一次查询结果不会超过这个数，故不用做滚动查询String jsonEntity = "{\"query\": \"" + sqlQuery2 + "\", " +"\"params\": ["+cStart+","+cEnd+"]," +" \"fetch_size\": 10 }"; // 65536 -- 不同的版本单次查询最大数据限制不一样，这里测试只查10条String body =null;try{body = service.request(    host,   port,username,   password,method,   endpoint,   jsonEntity );// 组装成实际需要的业务类型集合
//            ret = ESIntfLoad3Parser.parseESresult4SvcLoad( body, IDC, host ,port, rangeTime);}catch (Exception ex){log.error("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口)服务级 指标异常！message:{}",ex.getLocalizedMessage());log.error("sqlQuery2:{}", sqlQuery2 );log.error("jsonEntity:{}", jsonEntity);ex.printStackTrace();}log.debug(" body:{}", body );}/*** kql查询** @throws IOException*/@Test(  enabled = true )public void testKQLRequest() throws IOException, ParseException, InterruptedException {log.info("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口/不区分接口)服务级 ：{}:{}", host, port );RangeTime rangeTime = new RangeTime();String time = "2025-02-21 20:00:00";rangeTime.calcAllByTimeAndPeriod( time, 1 );List<Object> ret=null;String method = "GET";String endpoint = "/hlog_crm3_comm_undef_2025_02_21/_search";SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm");// 这里cStart、cEnd两者相关1分钟，在上面做了初始化long cStart = rangeTime.getStartDate().getTime();long cEnd = rangeTime.getEndDate().getTime();// 原索引名是sql关键字，索引需要创建别名SimpleDateFormat sdf_source = new SimpleDateFormat("yyyy.MM.dd" );SimpleDateFormat sdf_target = new SimpleDateFormat("yyyy_MM_dd" );// hlog-crm3-comm-undef.2024.01.06String idxName = ESConsts.INTERFLOG_INDEX_NAME_PREFIX + sdf_source.format(rangeTime.getStartDate());// hlog_crm3_comm_undef_2024_01_06String idxAliases = ESConsts.INTERFLOG_INDEX_ALIASES_PREFIX + sdf_target.format(rangeTime.getStartDate());//等待索引就绪service.waitIndexReady(host,   port,username,   password,idxName);// 判断别名是否存在，不存在则创建if(!service.isExistsAliases(  host,   port,username,   password,idxAliases)){service.createAliases(  username,   password,  host,   port,idxName, idxAliases);}String kqlQuery2 = "{ " +"    \"bool\": { " +"      \"must\": [ " +"        { \"match_phrase\": { \"logObj.app\":\"inst-service\" } }, " +"        { \"range\": { " +"          \"cTime\": {  " +"            \"gte\": ?, " +"            \"lt\": ? " +"          } " +"        }} " +"      ] " +"    } " +"  }"; // 将index_name、field_name和value替换为相应的索引名称、字段名和值// max fetch_size is 65536 ， 已验证一次查询结果不会超过这个数，故不用做滚动查询String jsonEntity = "{\"query\": \"" + kqlQuery2 + "\", " +"\"params\": ["+cStart+","+cEnd+"]," +" \"size\": 0,\" +\n" +" \"size\": 2 }"; // 65536 -- 不同的版本单次查询最大数据限制不一样, 这里测试只查两条String body =null;try{body = service.request(    host,   port,username,   password,method,   endpoint,   jsonEntity );// 组装成实际需要的业务类型集合
//            ret = ESIntfLoad3Parser.parseESresult4SvcLoad( body, IDC, host ,port, rangeTime);}catch (Exception ex){log.error("实时从ES统计接口请求数量、成功率、延迟指标  (所有接口)服务级 指标异常！message:{}",ex.getLocalizedMessage());log.error("kqlQuery2:{}", kqlQuery2 );log.error("jsonEntity:{}", jsonEntity);ex.printStackTrace();}log.debug(" body:{}", body );}/***  查询索引 logObj.status 字段类型*    如果是 text 则使用 logObj.status.keyword*    如果是 long 则使用 logObj.status** @Author:         brickman* @CreateDate:     2025-02-21 17:48:00* @Version:        1.0*/private String getStatusFieldName(String host,   int port,String username,   String password,String idxAliases ) throws IOException {// logObj.status 字段 default: longString statusFieldName = "logObj.status";// 查询索引 logObj.status 字段类型if("text".equalsIgnoreCase( service.getIndexPropertyType(  host,   port,username,   password,idxAliases, "logObj.status" ) )){statusFieldName = "logObj.status.keyword";}return statusFieldName;}
}

单元测试效果

作者使用的testng，配置中心nacos

常量类

package person.brickman.constant;/*** @Description:   ES 常量类* @Author:         brickman* @CreateDate:     2025/2/21 22:05* @Version:        1.0*/
public class ESConsts {/**  interf 索引名常量  */public static final String INTERFLOG_INDEX_NAME_PREFIX = "interf-log-dand-comm-undef.";/**  interf 索引别名常量  */public static final String INTERFLOG_INDEX_ALIASES_PREFIX = "interf_log_dand_comm_undef_";/**  csb 索引名常量  */public static final String CSB_INDEX_NAME_PREFIX = "csb-service.csb.";/**  csb 索引别名常量  */public static final String CSB_INDEX_ALIASES_PREFIX = "csb_service_csb_";/**  skywalking 索引名常量  */public static final String SW_INDEX_NAME_PREFIX = "sw_metrics-all-";/**  skywalking 索引别名常量  */public static final String SW_INDEX_ALIASES_PREFIX = "sw_metrics_all_";
}

工具类

RangeTimeUtils

用于通过指定时间（如当前时间）生成开始时间、截止时间、采集周期、采集时间（跨度）范围、指标采集时间

package person.brickman.util;import lombok.Data;
import org.apache.commons.lang3.time.DateUtils;import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;/*** @Description:** s_start    String(19)	开始时间	因为是时间段的数据，所以增加4个字段* s_end	String(19)	截止时间* period	smallint	采集周期	多久采一次。默认1，单位分钟，与范围不同* n_range	smallint	采集时间（跨度）范围	采多长时间的数据。默认1，单位分钟 eg：1min* time	String(19)	指标采集时间	冗余字段，yyyy-MM-dd HH:mm:ss* 部分时序数据库以long形式显示timestamp，此字段便于查看** @Author: brickman* @CreateDate: 2025/02/21 9:03 PM* @Version: 1.0*/
@Data
public class RangeTimeUtils {/** 开始时间 yyyy-MM-dd HH:mm:ss   */private String cStart;/** 截止时间 yyyy-MM-dd HH:mm:ss */private String cEnd;/** 采集周期	多久采一次。默认1，单位分钟，与范围不同  */private int period;/*** 采集时间（跨度）范围* @deprecated  一般与采集周期一致* */private int nRange;/*** 采集时间  yyyy-MM-dd HH:mm:ss* eg： 2025-02-21 17:48:00*/private String time;/*** 通过时间和周期计算所有字段值*   默认偏移1分钟，因之前调的默认方法且需要偏移，所有遵循开闭原则* @param time eg： 2025-02-21 17:48:00* @param period 单位：分钟* @return void**/public void calcAllByTimeAndPeriod( String time, int period ) throws ParseException {calcAllByTimeAndPeriod( time, period, -1 );}/*** @param shifting 波动时间、偏移时间* @return null**/public void calcAllByTimeAndPeriod( String time, int period, int shifting ) throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");Date dt = sdf1.parse(time);// dt:date timeSimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd HH:mm");Date de = sdf2.parse(sdf2.format(dt));//de:date endDate ds = DateUtils.addMinutes(de, -period ); // ds: date start/*   */ds = DateUtils.addMinutes(ds, shifting);de = DateUtils.addMinutes(de, shifting);this.cStart = sdf1.format(ds);this.cEnd = sdf1.format(de);this.period=period;// 采集时间（跨度）范围 与 采集周期一致this.nRange = this.period;this.time=time;}public Date getStartDate() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");return sdf1.parse(this.cStart);// dt:date time}public Date getEndDate() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");return sdf1.parse(this.cEnd);// dt:date time}public Date getGraphQLStartDate() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");return sdf1.parse(this.cStart);// dt:date time}public Date getGraphQLEndDate() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");return sdf1.parse(this.cEnd);// dt:date time}public String getGraphQLStart() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HHmm");return sdf1.format(getGraphQLStartDate());// dt:date time}public String getGraphQLEnd() throws ParseException {SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HHmm");return sdf1.format(getGraphQLEndDate());// dt:date time}
}

TimeUtils

取当前时间精确到分（取整分）

package person.brickman.util;import java.text.SimpleDateFormat;
import java.util.Date;/*** @Description:    时间工具类*     取当前时间精确到分（取整分）** @Author:         brickman* @CreateDate:     2025/2/21 22:09* @Version:        1.0*/
public class TimeUtils {/**  取当前时间精确到分（取整分），可优化为取数据库时间   **/public static String calcWholeMinute(){SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm");String time = sdf.format(new Date());System.out.println("#### WholeMinuteTime:"+time+":00");return time+":00";}/**  根椐入参取整分   **/public static String calcWholeMinute( Date date ){SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm");String time = sdf.format( date );System.out.println("#### WholeMinuteTime:"+time+":00");return time+":00";}
}

总结

1、ElasticSearch公共方法封装能降低开发成本、提高开发效率

2、非常通用的方法，如查Skywalking的索引数据、查自研的接口调用日志数据等

3、说本文将ES的日常开发代码一网代尽都不为过

附件一：ElasticSearch介绍

ElasticSearch介绍

在Elasticsearch（简称ES）中，它是一个基于Apache Lucene构建的开源、分布式、RESTful搜索引擎，旨在实时地存储、搜索和分析大量数据。Elasticsearch广泛应用于日志分析、全文搜索、实时分析等场景。下面我将介绍Elasticsearch的一些基本概念和功能：

1. 基本概念

索引（Index）：在Elasticsearch中，索引类似于传统关系数据库中的“数据库”。它是存储相关文档（数据）的地方。
类型（Type）：在Elasticsearch 7.x及以前版本中，每个索引可以包含多个类型。但从Elasticsearch 7.x开始，一个索引中只能有一个类型（默认为_doc），这一改动主要是因为Elasticsearch 8.x将完全废弃类型功能。
文档（Document）：文档是Elasticsearch中最小的数据单元，可以是JSON格式的数据。每个文档都有一个唯一的ID。
字段（Field）：文档由一个或多个字段组成，每个字段都有一个名称和一个值。
映射（Mapping）：映射定义了索引中文档的结构，包括字段的类型、是否索引、是否存储等属性。