Java Rest Client 操作elasticsearch详解

一、概述

在elasticsearch7.x中，主要还是使用java rest client的相关api来进行开发。

而8.x之后官方则推荐使用对lambda表达式支持更强的java client来进行开发。

这里主要还是说明7.x版本的api。

7.12 版本的文档

https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.12/java-rest-high-getting-started.html#java-rest-high-getting-started

二、SpringBoot项目中引入Maven依赖

<!-- RestHighLevelClient 坐标 -->
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
</dependency>

<!-- SpringBoot默认的ES版本是7.6.2，因此需要覆盖默认的ES版本，改为自己需要的版本 -->
<properties>
    <java.version>1.8</java.version>
    <elasticsearch.version>7.12.1</elasticsearch.version>
</properties>

三、初始化RestHighLevelClient的方式

在elasticsearch提供的API中，与elasticsearch的交互都封装在一个名为RestHighLevelClient的类中。

必须先完成这个对象的初始化，建立与elasticsearch的连接。

// 初始化RestHighLevelClient对象，可传入HttpHost对象，指定能够操作的elasticsearch
// 这一步就完成了和elasticsearch的连接
RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
        HttpHost.create("http://192.168.137.102:9200")
));

RestClient.builder()方法可以接收一个可变参数，也就是说它可以传入多个HttpHost对象

四、操作elasticsearch索引库

4.1 封装单元测试类

package cn.itcast.hotel;

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;

@SpringBootTest
public class HotelIndexTest {
    // 声明RestHighLevelClient对象
    private RestHighLevelClient client;			

    @BeforeEach
    void setUp() {
        // 初始化RestHighLevelClient对象
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.150.101:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        // 关闭RestHighLevelClient对象和elasticsearch的连接，并释放占用的资源
        this.client.close();
    }
}

之后就在这个单元测试类中添加有@Test修饰的方法来测试elasticsearch中的各种增删改查操作

4.2 创建索引库

@Test
void testCreateIndex() throws IOException {
    // 1.初始化创建索引库的请求对象，索引库的名称为参数hotel
    // 相当于PUT /hotel
    CreateIndexRequest request = new CreateIndexRequest("hotel");
    // 2.准备请求参数
    // MAPPING_TEMPLATE 是静态常量字符串，内容是创建索引库的DSL语句
    // XContentType.JSON 数据类型为json类型
    request.source(MAPPING_TEMPLATE, XContentType.JSON);
    // 3.发送请求
    client.indices().create(request, RequestOptions.DEFAULT);
}

索引库创建的DSL语句：

public class HotelIndexConstants {
    public static final String MAPPING_TEMPLATE = "{\n" +
            "  \"mappings\": {\n" +
            "    \"properties\": {\n" +
            "      \"id\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"name\": {\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"address\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"price\": {\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"score\": {\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"brand\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"city\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"starName\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"business\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"pic\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"location\": {\n" +
            "        \"type\": \"geo_point\"\n" +
            "      },\n" +
            "      \"all\": {\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\"\n" +
            "      }\n" +
            "    }\n" +
            "  }\n" +
            "}";
}

三个步骤的具体解析如下：

• 创建Request对象，因为是创建索引库的操作。

  因此使用CreateIndexRequest，它可以接收一个字符串参数作为索引库的名称。

• 添加请求参数，其实就是DSL语句中的JSON参数部分。

  source方法的两个参数分别为DSL语句的字符串和DSL语句字符串的格式。

  JSON字符串很长，这里是定义了静态字符串常量MAPPING_TEMPLATE，让代码看起来更加优雅。

• 发送请求，client.indices()的返回值是IndicesClient类型，封装了所有与索引库操作有关的方法。

  后面的create方法就是去创建索引库的方法，接收一个请求对象和请求选项参数(选项参数一般默认即可)。

4.3 查询索引库

