【Elasticsearch】（Java 版）

Elasticsearch（Java 版）

1. Elasticsearch 简介

1.1 什么是 Elasticsearch？

• 基于 Apache Lucene 的分布式搜索和分析引擎。
• 支持近实时（NRT）搜索、结构化查询、全文检索、复杂聚合分析。
• 适用于日志分析、监控系统、电商搜索、大数据分析等场景。

1.2 核心概念

• 文档（Document）：数据的基本单元（JSON 格式）。
• 索引（Index）：文档的集合（类似数据库中的表）。
• 分片（Shard）：索引的横向拆分，支持分布式存储。
• 副本（Replica）：分片的副本，提供高可用和负载均衡。
• 节点（Node）：单个 ES 实例，多个节点组成集群（Cluster）。
• 倒排索引（Inverted Index）：通过词项（Term）快速定位文档的数据结构。

2. 安装与配置

2.1 环境要求

• JDK 8 或更高版本。
• 推荐内存：4GB+，磁盘 SSD。

2.2 安装步骤

Linux/macOS

# 下载 Elasticsearch
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.10.0-linux-x86_64.tar.gz
tar -xzf elasticsearch-8.10.0-linux-x86_64.tar.gz
cd elasticsearch-8.10.0/

# 启动单节点集群
./bin/elasticsearch

Windows

1. 下载 ZIP 包并解压。
2. 运行 bin\elasticsearch.bat。

2.3 验证安装

访问 http://localhost:9200，返回 JSON 信息即成功：

{
  "name": "node-1",
  "cluster_name": "elasticsearch",
  "version": { ... }
}

2.4 配置参数

修改 config/elasticsearch.yml：

cluster.name: my-cluster
node.name: node-1
network.host: 0.0.0.0
http.port: 9200
discovery.type: single-node  # 单节点模式

3. Java 客户端操作

3.1 引入依赖

在 pom.xml 中添加 Elasticsearch Java 客户端和 Jackson 依赖：

<dependency>
    <groupId>co.elastic.clients</groupId>
    <artifactId>elasticsearch-java</artifactId>
    <version>8.10.0</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.15.2</version>
</dependency>

3.2 创建客户端

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.http.HttpHost;
import org.elasticsearch.client.RestClient;

public class ElasticsearchExample {
    public static void main(String[] args) {
        // 创建低级客户端，连接到本地 Elasticsearch 实例
        RestClient restClient = RestClient.builder(
            new HttpHost("localhost", 9200) // ES 服务器地址和端口
        ).build();

        // 创建传输层，使用 Jackson 作为 JSON 处理器
        ElasticsearchTransport transport = new RestClientTransport(
            restClient, new JacksonJsonpMapper()
        );

        // 创建 Elasticsearch Java 客户端
        ElasticsearchClient client = new ElasticsearchClient(transport);

        // 后续操作可以使用 client 对象进行
    }
}

3.3 索引操作

创建索引

// 创建名为 "products" 的索引
client.indices().create(c -> c.index("products"));
System.out.println("索引创建成功！");

删除索引

// 删除名为 "products" 的索引
client.indices().delete(c -> c.index("products"));
System.out.println("索引删除成功！");

3.4 文档操作

定义文档类

public class Product {
    private String name;
    private double price;
    private String category;

    // 构造函数、Getter 和 Setter 方法
    public Product(String name, double price, String category) {
        this.name = name;
        this.price = price;
        this.category = category;
    }

    // Getter 和 Setter 方法
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public double getPrice() { return price; }
    public void setPrice(double price) { this.price = price; }
    public String getCategory() { return category; }
    public void setCategory(String category) { this.category = category; }
}

插入文档

// 创建一个 Product 对象
Product product = new Product("Laptop", 999.99, "Electronics");

// 将文档插入到 "products" 索引中，ID 为 "1"
client.index(i -> i
    .index("products")  // 指定索引名称
    .id("1")           // 指定文档 ID
    .document(product) // 指定文档内容
);
System.out.println("文档插入成功！");

查询文档

// 根据 ID 查询文档
GetResponse<Product> response = client.get(g -> g
    .index("products")  // 指定索引名称
    .id("1"),           // 指定文档 ID
    Product.class       // 指定返回的文档类型
);

