第 15 章：Java 客户端与 Spring Boot 集成

zjc 于 2026-01-15 发布

这是《Elasticsearch 零基础实战指南》的独立章节版。本章从概念、实操和生产排查三个视角展开，代码块保留了原书可直接运行的版本。 Elasticsearch 8.x 官方推荐使用 Java API Client。它基于 Jackson，提供强类型请求构建和响应模型，替代了已经废弃的 High Level REST Client。

本章覆盖依赖、客户端创建、连接配置、搜索封装、聚合解析、Bulk 写入、异常处理和 Spring Boot 集成。

15.1 客户端选型

客户端	状态	建议
Java API Client	官方新客户端	新项目推荐
High Level REST Client	已废弃	老项目逐步迁移
Low Level REST Client	维护中	特殊场景或底层控制
Spring Data Elasticsearch	社区封装	快速开发，复杂搜索仍可透传 DSL
Jest	非官方，生态弱化	不建议新项目

Java API Client 可以与 Low Level REST Client 配合使用，自定义序列化、传输和拦截。

15.2 添加依赖

Maven：

<dependency>
    <groupId>co.elastic.clients</groupId>
    <artifactId>elasticsearch-java</artifactId>
    <version>8.15.0</version>
</dependency>

Spring Boot 3 示例：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>co.elastic.clients</groupId>
    <artifactId>elasticsearch-java</artifactId>
    <version>8.15.0</version>
</dependency>
<dependency>
    <groupId>jakarta.json</groupId>
    <artifactId>jakarta.json-api</artifactId>
    <version>2.1.3</version>
</dependency>

版本应与集群版本保持兼容，避免直接使用差距过大的客户端。

15.3 创建客户端

基础创建：

RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200, "http")
).build();

ElasticsearchTransport transport = new RestClientTransport(
        restClient,
        new JacksonJsonpMapper()
);

ElasticsearchClient client = new ElasticsearchClient(transport);

带认证：

final CredentialsProvider provider = new BasicCredentialsProvider();
provider.setCredentials(
        AuthScope.ANY,
        new UsernamePasswordCredentials("app_user", "password")
);

RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200, "https")
).setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
        .setDefaultCredentialsProvider(provider)
        .setSSLContext(sslContext)
).build();

15.4 Spring Boot 配置

application.yml：

elasticsearch:
  hosts: https://es01:9200
  username: app_user
  password: ${ES_PASSWORD}
  connect-timeout: 2s
  socket-timeout: 5s
  max-conn-total: 100
  max-conn-per-route: 20

配置类：

@Configuration
@EnableConfigurationProperties(ElasticsearchProperties.class)
public class ElasticsearchConfig {

    @Bean(destroyMethod = "close")
    public RestClient restClient(ElasticsearchProperties properties) {
        return RestClient.builder(HttpHost.create(properties.getHosts()))
                .setRequestConfigCallback(builder -> builder
                        .setConnectTimeout((int) properties.getConnectTimeout().toMillis())
                        .setSocketTimeout((int) properties.getSocketTimeout().toMillis()))
                .setHttpClientConfigCallback(builder -> builder
                        .setMaxConnTotal(properties.getMaxConnTotal())
                        .setMaxConnPerRoute(properties.getMaxConnPerRoute()))
                .build();
    }