@Test
void testGetIndex() throws IOException {
    // 1.初始化查询索引库的请求对象   索引库的名称为参数hotel
    // 相当于GET /hotel
    GetIndexRequest request = new GetIndexRequest("hotel");
    // 2.使用get()方法获取索引库的一些信息
    GetIndexResponse getIndexResponse = client.indices().get(request, 				          															RequestOptions.DEFAULT);
    System.out.println(getIndexResponse);
    System.out.println(getIndexResponse.getAliases());
    System.out.println(getIndexResponse.getIndices());
    System.out.println(getIndexResponse.getMappings());
    System.out.println(getIndexResponse.getDefaultSettings());
    System.out.println(getIndexResponse.getSettings());
}

4.4 判断索引库是否存在

@Test
void testExistsIndex() throws IOException {
    // 1.初始化查询索引库的请求对象   索引库的名称为参数hotel
    GetIndexRequest request = new GetIndexRequest("hotel");
    // 2.使用exists方法判断索引库是否存在
    boolean isExists = client.indices().exists(request, RequestOptions.DEFAULT);

    System.out.println(isExists ? "存在" : "不存在");
}

4.5 删除索引库

@Test
void testDeleteIndex() throws IOException {
    // 1.初始化删除索引库的请求对象   索引库的名称为参数hotel
    DeleteIndexRequest request = new DeleteIndexRequest("hotel");
    // 2.使用delete方法删除索引库
    client.indices().delete(request, RequestOptions.DEFAULT);
}

4.6 小结：

一般来说都不会有用java代码创建、查询、删除、更新索引库的需求，都是用kibana控制台去操作

在java代码中最多用只会判断索引库是否存在，也就是exists()方法会常用一些

五、操作elasticsearch文档

5.1 封装单元测试类

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.service.IHotelService;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

@SpringBootTest
public class HotelDocumentTest {
    // 业务接口基于mybatis-plus
    @Autowired
    private IHotelService hotelService;

    private RestHighLevelClient client;

    @BeforeEach
    void setUp() {
        // 初始化RestHighLevelClient对象
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.137.129:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        // 关闭RestHighLevelClient对象和elasticsearch的连接，并释放占用的资源
        this.client.close();
    }
}

之后就在这个单元测试类中添加有@Test修饰的方法来测试elasticsearch中的各种增删改查操作

5.2 新增文档数据

@Test
void testAddDocument() throws IOException {
    // 1.使用mybatis-plus 查询数据库hotel数据
    Hotel hotel = hotelService.getById(61083L);
    // 2.转换为HotelDoc
    HotelDoc hotelDoc = new HotelDoc(hotel);
    // 3.转JSON
    String json = JSON.toJSONString(hotelDoc);

    // 1.创建IndexRequest文档请求对象
    IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());
    // 2.准备请求参数DSL，其实就是文档的JSON字符串，文档类型为json
    request.source(json, XContentType.JSON);
    // 3.使用index方法发送新增文档的请求
    client.index(request, RequestOptions.DEFAULT);
}

关于两个实体类的说明：

// mysql数据库中的Hotel实体类
@Data
@TableName("tb_hotel")
public class Hotel {
    @TableId(type = IdType.INPUT)
    private Long id;
    private String name;
    private String address;
    private Integer price;
    private Integer score;
    private String brand;
    private String city;
    private String starName;
    private String business;
    private String longitude;
    private String latitude;
    private String pic;
}

// 对应elasticsearch文档数据的实体类
@Data
@NoArgsConstructor
public class HotelDoc {
    private Long id;
    private String name;
    private String address;
    private Integer price;
    private Integer score;
    private String brand;
    private String city;
    private String starName;
    private String business;
    private String location;
    private String pic;

    public HotelDoc(Hotel hotel) {
        this.id = hotel.getId();
        this.name = hotel.getName();
        this.address = hotel.getAddress();
        this.price = hotel.getPrice();
        this.score = hotel.getScore();
        this.brand = hotel.getBrand();
        this.city = hotel.getCity();
        this.starName = hotel.getStarName();
        this.business = hotel.getBusiness();
        this.location = hotel.getLatitude() + ", " + hotel.getLongitude();
        this.pic = hotel.getPic();
    }
}

小结：

上面的操作相当于以下DSL语句：

POST /{索引库名}/_doc/1
{
    "name": "Jack",
    "age": 21
}