// 获取查询结果
Product product = response.source();
if (product != null) {
    System.out.println("查询结果：" + product.getName() + ", " + product.getPrice());
} else {
    System.out.println("未找到文档！");
}

更新文档

// 更新 ID 为 "1" 的文档
client.update(u -> u
    .index("products")  // 指定索引名称
    .id("1")           // 指定文档 ID
    .doc(new Product("Laptop", 899.99, "Electronics")), // 更新后的文档内容
    Product.class       // 指定文档类型
);
System.out.println("文档更新成功！");

删除文档

// 删除 ID 为 "1" 的文档
client.delete(d -> d
    .index("products")  // 指定索引名称
    .id("1")           // 指定文档 ID
);
System.out.println("文档删除成功！");

3.5 批量操作

// 创建批量请求构建器
BulkRequest.Builder br = new BulkRequest.Builder();

// 添加第一个文档
br.operations(op -> op
    .index(i -> i
        .index("products")  // 指定索引名称
        .id("2")           // 指定文档 ID
        .document(new Product("Phone", 599.99, "Electronics")) // 文档内容
    )
);

// 添加第二个文档
br.operations(op -> op
    .index(i -> i
        .index("products")  // 指定索引名称
        .id("3")           // 指定文档 ID
        .document(new Product("Tablet", 299.99, "Electronics")) // 文档内容
    )
);

// 执行批量操作
client.bulk(br.build());
System.out.println("批量操作完成！");

4. 搜索与聚合

4.1 查询 DSL

简单匹配查询

// 查询 "products" 索引中 name 字段包含 "Laptop" 的文档
SearchResponse<Product> response = client.search(s -> s
    .index("products")  // 指定索引名称
    .query(q -> q       // 定义查询条件
        .match(m -> m   // 匹配查询
            .field("name")  // 指定字段
            .query("Laptop") // 查询值
        )
    ),
    Product.class      // 指定返回的文档类型
);

// 输出查询结果
for (Hit<Product> hit : response.hits().hits()) {
    System.out.println("查询结果：" + hit.source().getName());
}

复合查询（Bool Query）

// 查询 "products" 索引中 category 为 "Electronics" 且 price 大于等于 500 的文档
SearchResponse<Product> response = client.search(s -> s
    .index("products")  // 指定索引名称
    .query(q -> q       // 定义查询条件
        .bool(b -> b    // 布尔查询
            .must(m -> m.match(t -> t.field("category").query("Electronics"))) // 必须匹配的条件
            .filter(f -> f.range(r -> r.field("price").gte(JsonData.of(500)))) // 过滤条件
    ),
    Product.class      // 指定返回的文档类型
);

// 输出查询结果
for (Hit<Product> hit : response.hits().hits()) {
    System.out.println("查询结果：" + hit.source().getName() + ", " + hit.source().getPrice());
}

聚合分析

// 对 "products" 索引中的 price 字段进行平均值聚合
SearchResponse<Product> response = client.search(s -> s
    .index("products")  // 指定索引名称
    .aggregations("avg_price", a -> a // 定义聚合
        .avg(avg -> avg.field("price")) // 计算 price 字段的平均值
    ),
    Product.class      // 指定返回的文档类型
);

// 获取聚合结果
double avgPrice = response.aggregations().get("avg_price").avg().value();
System.out.println("平均价格：" + avgPrice);

5. 性能优化

5.1 分片与副本策略

• 分片数在创建索引后不可修改，需提前规划。
• 副本数可动态调整：PUT /products/_settings { "number_of_replicas": 2 }。

5.2 写入优化

• 使用批量 API 减少请求次数。
• 调整 refresh_interval 降低刷新频率。

5.3 查询优化

• 避免通配符查询（*）。
• 使用 keyword 类型做精确匹配。

6. 集群管理

6.1 查看集群健康状态

// 获取集群健康状态
HealthResponse response = client.cluster().health();
System.out.println("集群状态：" + response.status());

7. 学习资源

• 官方文档: https://www.elastic.co/guide
• 书籍: 《Elasticsearch 权威指南》
• 在线课程: Udemy 或 Coursera 上的 Elasticsearch 专项课程

通过本文档，您可以系统掌握 Elasticsearch 的核心功能与 Java 客户端操作。建议结合实际项目需求，进一步练习和优化代码。