Elastic Stack 实战教程 5：Elasticsearch Java API Client 开发

Se7en258

发布于 2023-02-26 16:02:55

2.1K0

发布于 2023-02-26 16:02:55

本系列 Elastic Stack 实战教程总共涵盖 5 个实验，目的是帮助初学者快速掌握 Elastic Stack 的基本技能。云起实验室在线体验地址：https://developer.aliyun.com/adc/scenarioSeries/24e7a7a4d56741d0bdcb3ee73c9c22f1

实验 1：Elastic Stack 8 快速上手
实验 2：ILM 索引生命周期管理
实验 3：快照备份与恢复
实验 4：使用 Fleet 管理 Elastic Agent 监控应用
实验 5：Elasticsearch Java API Client 开发

1 Elasticsearch Jave Client 介绍

Elastic 在 7.16 版本（2021年12月8日）推出了 Elasticsearch Java API Client。在此之前，我们通常使用 High Level REST Client 进行开发，但是 High Level REST Client 存在几个缺陷：

1.引入了许多非必要相关的依赖，并且暴露了很多服务器的内部接口。
2.一致性差，需要大量的维护工作。
3.没有集成 json/object 类型映射，需要自己借助字节缓存区实现。

Elasticsearch Java API Client 有以下三个典型特点：

1.对象构造基于建造者模式（Builder Pattern），使用多个简单的对象一步一步构建成一个复杂的对象，增强了客户端代码的可用性和可读性。
2.使用 Lambda 构建嵌套对象，使得编写干净、富有表现力的 DSL 变得容易。
3.应用程序类能自动映射为 Mapping。

本实验中我们将使用 Elasticsearch Java API Client 进行开发，实现常用的 CRUD 操作。

2 启动实验环境

首先执行以下命令修改系统参数以满足 Elasticsearch 的运行条件。

# 增加进程可使用的最大内存映射区域数
cat >> /etc/sysctl.conf << EOF
vm.max_map_count=262144
EOF

sysctl -p

# 增加进程可使用的最大文件描述符数量
cat >> /etc/security/limits.conf  << EOF
elastic - nofile  65535
EOF

ulimit -n 65535

为了方便实验，本节采用 Docker Compose 的方式快速部署 Elasticsearch 集群。执行如下命令，安装 Docker 和 Docker Compose。

apt install -y git
git https://gitee.com/cr7258/elastic-lab.git
curl -sSL https://get.daocloud.io/docker | sh
apt install -y docker-compose

执行如下命令，获取 docker-compose.yml 配置文件，并在后台启动 Elasticsearch 集群。

cd elastic-lab/5_java_develop/
docker-compose up -d

执行 docker-compose ps 命令查看容器运行状态，其中 5java_develop_setup_1 容器是用于创建证书以及设置 elastic 和 kibana_system 用户密码的，执行完毕后会自动退出，我们需要确保其他容器处于 Up 状态。

3 本地环境准备

本地需要提前安装好以下工具：

Git
JDK
IntelliJ IDEA

在本地电脑执行以下命令，克隆代码。

git clone https://gitee.com/cr7258/elastic-lab.git

打开 IntelliJ IDEA 工具，点击 OPEN。

选择打开 elastic-lab -> 5_java_develop -> elasticsearch_java_api_client 项目目录。注意在 IDEA 中打开的是 elasticsearch_java_api_client 目录，这样 IDEA 才可以正确识别出这是一个 gradle 项目，gradle 所需的文件 IDEA 会自动进行下载。

打开 IntelliJ IDEA 设置 -> Build, Execution, Deployment -> Build Tools -> Gradle，将 Run tests using 改为 IntelliJ IDEA。

3 引入依赖

在本实验中，将使用 gradle 项目进行演示，Elasticsearch 所需的依赖如下。

dependencies {
    implementation 'co.elastic.clients:elasticsearch-java:8.2.3'
    implementation 'com.fasterxml.jackson.core:jackson-databind:2.12.3'
}

另外为了方便我们编写代码，本实验中还使用了 lombok, spring boot 等组件，完整的 gradle 依赖文件请参见 build.gradle。

4 连接集群

创建 Elasticsearch Java API Client 主要分为以下 3 步：