使用java rest client操作文档与创建索引库类似，主要有三步：

• 创建IndexRequest对象
• 准备请求参数，也就是DSL中的JSON文档
• 发送请求

不同的地方在于，这里可以直接使用client.xxx()的API，不再需要client.indices()了。

5.3 查询文档

@Test
void testGetDocumentById() throws IOException {
    // 1.创建GetRequest文档请求对象
    GetRequest request = new GetRequest("hotel", "61083");
    // 2.get方法发送请求
    GetResponse response = client.get(request, RequestOptions.DEFAULT);
    // 3.解析响应结果为json字符串
    String json = response.getSourceAsString();
	// 4.把json转换为实体类对象（反序列化）
    HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
    System.out.println("hotelDoc = " + hotelDoc);
}

小结：

上面的操作相当于以下DSL语句：

GET /hotel/_doc/{id}

使用java rest client操作文档，主要也有三步：

• 准备GetRequest对象
• 调用get方法发送请求
• 解析结果，就是对JSON做反序列化

5.4 删除文档

@Test
void testDeleteDocumentById() throws IOException {
    // 1.创建DeleteRequest文档请求对象
    DeleteRequest request = new DeleteRequest("hotel", "61083");
    // 2.调用delete方法发送删除请求
    client.delete(request, RequestOptions.DEFAULT);
}

小结：

上面的操作相当于以下DSL语句：

DELETE /hotel/_doc/{id}

，主要也有三步：

• 准备GetRequest对象
• 调用get方法发送请求
• 解析结果，就是对JSON做反序列化

使用java rest client删除文档与查询相比，请求方式从DELETE变成GET，代码依然是三步走：

• 准备DeleteRequest对象，要指定索引库名和文档id
• 准备参数，无参，因为索引库名称和文档id已经可以唯一确定一个文档库了。
• 发送请求。因为是删除，所以是client.delete()方法

5.5 修改文档

@Test
void testUpdateById() throws IOException {
    // 1.创建UpdateRequest文档请求对象
    UpdateRequest request = new UpdateRequest("hotel", "61083");
    // 2.准备参数，调用doc方法更新文档
    // 这里的参数格式很特殊： "price"是字段名称， "870"是字段的值，两者使用逗号分隔
    request.doc(
        "price", "870",
        "starName", "黄金"
    );
    // 2.调用update方法发送更新文档数据的请求
    client.update(request, RequestOptions.DEFAULT);
}

修改文档有两种方式：

• 全量修改：本质是先根据id删除，再新增

  DSL语句如下：

  PUT /索引库名称/_doc/文档id
  {
    "字段1": "值1",
    "字段2": "值2",
    // ... 
  }
  
  

• 增量修改：修改文档中的指定字段值

  DSL语句如下：

  POST /索引库名称/_doc/文档id
  {
    "doc": {
       "字段名": "新的值",
    }
  }
  

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

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

一般来说，增量修改使用的会多一些。

5.6 批量从数据库导入数据进文档

1）步骤如下：

• 利用mybatis-plus查询Hotel数据

• 将查询到的Hotel数据转换为文档类型数据（HotelDoc）

• 利用JavaRestClient中的BulkRequest批处理，实现批量新增文档

2）语法说明

批量处理对象BulkRequest，其本质就是将多个普通的CRUD请求组合在一起发送。

其中提供了一个add方法，用来添加其他请求。

可以看到，能添加的请求类型包括：

• IndexRequest，也就是新增
• UpdateRequest，也就是修改
• DeleteRequest，也就是删除

也就是说对于插入、更新、删除请求都是可以进行批量操作的

@Test
void testBulkRequest() throws IOException {
    // 1.批量查询酒店数据
    List<Hotel> hotels = hotelService.list();
    // 2.创建BulkRequest文档请求对象
    BulkRequest request = new BulkRequest();
    // 3.准备参数，添加多个新增的Request对象
    for (Hotel hotel : hotels) {
        // 3.1.转换为文档类型HotelDoc
        HotelDoc hotelDoc = new HotelDoc(hotel);
        // 3.2.创建新增文档的Request对象
        request.add(new IndexRequest("hotel")
                    .id(hotelDoc.getId().toString())
                    .source(JSON.toJSONString(hotelDoc), XContentType.JSON));
    }
    // 4.调用bulk方法 发送请求
    client.bulk(request, RequestOptions.DEFAULT);
}

