13. RestClient操作elasticsearch

2022-05-23 本文已影响0人星野君

Rest Client操作索引库

初始化

引入es的RestHighLeveClient依赖

        <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：

@SpringBootTest
public class HotelIndexTest {
  private RestHighLevelClient restHighLevelClient;

  @BeforeEach
  void setUp() {
    this.restHighLevelClient =
        new RestHighLevelClient(RestClient.builder(HttpHost.create("192.168.187.128:9200")));
  }

  @AfterEach
  void tearDown() throws IOException {
    this.restHighLevelClient.close();
  }

  @Test
  void testInt() {
    System.out.println(restHighLevelClient);
  }
}

创建索引库

  @Test
  void createHotelIndex() throws IOException {
    // 1.准备Request      PUT /hotel
    CreateIndexRequest request = new CreateIndexRequest("hotel");
    // 2.准备请求参数
    request.source(MAPPING_TEMPLATE, XContentType.JSON);
    // 3.发送请求
    restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
  }

MAPPING_TEMPLATE

package cn.itcast.hotel.constants;

/**
 * @author ylf
 * @version 1.0
 */
public class HotelConstants {
  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"
          + "}";
}

删除索引库

  @Test
  void testDeleteHotelIndex() throws IOException {
    // 1.准备Request 
    DeleteIndexRequest request = new DeleteIndexRequest("hotel");
    // 2.发送请求
    restHighLevelClient.indices().delete(request, RequestOptions.DEFAULT);
  }

判断索引库是否存在

  @Test
  void testExistHotelIndex() throws IOException {
    // 1.准备Request 
    GetIndexRequest request = new GetIndexRequest("hotel");
    // 2.发送请求
    final boolean exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        System.out.println(exists ? "索引库存在" : "索引库不存在");
  }

Rest Client操作文档

从mysql读取数据然后存放到es中

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
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;

/**
 * @author ylf
 * @version 1.0
 */
@SpringBootTest
public class HotelDocumentTest {
  @Autowired IHotelService hotelService;
  private RestHighLevelClient restHighLevelClient;

  @BeforeEach
  void setUp() {
    this.restHighLevelClient =
        new RestHighLevelClient(RestClient.builder(HttpHost.create("192.168.187.128:9200")));
  }

  @AfterEach
  void tearDown() throws IOException {
    this.restHighLevelClient.close();
  }

  @Test
  void testAddDocument() throws IOException {
    // 根据id查询酒店数据
    final Hotel hotel = hotelService.getById("36934L");
    // 转换为文档类型
    final HotelDoc hotelDoc = new HotelDoc(hotel);
    // 1.准备Request对象
    final IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());
    // 2.准备json文档
    request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
    // 3.发送请求
    restHighLevelClient.index(request, RequestOptions.DEFAULT);
  }
}

查询

  @Test
  void testGetDocument() throws IOException {
    // 1.准备Request对象
    final GetRequest request = new GetRequest("hotel", "36934");
    // 2.发送请求
    final GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
    // 3. 解析响应结果
    String json = response.getSourceAsString();

    final HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
    System.out.println(hotelDoc);
  }

修改

  @Test
  void testUpdateDocument() throws IOException {
    // 1.准备Request对象
    final UpdateRequest request = new UpdateRequest("hotel", "36934");
    // 2.准备请求参数
    request.doc("price", "999", "starName", "一钻");
    // 3.发送请求
    restHighLevelClient.update(request, RequestOptions.DEFAULT);
  }

删除

  @Test
  void testDeleteDocument() throws IOException {
    // 1.准备Request对象
    final DeleteRequest request = new DeleteRequest("hotel", "36934");
    // 2.发送请求
    restHighLevelClient.delete(request, RequestOptions.DEFAULT);
  }

批量导入

  @Test
  void testBulkRequest() throws IOException {
    // 查询所有酒店数据
    final List<Hotel> list = hotelService.list();
    // 1.准备Request对象
    final BulkRequest request = new BulkRequest();
    // 2.准备请求参数
    // 转换为文档类型
    for (Hotel hotel : list) {
      final HotelDoc hotelDoc = new HotelDoc(hotel);
      request.add(
          new IndexRequest("hotel")
              .id(hotelDoc.getId().toString())
              .source(JSON.toJSONString(hotelDoc ), XContentType.JSON));
    }
    // 3.发送请求
    restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
  }

RestClient查询文档

match_all查询所有

  @Test
  void testSearchAll() throws IOException {
    // 1. 准备request
    final SearchRequest request = new SearchRequest("hotel");
    // 2. 准备DSL
    request.source().query(QueryBuilders.matchAllQuery());
    // 3. 发送请求
    final SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    // 4. 解析结果
    final SearchHits searchHits = response.getHits();
    // 4.1 查询的总条目
    final TotalHits totalHits = searchHits.getTotalHits();
    System.out.println("一共查询到" + totalHits.value + "条");
    // 4.2 查询的结果集
    final SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
      // 得到source
      final String json = hit.getSourceAsString();
      // 反序列化
      final HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
      // 打印
      System.out.println(hotelDoc);
    }
  }