// 1.创建 low-level client
RestClient restClient = RestClient.builder(
    new HttpHost("localhost", 9200)).build();

// 2.创建 transport 
ElasticsearchTransport transport = new RestClientTransport(
    restClient, new JacksonJsonpMapper());

// 3.创建 api client
ElasticsearchClient client = new ElasticsearchClient(transport);

由于我们部署的 Elasticsearch 8.x 集群设置了用户名和密码，并且启用了 HTTPS 加密，因此我们在构建 low-level client 的使用还需要进行相应的设置。

RestClientBuilder builder = RestClient.builder(httpHost)  
        .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder  
                .setSSLContext(sslContext)      // 设置 SSL 加密通信的方式  
                .setDefaultCredentialsProvider(credentialsProvider) // 设置用户名密码  
                .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)); // 不验证 SSL 证书主机名

连接 Elaticsearch 的完整代码可以在 config 目录中获取。

以下两个文件用户需要根据实际情况进行配置，文件路径在 elastic-lab/5_java_develop/elasticsearch_java_api_client/src/main/resources。

application.yml：设置 Elasticsearch 的连接信息。其中用户名和密码是在 elastic-lab/5_java_develop/.env 文件中设置的

elasticsearch:  
  schema: https  
  address: <公网 IP>:9200 # 需要修改  
  username: elastic  
  password: elastic123

ca.crt：由于我们部署的 Elasticsearch 使用的是自签名的 CA，需要设置信任的 CA 证书。执行以下命令获取证书内容，并保存到 ca.crt 文件中。

docker exec -it 5javadevelop_es01_1 cat config/certs/ca/ca.crt

5 创建实体类

创建一个 Product.java 文件用于保存商品字段。这里使用了 lombok 的 3 个注解会帮助我们生成需要的构造方法以及 Getter, Setter 方法。

import lombok.AllArgsConstructor;  
import lombok.Data;  
import lombok.NoArgsConstructor;  
  
/**  
 * @author chengzw  
 * @description 商品实体类  
 * @since 2022/8/1  
 */  
@Data  
@AllArgsConstructor  
@NoArgsConstructor  
public class Product {  
    String id;  
    String name;  
    double price;  
}

完成准备工作后，接下来我们将尝试对 Elasticsearch 执行增删改查操作。

6 写入文档

Elasticsearch Java API Client 提供了两种索引文档的方式：

1.提供一个应用对象，Elasticsearch Client 负责将对象映射为 JSON。
2.直接提供原始的 JSON 数据。

如下所示，我们创建了一个 Product 对象，将数据写入索引 products 中，并使用商品 id 作为 doc id。Elasticsearch Client 会自动将 Product 对象转换为 JSON 数据，然后向 Elasticsearch 发送索引请求。

Product product = new Product("sn10001", "computer", 9999.99);  
IndexResponse response = esClient.index(i -> i  
        .index("products")  // 索引名
        .id(product.getId())  // doc id
        .document(product)    // 数据
);

在 IndexingTest.java[1] 文件中可以找到完整代码，点击 indexObject() 方法左边的运行按钮执行上述代码。执行完成后，在最下方可以看到响应结果，可以看到数据已经成功写入 products 索引了。

浏览器输入 http://<ESC 公网 IP>:5601 访问 Kibana 界面。输入用户名 elastic，密码 elastic123，点击 Login in。其中密码是在 elastic-lab/5_java_develop/.env 文件中设置的。

点击 Management -> Dev Tools -> Console，打开 Kibana Console 界面。执行以下命令，查询 products 索引的数据，可以看到 id 为 sn10001 的文档已经被成功写入了。

GET products/_search

# 返回结果
{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "products",
        "_id" : "sn10001",
        "_score" : 1.0,
        "_source" : {
          "id" : "sn10001",
          "name" : "computer",
          "price" : 9999.99
        }
      }
    ]
  }
}

当你要索引的数据来自外部时，为这些数据创建实体类可能会很麻烦，这时候你可以使用 withJson() 方法将原始的 JSON 数据作为索引请求的文档内容。

Reader input = new StringReader(  
        "{'id': 'sn10003', 'name': 'television', 'price': 5500.5}"  
                .replace('\'', '"'));  
  