六、elasticsearch各类文档查询

6.1 match_all全文检索查询

@Test
void testMatchAll() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备请求参数
    request.source().query(QueryBuilders.matchAllQuery());
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析
    handleResponse(response);
}

代码解析：

• 第一步：创建SearchRequest对象，指定索引库名

• 第二步：利用request.source()来构建DSL，DSL语句中可以包含查询、分页、排序、高亮等操作

  • query()：代表查询条件，利用QueryBuilders.matchAllQuery()构建一个match_all查询的DSL

• 第三步：利用client.search()发送请求，得到响应

关键的API：

• request.source()，它的返回值其中包含了查询、排序、分页、高亮等所有功能

• QueryBuilders，这个类中包含match、term、function_score、bool等各种查询

6.2 查询结果解析

传入响应对象

private void handleResponse(SearchResponse response) {
    // 4.解析响应
    SearchHits searchHits = response.getHits();
    // 4.1.获取总条数
    long total = searchHits.getTotalHits().value;
    System.out.println("共搜索到" + total + "条数据");
    // 4.2.文档数组
    SearchHit[] hits = searchHits.getHits();
    // 4.3.遍历
    for (SearchHit hit : hits) {
        // 获取文档source
        String json = hit.getSourceAsString();
        // 反序列化
        HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
        System.out.println("hotelDoc = " + hotelDoc);
    }
}

elasticsearch返回的结果是一个JSON字符串，结构包含：

• hits：命中的结果
  • total：总条数，其中的value是具体的总条数值
  • max_score：所有结果中得分最高的文档的相关性算分
  • hits：搜索结果的文档数组，其中的每个文档都是一个json对象
    • _source：文档中的原始数据，也是json对象

因此，我们解析响应结果，就是逐层解析JSON字符串，流程如下：

• SearchHits：通过response.getHits()获取，就是JSON中的最外层的hits，代表命中的结果
  • SearchHits.getTotalHits().value：获取总条数信息
  • SearchHits.getHits()：获取SearchHit数组，也就是文档数组
    • SearchHit.getSourceAsString()：获取文档结果中的_source，也就是原始的json文档数据

6.3 match全文检索–根据字段模糊查询

@Test
void testMatch() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");
    
    // 2.准备请求参数(使用copy_to字段或者使用单独字段效果都是一样的，但是前者效率更高一些)
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    //request.source().query(QueryBuilders.multiMatchQuery("如家", "name", "brand"));        
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析（解析方法见查询结果解析部分）
    handleResponse(response);
}

6.4 精准查询

精确查询主要有两种：

• term：词条精确匹配
• range：范围查询

与之前的查询相比，只是查询条件不同而已，其它都一样。

6.4.1 term 查询

@Test
void testBool() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");

    // 2.准备请求参数（仅仅这里不同）
    request.source().query(QueryBuilders.termQuery("all", "如家"));     
    
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析（解析方法见查询结果解析部分）
    handleResponse(response);;
}

6.4.2 range 查询

@Test
void testBool() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");

    // 2.准备请求参数（仅仅这里不同）
    request.source().query(QueryBuilders.rangeQuery("price").gte(100),lte(200));     
    
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析（解析方法见查询结果解析部分）
    handleResponse(response);;
}

6.5 布尔查询

布尔查询是用must、must_not、filter等方式组合其它查询

@Test
void testBool() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备请求参数
    request.source().query(
        QueryBuilders.boolQuery()
        .must(QueryBuilders.termQuery("city", "杭州"))
        .filter(QueryBuilders.rangeQuery("price").lte(250))
    );
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析（解析方法见查询结果解析部分）
    handleResponse(response);
}

6.7 分页和排序

搜索结果的排序和分页是与query同级的参数，因此同样是使用request.source()来设置。

