基于 SpringBoot 2.7.x 使用最新的 Elasticsearch Java API Client 之 ElasticsearchClient

1. 从 RestHighLevelClient 到 ElasticsearchClient

从 Java Rest Client 7.15.0 版本开始，Elasticsearch 官方决定将 RestHighLevelClient 标记为废弃的，并推荐使用新的 Java API Client，即 ElasticsearchClient. 为什么要将 RestHighLevelClient 废弃，大概有以下几点：

维护成本高：RestHighLevelClient 需要和 Elasticsearch APIs 的更新保持一致，而 Elasticsearch APIs 更新较为频繁，因此每次 Elasticsearch APIs 有新的迭代，RestHighLevelClient 也要跟着迭代，维护成本高。
兼容性差: 由于 RestHighLevelClient 和 Elasticsearch 的内部数据结构紧耦合，而 Elasticsearch 不同版本的数据结构可能略有差异，因此很难跨不同版本的 Elasticsearch 保持兼容。
灵活度低: RestHighLevelClient 的灵活性扩展性较差，很难去扩展或者自定义一些功能。

而 Spring 官方对 Elasticsearch 客户端也进行了封装，集成于 spring-boot-starter-data-elasticsearch 模块，Elasticsearch 官方决定废弃 RestHighLevelClient 而支持 ElasticsearchClient 这一举措，必然也导致 Spring 项目组对 data-elasticserach 模块进行同步更新，以下是 Spring 成员对相关内容的讨论：

https://github.com/spring-projects/spring-boot/issues/28597

大概内容就是在对 ElasticsearchClient 自动装配的支持会在 springboot 3.0.x 版本中体现，而在 2.7.x 版本中会将 RestHighLevelClient 标记为废弃的。

由于我们的项目是基于 springboot 2.7.10 版本开发的，而 2.7.x 作为最后一个 2.x 版本，springboot 下个版本为 3.x，恰逢项目已经规划在半年后将 JDK 升级为17版本，全面支持 springboot 3.x 版本的替换，因此现阶段需要封装一个能够跨 2.7.x 和 3.x 版本都可以使用的 Elasticsearch 客户端。

2. 自定义 starter 模块实现 ElasticsearchTemplate 的自动装配

在调研了 spring-boot 2.7.10 版本的源码后发现，其实 2.7.x 版本已经引入了 ElasticsearchClient，并封装了新的客户端 ElasticsearchTemplate，但是并没有为其做自动装配，如果想要使用基于ElasticsearchClient 的 ElasticsearchTemplate，需要用户自己装配。否则，直接使用 ElasticsearchTemplate 会出现以下提示：

Consider defining a bean of type 'org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate' in your configuration.

即由提示可以知道，无法创建一个 ElasticsearchTemplate 类型的 bean.

因此需要自己实现 ElasticsearchTemplate 的装配，才可以使用。为了能够一次装配多项目复用，决定自己构建一个starter，之后需要使用 ElasticsearchTemplate，可以通过引入依赖的方式完成自动装配。

自定义的 starter 项目目录结构如下图所示：
在这里插入图片描述

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>
    <description>自定义elasticsearch-client组件</description>
    <parent>
        <artifactId>xxx-spring-boot-starters</artifactId>
        <groupId>com.xxx.commons</groupId>
        <version>${revision}</version>
    </parent>
    <artifactId>xxx-elasticsearch-client-spring-boot-starter</artifactId>
    <packaging>jar</packaging>
    <name>xxx-elasticsearch-client-spring-boot-starter</name>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>jakarta.json</groupId>
                    <artifactId>jakarta.json-api</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>jakarta.json</groupId>
            <artifactId>jakarta.json-api</artifactId>
            <version>2.0.1</version>
        </dependency>
    </dependencies>
</project>

org.springframework.boot.autoconfigure.AutoConfiguration.imports 文件

com.xxx.commons.springboot.elasticsearch.ElasticsearchTemplateAutoConfiguration
com.xxx.commons.springboot.elasticsearch.actuate.xxxElasticsearchHealthIndicatorAutoConfiguration

PackageInfo 接口：

package com.xxx.commons.springboot.elasticsearch;

/**
 * @author reader
 * Date: 2023/9/18 22:21
 **/