IndexRequest<JsonData> request = IndexRequest.of(i -> i  
        .index("products")  
        .withJson(input)  
);  
  
IndexResponse response = esClient.index(request);

在 IndexingTest.java[2] 文件中可以找到完整代码，点击 indexWithJson() 方法左边的运行按钮执行上述代码。

为了提升写入数据的效率，我们可以提前准备好一批数据，使用 bulk API 在一次网络请求中将数据批量写入 Elasticsearch 中。

List<Product> products = new ArrayList<>();  
products.add(new Product("sn10004", "T-shirt", 100.5));  
products.add(new Product("sn10005", "phone", 8999.9));  
products.add(new Product("sn10006", "ipad", 6555.5));  
  
BulkRequest.Builder br = new BulkRequest.Builder();  
for (Product product : products) {  
    br.operations(op -> op  
            .index(idx -> idx  
                    .index("products")  
                    .id(product.getId())  
                    .document(product)  
            )    );  
}  
  
BulkResponse response = esClient.bulk(br.build());

在 IndexingTest.java[3] 文件中可以找到完整代码，点击 indexBulkObject() 方法左边的运行按钮执行上述代码。

7 查询文档

使用 get 请求可以根据 id 来获取文档。get 请求有两个参数：

第一个参数是实际请求，在下面用 fluent DSL 构建。
第二个参数是希望将文档的 JSON 映射到的类。

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

在 SearchingTest.java[4] 文件中可以找到完整代码，点击 getById() 方法左边的运行按钮执行上述代码。

接下来我们尝试使用 match 查询语句来搜索 name 字段中包含 computer 关键字的文档，DSL 语句如下所示，我们可以先在 Kibana 上执行以下命令查看结果。

GET products/_search
{
  "query": {
    "match": {
      "name": {
        "query": "computer"
      }
    }
  }
}

# 返回结果
{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.4877305, 
    "hits" : [
      {
        "_index" : "products",
        "_id" : "sn10001",
        "_score" : 1.4877305, # 文档评分
        "_source" : { # 文档内容
          "id" : "sn10001",
          "name" : "computer",
          "price" : 9999.99
        }
      }
    ]
  }
}

我们可以使用以下代码实现上述的全文查询，可以看出 search 是 Elasticsearch Java API Client 的一大优势，使用 Lambda 构建嵌套对象，大大简化了代码量，并且增强了代码的可读性。

String searchText = "computer";  
SearchResponse<Product> response = esClient.search(s -> s  
                .index("products")  
                .query(q -> q  
                        .match(t -> t  
                                .field("name")  
                                .query(searchText)  
                        )                )        , Product.class);

在 SearchingTest.java[5] 文件中可以找到完整代码，点击 search() 方法左边的运行按钮执行上述代码。

使用聚合查询可以对索引中的数据进行统计，如下所示，我们使用 terms 聚合查询来计算每种商品的数量。由于 name 字段的类型是 text 类型，默认情况下 text 类型是不允许进行聚合操作的，因此这里使用 name 字段的 keyword 类型的子字段 keyword 来进行聚合查询。

GET products/_search
{
  "size": 0, // 不返回文档 
  "aggs": {
    "product-term": {
      "terms": {
        "field": "name.keyword" 
      }
    }
  }
}


# 返回结果
{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "product-term" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "T-shirt",
          "doc_count" : 1
        },
        {
          "key" : "computer",
          "doc_count" : 1
        },
        {
          "key" : "ipad",
          "doc_count" : 1
        },
        {
          "key" : "phone",
          "doc_count" : 1
        },
        {
          "key" : "television",
          "doc_count" : 1
        }
      ]
    }
  }
}

我们可以使用以下代码实现上述的聚合查询，这个例子是一个分析类型的聚合，不需要返回文档内容，因此可以将 size 设置为零，并将搜索结果的目标类设置为 Void.class。

SearchResponse<Void> response = esClient.search(b -> b  
        .index("products")  
        .size(0)  
        .aggregations("product-term", a -> a  
                .terms(t -> t  
                        .field("name.keyword"))  
        ), Void.class);