条件查询
与查询所有基本一致，只需要修改第二步dsl

    // 1. 单字段查询
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 2. 多字段查询
    request.source().query(QueryBuilders.multiMatchQuery("如家", "name", "brand"));
    // 词条查询
    request.source().query(QueryBuilders.termQuery("city", "上海"));
    // 范围查询
    request.source().query(QueryBuilders.rangeQuery("price").gte(100).lte(200));

布尔查询

    // 布尔查询
    BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
    // 添加must条件
    boolQueryBuilder.must(QueryBuilders.termQuery("city", "上海"));
    // 添加filter条件
    boolQueryBuilder.filter(QueryBuilders.rangeQuery("price").gte(100).lte(200));

    request.source().query(boolQueryBuilder);

排序分页

   request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 分页
    request.source().from(0).size(20);
    // 排序
    request.source().sort("price", SortOrder.DESC);
    // 地理位置排序
    request
        .source()
        .sort(
            SortBuilders.geoDistanceSort("location", new GeoPoint(location))
                .order(SortOrder.ASC)
                .unit(DistanceUnit.KILOMETERS));

高亮

  @Test
  void testSearch() throws IOException {
    // 1. 准备request
    final SearchRequest request = new SearchRequest("hotel");
    // 单字段查询
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 高亮
    request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));
    // 3. 发送请求
    final SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    handleResponse(response);
  }

  private void handleResponse(SearchResponse response) {
    // 4. 解析结果
    final SearchHits searchHits = response.getHits();
    // 4.1 查询的总条目
    final TotalHits totalHits = searchHits.getTotalHits();
    System.out.println("一共查询到" + totalHits.value + "条");
    // 4.2 查询的结果集
    final SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
      // 得到source
      final String json = hit.getSourceAsString();
      // 反序列化
      final HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
      // 获取高亮结果
      final Map<String, HighlightField> highlightFields = hit.getHighlightFields();
      if (!CollectionUtils.isEmpty(highlightFields)) {
        // 根据名称获取高亮结果
        final HighlightField highlightField = highlightFields.get("name");
        if (highlightField != null) {
          // 获取高亮值
          final String name = highlightField.getFragments()[0].string();
          // 覆盖高亮结果
          hotelDoc.setName(name);
        }
      }
      // 打印
      System.out.println(hotelDoc);
    }
  }

function score组合查询

    // 2.算分控制
    final FunctionScoreQueryBuilder functionScoreQueryBuilder =
        QueryBuilders.functionScoreQuery(
            // 原始查询，相关性算分
            boolQueryBuilder,
            // function score的数组
            new FunctionScoreQueryBuilder.FilterFunctionBuilder[] {
              // 其中的一个function score元素
              new FunctionScoreQueryBuilder.FilterFunctionBuilder(
                  // 过滤条件
                  QueryBuilders.termQuery("isAd", true),
                  // 算分函数
                  ScoreFunctionBuilders.weightFactorFunction(10))
            });

    request.source().query(functionScoreQueryBuilder);

聚合

  @Test
  void testAggregation() throws IOException {
    // 1. 准备request
    final SearchRequest request = new SearchRequest("hotel");
    // 2. 准备DSL
    // 2.1 设置size
    request.source().size(0);
    // 2.2 聚合
    request.source().aggregation(AggregationBuilders.terms("brandAgg").field("brand").size(20));
    // 3. 发送请求
    final SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    // 4. 解析结果
    // 4.1 解析聚合结果
    final Aggregations aggregations = response.getAggregations();
    // 4.2 根据名称获取聚合结果
    final Terms brandAgg = aggregations.get("brandAgg");
    // 4.3 获取Buckets
    final List<? extends Terms.Bucket> buckets = brandAgg.getBuckets();
    // 4.4 遍历
    for (Terms.Bucket bucket : buckets) {
      final String keyAsString = bucket.getKeyAsString();
      System.out.println(keyAsString);
    }
  }

自动补全

  @Test
  void testSuggest() throws IOException {
    // 1. 准备request
    final SearchRequest request = new SearchRequest("hotel");
    // 2. 准备DSL
    request
        .source()
        .suggest(
            new SuggestBuilder()
                .addSuggestion(
                    "suggest",
                    SuggestBuilders.completionSuggestion("suggestion")
                        .prefix("s")
                        .skipDuplicates(true)
                        .size(10)));
    // 3. 发送请求
    final SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    // 4. 解析结果
    final Suggest suggest = response.getSuggest();
    // 4.1 根据补全名称，查询结果
    final CompletionSuggestion suggestion = suggest.getSuggestion("suggest");
    // 4.2 获取options
    final List<CompletionSuggestion.Entry.Option> options = suggestion.getOptions();
    for (CompletionSuggestion.Entry.Option option : options) {
      final String text = option.getText().toString();
      System.out.println(text);
    }
  }

数据同步

方式一：同步调用

优点：实现简单，粗暴
缺点：业务耦合度高

方式二：异步通知

优点：低耦合，实现难度一般
缺点：依赖mq的可靠性

方式三：监听binlog

优点：完全解除服务间耦合
缺点：开启binlog增加数据库负担、实现复杂度高

13. RestClient操作elasticsearch

Rest Client操作索引库

Rest Client操作文档

RestClient查询文档

聚合

自动补全

数据同步

猜你喜欢

热点阅读