Elasticsearch搜索引擎(二)

---

前言

ES官方提供了各种不同语言的客户端用来操作ES，这些客户端的本质就是组装DSL语句，通过http请求发送给ES。

---

一、RestAPI

1. 初始化 RestClient

在ES提供的API中，与ES一切交互都封装在一个名为 RestHighLevelClient 的类中，必须完成这个对象的初始化，建立与ES的连接。

RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.74.129:9200")
));

2. 创建索引库

由于要实现对商品的搜索，所以我们需要将商品添加到ES中，不过需要根据搜索业务的需求来设定索引库的结构，而不是把MySQL数据全部写到ES。

首先要创建 Mapping映射，然后创建索引，创建索引主要分为三步：

• 创建Request对象：因为是创建索引库的操作，因此Request是 CreateIndexRequest。
• 添加请求参数：其实就是Json格式的Mapping映射参数。因为json字符串很长，可以定义一个静态字符常量MAPPING_TEMPLATE，使代码看起来更优雅。
• 发送请求：client.indices()方法的返回值是IndicesClient类型，封装了所有与索引库操作相关的方法。例如创建索引、删除索引、判断索引是否存在等。

3. 删除索引库

• 创建Request对象，这次是DeleteIndexRequest对象。
• 准备参数。这里无参，因此省略。
• 发送请求，该用delete方法。

@Test
void testDeleteIndex() throws IOException {// 1.创建Request对象DeleteIndexRequest request = new DeleteIndexRequest("items");// 2.发送请求client.indices().delete(request, RequestOptions.DEFAULT);
}

4. 判断索引库是否存在

• 创建Request对象。这次是GetIndexRequest对象
• 准备参数。这里是无参，直接省略
• 发送请求。改用exists方法

@Test
void testExistsIndex() throws IOException {// 1.创建Request对象GetIndexRequest request = new GetIndexRequest("items");// 2.发送请求boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);// 3.输出System.err.println(exists ? "索引库已经存在！" : "索引库不存在！");
}

二、RestClient操作文档

1.新增文档

我们需要将数据库中的商品信息导入ES中，由于索引库结构与数据库结构还存在一些差异，因此我们要定义一个索引库结构对应的实体ItemDoc。

接下来与索引库操作的API非常相似，也是分三步走，变化的地方在于：这里直接使用client.xxx() 的API，不再需要client.indices()了。

由于导入了真实数据，除了三步走之外，还需要做几点工作：

• 根据id查询商品数据：商品数据来自于数据库，我们需要先查询处理，得到Item对象。
• 将Item对象封装为ItemDoc，即转换为文档类型。
• ItemDoc需要序列化为JSON

@Test
void testAddDocument() throws IOException {// 1.根据id查询商品数据Item item = itemService.getById(100002644680L);// 2.转换为文档类型ItemDoc itemDoc = BeanUtil.copyProperties(item, ItemDoc.class);// 3.将ItemDTO转jsonString doc = JSONUtil.toJsonStr(itemDoc);// 1.准备Request对象IndexRequest request = new IndexRequest("items").id(itemDoc.getId());// 2.准备Json文档request.source(doc, XContentType.JSON);// 3.发送请求client.index(request, RequestOptions.DEFAULT);
}

总体流程如下：

• 1）根据id查询商品数据Item
• 2）将Item封装为ItemDoc
• 3）将ItemDoc序列化为JSON
• 4）创建IndexRequest，指定索引库名和id
• 5）准备请求参数，也就是JSON文档
• 6）发送请求

2.查询文档

查询的目的是得到结果，解析为ItemDoc，就是对JSON作反序列化

@Test
void testGetDocumentById() throws IOException {// 1.准备Request对象GetRequest request = new GetRequest("items").id("100002644680");// 2.发送请求GetResponse response = client.get(request, RequestOptions.DEFAULT);// 3.获取响应结果中的sourceString json = response.getSourceAsString();ItemDoc itemDoc = JSONUtil.toBean(json, ItemDoc.class);System.out.println("itemDoc= " + ItemDoc);
}

3. 删除文档

@Test
void testDeleteDocument() throws IOException {// 1.准备Request，两个参数，第一个是索引库名，第二个是文档idDeleteRequest request = new DeleteRequest("item", "100002644680");// 2.发送请求client.delete(request, RequestOptions.DEFAULT);
}

4. 修改文档

修改主要有两种方式：

• 全量修改：本质是先根据id删除，再新增
• 局部修改：修改文档中的指定字段值

由于在RestClient的API中，全量修改与新增的API完全一致，判断依据是ID：

• 如果新增时，ID已经存在，则修改
• 如果新增时，ID不存在，则新增

因此我们主要关注局部修改的API即可。与之前类似，也是三步走：

• 1）准备Request对象。这次是修改，所以是UpdateRequest。
• 2）准备参数。也就是JSON文档，里面包含要修改的字段。
• 3）更新文档。这里调用client.update()方法。

@Test
void testUpdateDocument() throws IOException {// 1.准备RequestUpdateRequest request = new UpdateRequest("items", "100002644680");// 2.准备请求参数request.doc("price", 58800,"commentCount", 1);// 3.发送请求client.update(request, RequestOptions.DEFAULT);
}

5. 批量导入文档

在实际项目中，数据库的商品数据会达到数十万甚至数百万条，我们如果要将这些数据导入索引库，肯定不能逐条导入，而是采用批处理方案。常见的方案有：

• 利用Logstash批量导入
  • 需要安装Logstash
  • 对数据的再加工能力较弱
  • 无需编码，但要学习编写Logstash导入配置
• 利用JavaAPI批量导入
  • 需要编码，但基于JavaAPI，学习成本低
  • 更加灵活，可以任意对数据做再加工处理后写入索引库

接下来，我们利用JavaAPI实现批量文档导入。

批处理与前面讲的文档的CRUD步骤基本一致：

• 创建Request，但这次用的是BulkRequest
• 准备请求参数
• 发送请求，这次要用到client.bulk()方法

当我们要导入商品数据时，由于商品数量达到数十万，因此不可能一次性全部导入。建议采用循环遍历方式，每次导入1000条左右的数据。

@Test
void testLoadItemDocs() throws IOException {// 分页查询商品数据int pageNo = 1;int size = 1000;while (true) {Page<Item> page = itemService.lambdaQuery().eq(Item::getStatus, 1).page(new Page<Item>(pageNo, size));// 非空校验List<Item> items = page.getRecords();if (CollUtils.isEmpty(items)) {return;}log.info("加载第{}页数据，共{}条", pageNo, items.size());// 1.创建RequestBulkRequest request = new BulkRequest("items");// 2.准备参数，添加多个新增的Requestfor (Item item : items) {// 2.1.转换为文档类型ItemDTOItemDoc itemDoc = BeanUtil.copyProperties(item, ItemDoc.class);// 2.2.创建新增文档的Request对象request.add(new IndexRequest().id(itemDoc.getId()).source(JSONUtil.toJsonStr(itemDoc), XContentType.JSON));}// 3.发送请求client.bulk(request, RequestOptions.DEFAULT);// 翻页pageNo++;}
}

---