在 AggregationsTest.java[6] 文件中可以找到完整代码，点击 productTerm() 方法左边的运行按钮执行上述代码。

8 删除文档

删除文档通常有两种方式：

1.指定 doc id 删除单条文档。
2.通过 deleteByQuery 查询匹配删除单条或多条文档。

如下所示，指定删除索引 products 中 id 为 sn10005 的文档。

DeleteResponse response = esClient.delete(d -> d  
        .index("products")  
        .id("sn10005"));

在 DeletingTest.java[7] 文件中可以找到完整代码，点击 deleteById() 方法左边的运行按钮执行上述代码。

批量删除文档可以使用 deleteByQuery。如下所示，删除索引 products 中所有 name 字段的值是 ipad 的文档。

String searchText = "ipad";
DeleteByQueryResponse response = esClient.deleteByQuery(d ->  
        d.index("products")  
                .query(q -> q  
                        .match(t -> t  
                                .field("name")  
                                .query(searchText))));

在 DeletingTest.java[8] 文件中可以找到完整代码，点击 deleteByQuery() 方法左边的运行按钮执行上述代码。

9 更新文档

和删除文档类似，更新文档通常也有两种方式：

1.指定 doc id 更新单条文档。
2.通过 updateByQuery 查询匹配更新单条或多条文档。

我们在更新文档的时候可以在请求体的 doc 参数中指定要更新的字段内容。如下所示，更新 doc id 为 sn10001 的文档的 price 字段值为 7777.77。

POST products/_update/sn10001
{
  "doc": {
    "price": 7777.77
  }
}

我们可以使用以下代码实现上述的单条更新操作，首先创建一个 Product 对象，然后设置对象的 price 字段值为 7777.77，然后使用 update 请求设置更新的索引名和 doc id，并传入 Product 对象。

Product product = new Product();  
product.setPrice(7777.77);  
UpdateResponse<Product> response = esClient.update(u -> u  
        .index("products")  
        .id("sn10005")  
        .doc(product), Product.class);

在 UpdatingTest.java[9] 文件中可以找到完整代码，点击 updateById() 方法左边的运行按钮执行上述代码。

在 Kibana 查询 doc id 为 sn10001 的文档，可以看到 price 字段值已经被成功修改为 7777.77 了。

GET products/_doc/sn10001

# 返回结果
{
  "_index" : "products",
  "_id" : "sn10001",
  "_version" : 2,
  "_seq_no" : 7,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : "sn10001",
    "name" : "computer",
    "price" : 7777.77
  }
}

批量更新文档可以使用 updateByQuery。如下所示，将索引 products 中所有 name 字段的值是 T-shirt 的文档的 price 字段值增加 1000，使用 painless script 可以实现更加复杂的操作。

POST products/_update_by_query
{
  "query": {
    "match": {
      "name": "T-shirt"
    }
  },
  "script": {
    "source": """
    ctx._source.price += 1000
    """,
    "lang": "painless"
  }
}

我们可以使用以下代码实现上述的批量更新操作，在 InlineScript 中可以设置批量更新使用的 painless script。

UpdateByQueryResponse response = esClient.updateByQuery(u -> u  
        .index("products")  
        .script(s -> s  
                .inline(InlineScript.of(i -> i.lang("painless").source("ctx._source.price += 1000"))))  
        .query(q -> q  
                .match(m -> m  
                        .field("name")  
                        .query("T-shirt"))));

在 UpdatingTest.java[10] 文件中可以找到完整代码，点击 updateByQuery() 方法左边的运行按钮执行上述代码。

查询 name 字段值是 T-shirt 的文档，可以看到 price 字段值已经增加 1000 了。

GET products/_search
{
  "query": {
    "match": {
      "name": "T-shirt"
    }
  }
}

# 返回结果
{
  "took" : 3,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.2199391,
    "hits" : [
      {
        "_index" : "products",
        "_id" : "sn10004",
        "_score" : 1.2199391,
        "_source" : {
          "price" : 1100.5,
          "name" : "T-shirt",
          "id" : "sn10004"
        }
      }
    ]
  }
}

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2022-12-15，如有侵权请联系 cloudcommunity@tencent.com 删除

api