public interface PackageInfo {
}

RestClientBuilder 类：

package com.xxx.commons.springboot.elasticsearch;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchProperties;

import java.net.URI;
import java.net.URISyntaxException;

/**
 * @author reader
 * Date: 2023/9/20 15:16
 **/
public final class RestClientBuilder {

    private RestClientBuilder() {
    }

    public static RestClient buildWithProperties(ElasticsearchProperties properties) {
        HttpHost[] hosts = properties.getUris().stream().map(RestClientBuilder::createHttpHost).toArray((x$0) -> new HttpHost[x$0]);
        org.elasticsearch.client.RestClientBuilder builder = RestClient.builder(hosts);
        builder.setDefaultHeaders(new BasicHeader[]{new BasicHeader("Content-type", "application/json")});
        builder.setHttpClientConfigCallback((httpClientBuilder) -> {
            httpClientBuilder.addInterceptorLast((HttpResponseInterceptor) (response, context) -> response.addHeader("X-Elastic-Product", "Elasticsearch"));
            if (hasCredentials(properties.getUsername(), properties.getPassword())) {
                // 密码配置
                CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(properties.getUsername(), properties.getPassword()));
                httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
            // httpClient配置
            return httpClientBuilder;
        });
        builder.setRequestConfigCallback((requestConfigBuilder) -> {
            // request配置
            requestConfigBuilder.setConnectionRequestTimeout((int)properties.getConnectionTimeout().getSeconds() * 1000);
            requestConfigBuilder.setSocketTimeout((int)properties.getSocketTimeout().getSeconds() * 1000);
            return requestConfigBuilder;
        });
        if (properties.getPathPrefix() != null) {
            builder.setPathPrefix(properties.getPathPrefix());
        }
        return builder.build();
    }

    private static boolean hasCredentials(String username, String password) {
        return StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password);
    }

    private static HttpHost createHttpHost(String uri) {
        try {
            return createHttpHost(URI.create(uri));
        } catch (IllegalArgumentException var2) {
            return HttpHost.create(uri);
        }
    }

    private static HttpHost createHttpHost(URI uri) {
        if (StringUtils.isBlank(uri.getUserInfo())) {
            return HttpHost.create(uri.toString());
        } else {
            try {
                return HttpHost.create((new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment())).toString());
            } catch (URISyntaxException var2) {
                throw new IllegalStateException(var2);
            }
        }
    }
}

ElasticsearchClientConfiguration 类：

package com.xxx.commons.springboot.elasticsearch;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.elasticsearch.client.RestClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author reader
 * Date: 2023/9/20 14:59
 **/
@Configuration
@EnableConfigurationProperties({ElasticsearchProperties.class})
@ConditionalOnClass({ElasticsearchClient.class, ElasticsearchTransport.class})
public class ElasticsearchClientConfiguration {

    protected static final Log LOGGER = LogFactory.getLog(ElasticsearchClientConfiguration.class);

    private ElasticsearchProperties elasticsearchProperties;

    public ElasticsearchClientConfiguration(ElasticsearchProperties elasticsearchProperties) {
        LOGGER.info("框架 elasticsearch-client-starter elasticsearchProperties 装载开始");
        this.elasticsearchProperties = elasticsearchProperties;
    }

    @Bean
    public ElasticsearchClient elasticsearchClient() {
        LOGGER.info("框架 elasticsearch-client-starter elasticsearchClient 装载开始");
        RestClient restClient = RestClientBuilder.buildWithProperties(elasticsearchProperties);
        RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
    }
}

package com.xxx.commons.springboot.elasticsearch;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import com.xxx.commons.springboot.elasticsearch.actuate.ElasticsearchInfoContributor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.actuate.autoconfigure.info.ConditionalOnEnabledInfoContributor;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.data.elasticsearch.ElasticsearchDataAutoConfiguration;
import org.springframework.boot.autoconfigure.domain.EntityScanner;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchProperties;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchRestClientAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
import org.springframework.data.elasticsearch.core.convert.ElasticsearchCustomConversions;
import org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverter;
import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;

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


/**
 * @author reader
 * Date: 2023/9/19 16:35
 **/