    @Bean
    public ElasticsearchClient elasticsearchClient(RestClient restClient) {
        ElasticsearchTransport transport =
                new RestClientTransport(restClient, new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
    }
}

属性类：

@ConfigurationProperties(prefix = "elasticsearch")
public class ElasticsearchProperties {
    private String hosts;
    private String username;
    private String password;
    private Duration connectTimeout = Duration.ofSeconds(2);
    private Duration socketTimeout = Duration.ofSeconds(5);
    private int maxConnTotal = 100;
    private int maxConnPerRoute = 20;
}

15.5 索引和文档操作

创建索引：

client.indices().create(c -> c.index("products")
        .mappings(m -> m
                .properties("title", p -> p.text(t -> t.analyzer("ik_max_word")))
                .properties("brand", p -> p.keyword(k -> k.ignoreAbove(64)))
                .properties("price", p -> p.scaledFloat(f -> f.scalingFactor(100.0)))
        )
        .settings(s -> s.numberOfShards(3).numberOfReplicas(1))
);

写入文档：

Product product = new Product(10001L, "轻薄笔记本", "NOVA", 6999.00);

IndexResponse response = client.index(i -> i
        .index("products")
        .id(String.valueOf(product.getId()))
        .document(product)
);

读取：

GetResponse<Product> response = client.get(g -> g
        .index("products")
        .id("10001"),
        Product.class
);

if (response.found()) {
    Product product = response.source();
}

15.6 构建查询

public SearchRequest buildSearch(ProductSearchQuery query) {
    return SearchRequest.of(r -> r
            .index("products")
            .from(query.from())
            .size(query.size())
            .query(q -> q.bool(b -> {
                b.must(m -> m.multiMatch(mm -> mm
                        .query(query.keyword())
                        .fields("title^3", "brand", "description")
                        .type(TextQueryType.BestFields)));

                b.filter(f -> f.term(t -> t.field("status").value("ON_SALE")));

                if (query.brand() != null && !query.brand().isEmpty()) {
                    b.filter(f -> f.terms(t -> t.field("brand")
                            .terms(v -> v.value(query.brand().stream()
                                    .map(FieldValue::of)
                                    .toList()))));
                }
                return b;
            }))
            .sort(s -> s.field(f -> f.field("sales").order(SortOrder.Desc)))
            .source(src -> src.filter(f -> f.includes("id", "title", "brand", "price")))
    );
}

执行：

SearchResponse<Product> response =
        client.search(buildSearch(query), Product.class);

解析：

List<Product> products = response.hits().hits().stream()
        .map(Hit::source)
        .filter(Objects::nonNull)
        .toList();

15.7 分页与排序封装

public record PageCursor(List<Object> values) {}

public SearchRequest buildSearchAfter(ProductSearchQuery query, PageCursor cursor) {
    return SearchRequest.of(r -> r
            .index("products")
            .size(query.size())
            .query(buildQuery(query))
            .sort(s -> s.field(f -> f.field("sales").order(SortOrder.Desc)))
            .sort(s -> s.field(f -> f.field("id").order(SortOrder.Asc)))
            .searchAfter(cursor == null ? List.of() : cursor.values())
    );
}

返回下一页游标：

List<Hit<Product>> hits = response.hits().hits();
if (!hits.isEmpty()) {
    List<Object> nextCursor = hits.get(hits.size() - 1).sort();
}

15.8 聚合解析

SearchResponse<Void> response = client.search(s -> s
        .index("orders")
        .size(0)
        .query(q -> q.range(r -> r.date(d -> d
                .field("created_at")
                .gte(JsonData.of("now-30d/d")))))
        .aggregations("status", a -> a.terms(t -> t
                .field("status")
                .size(20))),
        Void.class
);

Aggregate aggregate = response.aggregations().get("status");
List<StringTermsBucket> buckets = aggregate.sterms().buckets().array();

for (StringTermsBucket bucket : buckets) {
    System.out.println(bucket.key().stringValue() + ": " + bucket.docCount());
}

类型与字段 Mapping 相关，数值字段可能是 lterms，日期 histogram 是 dateHistogram。建议在服务层封装聚合结果 DTO，避免上层直接依赖客户端模型。

15.9 Bulk 写入

public void bulkUpsert(List<Product> products) throws IOException {
    BulkRequest.Builder builder = new BulkRequest.Builder();

    for (Product product : products) {
        Map<String, Object> doc = Map.of(
                "id", product.getId(),
                "title", product.getTitle(),
                "brand", product.getBrand(),
                "price", product.getPrice()
        );

        builder.operations(op -> op.index(idx -> idx
                .index("products")
                .id(String.valueOf(product.getId()))
                .document(doc)
                .version(product.getVersion())
                .versionType(VersionType.External)
        ));
    }

    BulkResponse response = client.bulk(builder.build());
    if (response.errors()) {
        for (BulkItemResponse item : response.items()) {
            if (item.isFailed()) {
                log.warn("bulk failed index={} id={} error={}",
                        item.index(), item.id(), item.error().reason());
            }
        }
    }
}

批量大小应压测确定，常见 500 到 5000 条或 5 到 15MB。

15.10 异常处理

常见异常：

异常	场景	处理
`IOException`	网络失败	重试或熔断
`ElasticsearchException`	服务端返回错误	解析 status/reason
`429`	请求被拒绝	降低并发、退避
`401/403`	认证或权限错误	不重试，告警
`404`	索引或文档不存在	按业务处理
`409`	版本冲突	重新读取或幂等处理

示例：

public <T> T executeWithRetry(Supplier<T> action) {
    int maxRetry = 2;
    for (int i = 0; i <= maxRetry; i++) {
        try {
            return action.get();
        } catch (ElasticsearchException e) {
            if (e.status() == 401 || e.status() == 403) {
                throw e;
            }
            if (i == maxRetry) {
                throw e;
            }
            sleepQuietly(Duration.ofMillis(100L << i));
        }
    }
    throw new IllegalStateException("unreachable");
}

15.11 搜索服务分层

推荐分层：

Controller
  -> SearchApplicationService
     -> QueryBuilder
     -> EsClientGateway
     -> ResponseAssembler

职责：

层	职责
Controller	参数校验、鉴权、协议转换
ApplicationService	业务流程、缓存、降级
QueryBuilder	业务条件转 DSL
Gateway	ES 客户端、重试、异常转换
ResponseAssembler	ES 响应转业务 DTO

不要在 Controller 中直接拼复杂 DSL。

15.12 观测与慢日志

long start = System.nanoTime();
try {
    SearchResponse<Product> response = client.search(request, Product.class);
    metrics.searchSuccess(query.type(), response.took());
    return response;
} catch (Exception e) {
    metrics.searchFailure(query.type(), e.getClass().getSimpleName());
    throw new SearchUnavailableException(e);
} finally {
    long cost = System.nanoTime() - start;
    if (cost > Duration.ofMillis(500).toNanos()) {
        log.warn("slow es query type={} request={}", query.type(), request);
    }
}

记录内容：

traceId；
索引名；
查询类型；
DSL 摘要；
took；
shards 统计；
命中数；
失败原因。

15.13 Spring Data Elasticsearch

适合简单 CRUD：

public interface ProductRepository
        extends ElasticsearchRepository<Product, Long> {

    List<Product> findByBrandAndStatus(String brand, String status);
}

复杂查询可以使用 @Query 或直接注入 ElasticsearchOperations。当查询逻辑复杂、性能要求高、需要精细化控制时，直接使用 Java API Client 更清晰。

15.14 本章小结

新项目使用 Java API Client，逐步迁移 High Level REST Client；
客户端要统一配置连接池、认证、超时和 TLS；
查询构建、执行、解析分层，避免业务层直接暴露复杂 DSL；
bulk 必须逐条检查失败；
搜索调用要有重试、熔断、指标和慢日志；
Spring Data 适合简单场景，复杂搜索建议原生客户端。

15.15 思考题

High Level REST Client 废弃后，新项目应如何选择客户端？
搜索超时应如何设置？超时后是否无限重试？
为什么要在 Gateway 层统一转换 ES 异常？
如何记录一次搜索请求的可观测信息？
Bulk 写入失败后如何设计重试和死信？