@Test
void testSortAndPage() throws IOException {
    int page = 2,size = 5;

    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备请求参数
    // 2.1.query
    request.source().query(QueryBuilders.matchAllQuery());
    // 2.2.排序sort
    request.source().sort("price", SortOrder.ASC);
    // 2.3.分页 from（从第几页开始） size（每页几条数据）
    request.source().from((page - 1) * size).size(size);

    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析
    handleResponse(response);
}

6.8 结果高亮

高亮的代码与之前代码差异较大，有两点：

• 查询的DSL：其中除了查询条件，还需要添加高亮条件，同样是与query同级
• 结果解析：结果除了要解析_source文档数据，还要解析高亮结果

6.8.1 高亮请求

@Test
void testHighlight() throws IOException {
    // 1.准备request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备请求参数
    // 2.1.query
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 2.2.高亮  
    // field("name")指定要高亮的字段  requireFieldMatch(false)对非搜索字段高亮
    request.source().highlighter(
        				new HighlightBuilder().field("name").requireFieldMatch(false));
    // 3.发送请求，得到响应
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.结果解析
    handleResponse(response);
}

6.8.2 .解析高亮结果

高亮的结果与查询的文档结果默认是分离的，并不在一起。

因此解析高亮的代码需要额外处理：

private void handleResponse(SearchResponse response) {
    SearchHits searchHits = response.getHits();
    // 4.1.总条数
    long total = searchHits.getTotalHits().value;
    System.out.println("总条数：" + total);
    // 4.2.获取文档数组
    SearchHit[] hits = searchHits.getHits();
    // 4.3.遍历
    for (SearchHit hit : hits) {
        // 4.4.获取source
        String json = hit.getSourceAsString();
        // 4.5.反序列化，非高亮的
        HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
        // 4.6.处理高亮结果
        // 1)获取高亮map
        Map<String, HighlightField> map = hit.getHighlightFields();
        if (!CollectionUtils.isEmpty(map)) {
            // 2）根据字段名，获取高亮结果
            HighlightField highlightField = map.get("name");
            // 3）获取高亮结果字符串数组中的第1个元素
            String hName = highlightField.getFragments()[0].toString();
            // 4）把高亮结果放到HotelDoc中
            hotelDoc.setName(hName);
        }
        // 4.7.打印
        System.out.println(hotelDoc);
    }
}

代码解读：

• 第一步：从结果中获取source。hit.getSourceAsString()返回非高亮结果的json字符串。

  还需要反序列化为HotelDoc对象

• 第二步：获取高亮结果。hit.getHighlightFields()返回值是一个Map集合。

  key是高亮字段名称，值是HighlightField对象，代表高亮值

• 第三步：从map中根据高亮字段名称，获取高亮字段值对象HighlightField

• 第四步：从HighlightField中获取Fragments，并且转为字符串。这部分就是真正的高亮字符串了

• 第五步：用高亮的结果替换HotelDoc中的非高亮结果

6.9 文档聚合

聚合条件与query条件同级别，因此需要使用request.source()来指定聚合条件。

 @Test
void testAgg() throws IOException {
    // 1.准备请求
    SearchRequest request = new SearchRequest("hotel");
    // 2.请求参数
    // 2.1.size
    request.source().size(0);
    // 2.2.聚合 （指定名称、类型、字段）
    request.source().aggregation(
        AggregationBuilders.terms("brandAgg").field("brand").size(20));
    // 3.发出请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析聚合结果
    Aggregations aggregations = response.getAggregations();
    // 4.1.根据聚合名称，获取聚合结果
    Terms brandAgg = aggregations.get("brandAgg");
    // 4.2.获取buckets
    List<? extends Terms.Bucket> buckets = brandAgg.getBuckets();
    // 4.3.遍历
    for (Terms.Bucket bucket : buckets) {
        String brandName = bucket.getKeyAsString();
        System.out.println("brandName = " + brandName);
        long docCount = bucket.getDocCount();
        System.out.println("docCount = " + docCount);
    }
}

聚合的结果也与查询结果不同，API也比较特殊。不过同样是JSON逐层解析。