@AutoConfiguration(before = {ElasticsearchRestClientAutoConfiguration.class, ElasticsearchDataAutoConfiguration.class})
@ConditionalOnClass({ElasticsearchTemplate.class})
@EnableConfigurationProperties({ElasticsearchProperties.class})
@Import({ElasticsearchClientConfiguration.class})
public class ElasticsearchTemplateAutoConfiguration {

    protected static final Log LOGGER = LogFactory.getLog(ElasticsearchTemplateAutoConfiguration.class);

    @Bean
    ElasticsearchCustomConversions elasticsearchCustomConversions() {
        return new ElasticsearchCustomConversions(Collections.emptyList());
    }

    @Bean
    public SimpleElasticsearchMappingContext elasticsearchMappingContext(ApplicationContext applicationContext,
                                                                         ElasticsearchCustomConversions elasticsearchCustomConversions) throws ClassNotFoundException {
        SimpleElasticsearchMappingContext mappingContext = new SimpleElasticsearchMappingContext();
        mappingContext.setInitialEntitySet(new EntityScanner(applicationContext).scan(Document.class));
        mappingContext.setSimpleTypeHolder(elasticsearchCustomConversions.getSimpleTypeHolder());
        return mappingContext;
    }

    @Bean
    ElasticsearchConverter elasticsearchConverter(SimpleElasticsearchMappingContext mappingContext,
                                                  ElasticsearchCustomConversions elasticsearchCustomConversions) {
        MappingElasticsearchConverter converter = new MappingElasticsearchConverter(mappingContext);
        converter.setConversions(elasticsearchCustomConversions);
        return converter;
    }

    @Bean
    ElasticsearchTemplate elasticsearchTemplate(ElasticsearchClient client, ElasticsearchConverter converter) {
        LOGGER.info("框架 elasticsearch-client-starter elasticsearchTemplate 装载开始");
        return new ElasticsearchTemplate(client, converter);
    }

    @Bean
    @ConditionalOnEnabledInfoContributor("elasticsearch")
    public ElasticsearchInfoContributor elasticsearchInfoContributor(ObjectProvider<ElasticsearchProperties> propertiesObjectProvider) {
        List<ElasticsearchProperties> properties = new ArrayList<>();
        propertiesObjectProvider.forEach(properties::add);
        return new ElasticsearchInfoContributor(properties);
    }
}

健康度指标相关的封装有：

ElasticsearchHealthIndicator 类：

package com.xxx.commons.springboot.elasticsearch.actuate;

import org.elasticsearch.client.Node;
import org.elasticsearch.client.RestClient;
import org.springframework.boot.actuate.health.AbstractHealthIndicator;
import org.springframework.boot.actuate.health.Health;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author reader
 * Date: 2023/9/20 19:26
 **/
public class ElasticsearchHealthIndicator extends AbstractHealthIndicator {

    private final List<RestClient> clients;

    public ElasticsearchHealthIndicator(List<RestClient> clients) {
        this.clients = clients;
    }

    @Override
    protected void doHealthCheck(Health.Builder builder) throws Exception {
            boolean success = true;
            Map<String, Object> properties = new HashMap<>();

            for (RestClient client : clients) {
                List<Node> nodes = client.getNodes();
                if (null == nodes || nodes.isEmpty()){
                    continue;
                }
                String id = nodes.stream().map(Node::toString).collect(Collectors.joining(";"));
                boolean ps = client.isRunning();
                properties.put("ElasticsearchClient[" + id + "]", ps);
                if (!ps) {
                    success = false;
                }
            }

            if (success) {
                builder.up();
            } else {
                builder.withDetails(properties).down();
            }
    }
}

ElasticsearchInfoContributor 类：

package com.xxx.commons.springboot.elasticsearch.actuate;

import com.xxx.commons.springboot.elasticsearch.PackageInfo;
import org.springframework.boot.actuate.info.Info;
import org.springframework.boot.actuate.info.InfoContributor;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchProperties;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author reader
 * Date: 2023/9/20 19:32
 **/
public class ElasticsearchInfoContributor implements InfoContributor {

    private final List<ElasticsearchProperties> elasticsearchProperties;

    public ElasticsearchInfoContributor(List<ElasticsearchProperties> elasticsearchProperties) {
        this.elasticsearchProperties = elasticsearchProperties;
    }

