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转json
    String 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.获取响应结果中的source
    String json = response.getSourceAsString();
    
    ItemDoc itemDoc = JSONUtil.toBean(json, ItemDoc.class);
    System.out.println("itemDoc= " + ItemDoc);
}

3. 删除文档

@Test
void testDeleteDocument() throws IOException {
    // 1.准备Request，两个参数，第一个是索引库名，第二个是文档id
    DeleteRequest 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.准备Request
    UpdateRequest 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.创建Request
        BulkRequest request = new BulkRequest("items");
        // 2.准备参数，添加多个新增的Request
        for (Item item : items) {
            // 2.1.转换为文档类型ItemDTO
            ItemDoc 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++;
    }
}