    @Override
    public void contribute(Info.Builder builder) {
        Map<String, Object> properties = new HashMap<>();
        properties.put("version", PackageInfo.class.getPackage().getImplementationVersion());
        properties.put("_title_", "ElasticsearchTemplate组件");
        elasticsearchProperties.forEach(p -> {
            Map<String, Object> sp = new HashMap<>();
            String id = String.join(";", p.getUris());
            properties.put(id, sp);
            sp.put("nodes", String.join(";", p.getUris()));
            sp.put("user", p.getUsername());
            sp.put("connectionTimeout[ms]", p.getConnectionTimeout().toMillis());
            sp.put("socketTimeout[ms]", p.getSocketTimeout().toMillis());
        });
        builder.withDetail("xxx-elasticsearch-client", properties);
    }
}

xxxElasticsearchHealthIndicatorAutoConfiguration 类：

package com.xxx.commons.springboot.elasticsearch.actuate;

import org.elasticsearch.client.RestClient;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
import org.springframework.boot.actuate.autoconfigure.health.HealthContributorAutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;

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

/**
 * @author reader
 * Date: 2023/9/20 19:45
 **/
@AutoConfiguration(before = {HealthContributorAutoConfiguration.class})
@ConditionalOnEnabledHealthIndicator("elasticsearch")
public class xxxElasticsearchHealthIndicatorAutoConfiguration {

    @Bean("elasticsearchHealthIndicator")
    @ConditionalOnMissingBean
    public ElasticsearchHealthIndicator xxxElasticHealthIndicator(ObjectProvider<RestClient> elasticsearchClientProvider) {
        List<RestClient> restClients = new ArrayList<>();
        elasticsearchClientProvider.forEach(restClients::add);
        return new ElasticsearchHealthIndicator(restClients);
    }
}

3. 使用自定义的 starter

1、在自己封装了一个 starter 工具模块之后，通过引入依赖的方式使用，引入的依赖为：

<dependency>
    <groupId>com.xxx.commons</groupId>
    <artifactId>xxx-elasticsearch-client-spring-boot-starter</artifactId>
    <version>${version}</version>
</dependency>

在 yaml 文件中配置的相关属性信息：

spring:
  elasticsearch:
    uris: http://127.0.0.1:9200         
    username: elastic
    password: password

注入并使用 ElasticsearchTemplate 对 ES 进行操作：

package com.xxx.xxx;

import com.xxx.commons.result.query.PaginationBuilder;
import com.xxx.commons.result.query.Query;
import com.xxx.commons.result.query.QueryBuilder;
import com.xxx.push.domain.AliPushRecordDO;
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.data.domain.PageRequest;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.client.elc.NativeQuery;
import org.springframework.data.elasticsearch.client.elc.NativeQueryBuilder;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.CollectionUtils;

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

/**
 * @author reader
 * Date: 2023/9/26 14:42
 **/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class, properties = {"profile=dev", "debug=true"})
public class ElasticsearchTemplateTest {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Test
    public void testSearch() {
        Query query = QueryBuilder.page(1).pageSize(20).build();
        NativeQueryBuilder nativeQueryBuilder = new NativeQueryBuilder();
        nativeQueryBuilder.withPageable(PageRequest.of(query.getPage() - 1, query.getPageSize()));
        NativeQuery searchQuery = nativeQueryBuilder.build();
        // 查询总数
        long count = elasticsearchTemplate.count(searchQuery, AliPushRecordDO.class);
        PaginationBuilder<AliPushRecordDO> builder = PaginationBuilder.query(query);
        builder.amount((int) count);
        if (count > 0) {
            SearchHits<AliPushRecordDO> aliPushRecordDOSearchHits = elasticsearchTemplate.search(searchQuery, AliPushRecordDO.class);
            List<SearchHit<AliPushRecordDO>> searchHits = aliPushRecordDOSearchHits.getSearchHits();
            List<AliPushRecordDO> aliPushRecordDOList = new ArrayList<>();
            if (!CollectionUtils.isEmpty(searchHits)) {
                searchHits.forEach(searchHit -> aliPushRecordDOList.add(searchHit.getContent()));
            }
            builder.result(aliPushRecordDOList);
        } else {
            builder.result(new ArrayList<>());
        }
    }
}