數(shù)據(jù)聚合

聚合的種類

聚合（aggregations）可以實(shí)現(xiàn)對(duì)文檔數(shù)據(jù)的統(tǒng)計(jì)、分析、運(yùn)算。聚合常見的有三類：

• 桶（Bucket）聚合：用來對(duì)文檔做分組

  • TermAggregation：按照文檔字段值分組
  • Date Histogram：按照日期階梯分組，例如一周為一組，或者一月為一組

• 度量（Metric）聚合：用以計(jì)算一些值，比如：最大值、最小值、平均值等

  • Avg：求平均值
  • Max：求最大值
  • Min：求最小值
  • Stats：同時(shí)求max、min、avg、sum等

• 管道（pipeline）聚合：其它聚合的結(jié)果為基礎(chǔ)做聚合

可以類比mysql數(shù)據(jù)庫(kù)，（桶=》group by 分組，度量=》聚合函數(shù)，管道=》）

參與聚合的字段類型必須是：

• keyword
• 數(shù)值
• 日期
• 布爾

DSL實(shí)現(xiàn)聚合

DSL實(shí)現(xiàn)Bucket聚合

現(xiàn)在，我們要統(tǒng)計(jì)所有數(shù)據(jù)中的酒店品牌有幾種，此時(shí)可以根據(jù)酒店品牌的名稱做聚合。
類型為 term 類型，DSL示例：

GET /hotel/_search
{
  "size": 0,  // 設(shè)置size為0，結(jié)果中不包含文檔，只包含聚合結(jié)果
  "aggs": { // 定義聚合
    "brandAgg": { //給聚合起個(gè)名字
      "terms": { // 聚合的類型，按照品牌值聚合，所以選擇term
        "field": "brand", // 參與聚合的字段
        "size": 20 // 希望獲取的聚合結(jié)果數(shù)量
      }
    }
  }
}

Bucket聚合-聚合結(jié)果排序

默認(rèn)情況下，Bucket 聚合會(huì)統(tǒng)計(jì) Bucket 內(nèi)的文檔數(shù)量，記為 _count，并且按照 _count 降序排序。
我們可以修改結(jié)果排序方式：

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "order": {
          "_count": "asc" // 按照_count升序排列
        },
        "size": 20
      }
    }
  }
}

Bucket聚合-限定聚合范圍

默認(rèn)情況下，Bucket聚合是對(duì)索引庫(kù)的所有文檔做聚合，我們可以限定要聚合的文檔范圍，只要添加 query 條件即可：

GET /hotel/_search
{
  "query": {
    "range": {
      "price": {
        "lte": 200 // 只對(duì)200元以下的文檔聚合
      }
    }
  }, 
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "size": 20
      }
    }
  }
}

aggs代表聚合，與query同級(jí)，此時(shí)query的作用是？

• 限定聚合的的文檔范圍

聚合必須的三要素：

• 聚合名稱
• 聚合類型
• 聚合字段

聚合可配置屬性有：

• size：指定聚合結(jié)果數(shù)量
• order：指定聚合結(jié)果排序方式
• field：指定聚合字段

DSL實(shí)現(xiàn)Metrics 聚合

例如，我們要求獲取每個(gè)品牌的用戶評(píng)分的 min、max、avg 等值.
我們可以利用 stats 聚合：

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": { 
      "terms": { 
        "field": "brand", 
        "size": 20
      },
      "aggs": { // 是brands聚合的子聚合，也就是分組后對(duì)每組分別計(jì)算
        "score_stats": { // 聚合名稱
          "stats": { // 聚合類型，這里stats可以計(jì)算min、max、avg等
            "field": "score" // 聚合字段，這里是score
          }
        }
      }
    }
  }
}

RestAPI實(shí)現(xiàn)聚合

我們以品牌聚合為例，演示下 Java 的 RestClient 使用，先看請(qǐng)求組裝：

再看下聚合結(jié)果解析

在IUserService中定義方法，實(shí)現(xiàn)對(duì)品牌、城市、星級(jí)的聚合

需求：搜索頁(yè)面的品牌、城市等信息不應(yīng)該是在頁(yè)面寫死，而是通過聚合索引庫(kù)中的酒店數(shù)據(jù)得來的：

在IUserService中定義一個(gè)方法，實(shí)現(xiàn)對(duì)品牌、城市、星級(jí)的聚合，方法聲明如下：

對(duì)接前端接口

前端頁(yè)面會(huì)向服務(wù)端發(fā)起請(qǐng)求，查詢品牌、城市、星級(jí)等字段的聚合結(jié)果：

可以看到請(qǐng)求參數(shù)與之前search時(shí)的RequestParam完全一致，這是在限定聚合時(shí)的文檔范圍。
例如：用戶搜索“外灘”，價(jià)格在300~600，那聚合必須是在這個(gè)搜索條件基礎(chǔ)上完成。
因此我們需要：

1. 編寫controller接口，接收該請(qǐng)求
2. 修改IUserService#getFilters()方法，添加RequestParam參數(shù)
3. 修改getFilters方法的業(yè)務(wù)，聚合時(shí)添加query條件

@Test
void testAggregation() throws IOException {
    // 1. 準(zhǔn)備Request
    SearchRequest request = new SearchRequest("hotel");
    // 2. 準(zhǔn)備DSL
    // 2.1 設(shè)置size
    request.source().size(0);
    // 2.2 聚合
    request.source().aggregation(AggregationBuilders
            .terms("brandAgg")
            .field("brand")
            .size(10)
    );
    // 3. 發(fā)出請(qǐng)求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4. 解析結(jié)果
    Aggregations aggregations = response.getAggregations();
    // 4.1 根據(jù)聚合名稱獲取聚合結(jié)果
    Terms brandTerms = aggregations.get("brandAgg");
    // 4.2 獲取 buckets
    List<? extends Terms.Bucket> buckets = brandTerms.getBuckets();
    for (Terms.Bucket bucket : buckets) {
        String key = bucket.getKeyAsString();
        System.out.println(key);
    }
}

Controller

@PostMapping("filters")
public Map<String, List<String>> getFilters(@RequestBody RequestParams params){
    return hotelService.filters(params);
}

Service接口

@Override
public Map<String, List<String>> filters(RequestParams params) {
    try {
        // 1. 準(zhǔn)備Request
        SearchRequest request = new SearchRequest("hotel");
        // 2. 準(zhǔn)備DSL
        // 2.1 query
        buildBasicQuery(params, request);
        // 2.2 設(shè)置size
        request.source().size(0);
        // 2.3 聚合
        buildAggregation(request);
        // 3. 發(fā)出請(qǐng)求
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        // 4. 解析結(jié)果
        Map<String, List<String>> result = new HashMap<>();
        Aggregations aggregations = response.getAggregations();
        // 4.1 根據(jù)品牌名稱，獲取品牌結(jié)果
        List<String> brandList = getAggByName(aggregations, "brandAgg");
        // 4.2 根據(jù)城市名稱，獲取城市結(jié)果
        List<String> cityList = getAggByName(aggregations, "cityAgg");
        // 4.3 根據(jù)星級(jí)名稱，獲取星級(jí)結(jié)果
        List<String> starList = getAggByName(aggregations, "starAgg");

        // 4.4 放入map
        result.put("品牌", brandList);
        result.put("城市", cityList);
        result.put("星級(jí)", starList);
        return result;
    } catch (IOException e) {
        throw new RuntimeException(e);
    } 
}

private List<String> getAggByName(Aggregations aggregations, String aggName) {
    // 4.1 根據(jù)聚合名稱獲取聚合結(jié)果
    Terms brandTerms = aggregations.get(aggName);
    // 4.2 獲取 buckets
    List<? extends Terms.Bucket> buckets = brandTerms.getBuckets();
    // 4.3 遍歷
    List<String> brandList = new ArrayList<>();
    for (Terms.Bucket bucket : buckets) {
        String key = bucket.getKeyAsString();
        brandList.add(key);
    }
    return brandList;
}

private void buildAggregation(SearchRequest request) {
    request.source().aggregation(AggregationBuilders
            .terms("brandAgg")
            .field("brand")
            .size(100)
    );
    request.source().aggregation(AggregationBuilders
            .terms("cityAgg")
            .field("city")
            .size(100)
    );
    request.source().aggregation(AggregationBuilders
            .terms("starAgg")
            .field("star")
            .size(100)
    );
}

private void buildBasicQuery(RequestParams params, SearchRequest request) {
    // 1. 構(gòu)建BooleanQuery
    BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
    // 關(guān)鍵字搜索
    String key = params.getKey();
    if(key == null || "".equals(key)){
        boolQuery.must(QueryBuilders.matchAllQuery());
    }else{
        boolQuery.must(QueryBuilders.matchQuery("all", key));
    }
    // 條件過濾
    // 城市條件
    if (params.getCity() != null && !params.getCity().equals("")){
        boolQuery.filter(QueryBuilders.termQuery("city", params.getCity()));
    }
    // 品牌條件
    if (params.getBrand() != null && !params.getBrand().equals("")){
        boolQuery.filter(QueryBuilders.termQuery("brand", params.getBrand()));
    }
    // 星級(jí)條件
    if (params.getStarName() != null && !params.getStarName().equals("")){
        boolQuery.filter(QueryBuilders.termQuery("starName", params.getBrand()));
    }
    // 價(jià)格
    if (params.getMinPrice() != null && params.getMaxPrice() != null){
        boolQuery.filter(QueryBuilders
                 .rangeQuery("price").gte(params.getMinPrice()).lte(params.getMaxPrice()));
    }
    // 2. 算分控制
    FunctionScoreQueryBuilder functionScoreQuery = QueryBuilders.functionScoreQuery(
            // 原始查詢，相關(guān)性算分查詢
            boolQuery,
            // function score 的數(shù)組
            new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
                    // 其中的一個(gè) function score 元素
                    new FunctionScoreQueryBuilder.FilterFunctionBuilder(
                            // 過濾條件
                            QueryBuilders.termQuery("isAD", true),
                            // 算分函數(shù)
                            ScoreFunctionBuilders.weightFactorFunction(10)
                    )
            });

    request.source().query(functionScoreQuery);
}

自動(dòng)補(bǔ)全

自動(dòng)補(bǔ)全需求說明

當(dāng)用戶在搜索框輸入字符時(shí)，我們應(yīng)該提示出與該字符有關(guān)的搜索項(xiàng)，如圖：

拼音分詞器

要實(shí)現(xiàn)根據(jù)字母做補(bǔ)全，就必須對(duì)文檔按照拼音分詞。在GitHub上恰好有elasticsearch的拼音分詞插件。地址：https://github.com/medcl/elasticsearch-analysis-pinyin

安裝方式與IK分詞器一樣，分三步：

1. 解壓
2. 上傳到虛擬機(jī)中，elasticsearch的plugin目錄
3. 重啟elasticsearch
4. 測(cè)試

POST /_analyze
{
 "text": "如家酒店整挺好",
 "analyzer": "pinyin"
}

自定義分詞器

elasticsearch中分詞器（analyzer）的組成包含三部分：

• character filters：在tokenizer之前對(duì)文本進(jìn)行處理。例如刪除字符、替換字符
• tokenizer：將文本按照一定的規(guī)則切割成詞條（term）。例如keyword，就是不分詞；還有ik_smart
• tokenizer filter：將tokenizer輸出的詞條做進(jìn)一步處理。例如大小寫轉(zhuǎn)換、同義詞處理、拼音處理等

我們可以在創(chuàng)建索引庫(kù)（自定義分詞器只對(duì)指定的索引庫(kù)適用）時(shí)，通過settings來配置自定義的analyzer（分詞器）：

拼音分詞器適合在創(chuàng)建倒排索引的時(shí)候使用，但不能在搜索的時(shí)候使用。
創(chuàng)建倒排索引時(shí)：

因此字段在創(chuàng)建倒排索引時(shí)應(yīng)該用 my_analyzer 分詞器；字段在搜索時(shí)應(yīng)該使用 ik_smart 分詞器;

DELETE /test

# 自定義拼音分詞器
PUT /test
{
  "settings": {
    "analysis": {
      "analyzer": { 
        "my_analyzer": {
          "tokenizer": "ik_max_word",
          "filter": "py"
        }
      },
      "filter": {
        "py": { 
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name":{
        "type": "text",
        "analyzer": "my_analyzer",  
        "search_analyzer": "ik_smart"
      }
    }
  }
}

POST /test/_doc/1
{
  "id": 1,
  "name": "獅子"
}
POST /test/_doc/2
{
  "id": 2,
  "name": "虱子"
}

GET /test/_search
{
  "query": {
    "match": {
      "name": "掉入獅子籠咋辦"
    }
  }
}

如何使用拼音分詞器？

1. 下載pinyin分詞器
2. 解壓并放到elasticsearch的plugin目錄
3. 重啟即可

如何自定義分詞器？

1. 創(chuàng)建索引庫(kù)時(shí)，在settings中配置，可以包含三部分
2. character filter
3. tokenizer
4. filter

拼音分詞器注意事項(xiàng)？

• 為了避免搜索到同音字，搜索時(shí)不要使用拼音分詞器

自動(dòng)補(bǔ)全查詢

completion suggester查詢

elasticsearch提供了Completion Suggester 查詢來實(shí)現(xiàn)自動(dòng)補(bǔ)全功能。這個(gè)查詢會(huì)匹配以用戶輸入內(nèi)容開頭的詞條并返回。為了提高補(bǔ)全查詢的效率，對(duì)于文檔中字段的類型有一些約束：

• 參與補(bǔ)全查詢的字段必須是completion類型。
• 字段的內(nèi)容一般是用來補(bǔ)全的多個(gè)詞條形成的數(shù)組。

查詢語(yǔ)法如下：

# 自動(dòng)補(bǔ)全的索引庫(kù)
PUT test2
{
  "mappings": {
    "properties": {
      "title":{
        "type": "completion"
      }
    }
  }
}
# 示例數(shù)據(jù)
POST test2/_doc
{
  "title": ["Sony", "WH-1000XM3"]
}
POST test2/_doc
{
  "title": ["SK-II", "PITERA"]
}
POST test2/_doc
{
  "title": ["Nintendo", "switch"]
}


# 自動(dòng)補(bǔ)全查詢
GET /test2/_search
{
  "suggest": {
    "titelSuggest": {
      "text": "s",
      "completion": {
        "field": "title",
        "skip_duplicates": true,
        "size": 10
      }
    }
  }
}

自動(dòng)補(bǔ)全對(duì)字段的要求：

• 類型是completion類型
• 字段值是多詞條的數(shù)組

酒店數(shù)據(jù)自動(dòng)補(bǔ)全

實(shí)現(xiàn)hotel索引庫(kù)的自動(dòng)補(bǔ)全、拼音搜索功能

實(shí)現(xiàn)思路如下：

1. 修改hotel索引庫(kù)結(jié)構(gòu)，設(shè)置自定義拼音分詞器
2. 修改索引庫(kù)的name、all字段，使用自定義分詞器
3. 索引庫(kù)添加一個(gè)新字段suggestion，類型為completion類型，使用自定義的分詞器
4. 給HotelDoc類添加suggestion字段，內(nèi)容包含brand、business
5. 重新導(dǎo)入數(shù)據(jù)到hotel庫(kù)

注意：name、all是可分詞的，自動(dòng)補(bǔ)全的brand、business是不可分詞的，要使用不同的分詞器組合

# 酒店數(shù)據(jù)索引庫(kù)
PUT /hotel
{
  "settings": {
    "analysis": {
      "analyzer": {
        "text_anlyzer": {
          "tokenizer": "ik_max_word",
          "filter": "py"
        },
        "completion_analyzer": {
          "tokenizer": "keyword",
          "filter": "py"
        }
      },
      "filter": {
        "py": {
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "id":{
        "type": "keyword"
      },
      "name":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart",
        "copy_to": "all"
      },
      "address":{
        "type": "keyword",
        "index": false
      },
      "price":{
        "type": "integer"
      },
      "score":{
        "type": "integer"
      },
      "brand":{
        "type": "keyword",
        "copy_to": "all"
      },
      "city":{
        "type": "keyword"
      },
      "starName":{
        "type": "keyword"
      },
      "business":{
        "type": "keyword",
        "copy_to": "all"
      },
      "location":{
        "type": "geo_point"
      },
      "pic":{
        "type": "keyword",
        "index": false
      },
      "all":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart"
      },
      "suggestion":{
          "type": "completion",
          "analyzer": "completion_analyzer"
      }
    }
  }
}

GET /hotel/_search
{
  "query": {
    "match_all": {}
  }
}


GET /hotel/_search
{
  "suggest": {
    "titelSuggest": {
      "text": "h",
      "completion": {
        "field": "suggestion",
        "skip_duplicates": true,
        "size": 10
      }
    }
  }
}

RestAPI實(shí)現(xiàn)自動(dòng)補(bǔ)全

先看請(qǐng)求參數(shù)構(gòu)造的API：

再來看結(jié)果解析：

實(shí)現(xiàn)酒店搜索頁(yè)面輸入框的自動(dòng)補(bǔ)全

查看前端頁(yè)面，可以發(fā)現(xiàn)當(dāng)我們?cè)谳斎肟蜴I入時(shí)，前端會(huì)發(fā)起ajax請(qǐng)求：

在服務(wù)端編寫接口，接收該請(qǐng)求，返回補(bǔ)全結(jié)果的集合，類型為List<String>

controller

@GetMapping("suggestion")
public List<String> getSuggestions(@RequestParam("key") String prefix){
    return hotelService.getSuggestions(prefix);
}

service

@Override
public List<String> getSuggestions(String prefix) {
    try {
        // 1. 準(zhǔn)備Request
        SearchRequest request = new SearchRequest("hotel");
        // 2. 準(zhǔn)備DSL
        request.source().suggest(new SuggestBuilder().addSuggestion(
                "suggestions",
                SuggestBuilders.completionSuggestion("suggestion")
                .prefix(prefix)
                .skipDuplicates(true)
                .size(10)
        ));
        // 3. 發(fā)送請(qǐng)求
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        // 4. 解析結(jié)果
        Suggest suggest = response.getSuggest();
        // 4.1 根據(jù)補(bǔ)全查詢名稱，獲取補(bǔ)全結(jié)果
        CompletionSuggestion suggestions =  suggest.getSuggestion("suggestions");
        // 4.2 獲取options
        List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions();
        // 4.3 遍歷
        List<String> list = new ArrayList<>(options.size());
        for (CompletionSuggestion.Entry.Option option : options) {
            String text = option.getText().toString();
            list.add(text);
        }
        return list;
    } catch (IOException e) {
        throw new RuntimeException();
    }
}

數(shù)據(jù)同步

數(shù)據(jù)同步思路分析

elasticsearch中的酒店數(shù)據(jù)來自于mysql數(shù)據(jù)庫(kù)，因此mysql數(shù)據(jù)發(fā)生改變時(shí)，elasticsearch也必須跟著改變，這個(gè)就是elasticsearch與mysql之間的數(shù)據(jù)同步。

方案一：同步調(diào)用

方案二：異步通知

方案三：監(jiān)聽binlog

方式一：同步調(diào)用

• 優(yōu)點(diǎn)：實(shí)現(xiàn)簡(jiǎn)單，粗暴
• 缺點(diǎn)：業(yè)務(wù)耦合度高

方式二：異步通知

• 優(yōu)點(diǎn)：低耦合，實(shí)現(xiàn)難度一般
• 缺點(diǎn)：依賴mq的可靠性

方式三：監(jiān)聽binlog

• 優(yōu)點(diǎn)：完全解除服務(wù)間耦合
• 缺點(diǎn)：開啟binlog增加數(shù)據(jù)庫(kù)負(fù)擔(dān)、實(shí)現(xiàn)復(fù)雜度高

實(shí)現(xiàn)elasticsearch與數(shù)據(jù)庫(kù)數(shù)據(jù)同步

利用MQ實(shí)現(xiàn)mysql與elasticsearch數(shù)據(jù)同步

利用課前資料提供的hotel-admin項(xiàng)目作為酒店管理的微服務(wù)。當(dāng)酒店數(shù)據(jù)發(fā)生增、刪、改時(shí)，要求對(duì)elasticsearch中數(shù)據(jù)也要完成相同操作。
步驟：

• 導(dǎo)入課前資料提供的hotel-admin項(xiàng)目，啟動(dòng)并測(cè)試酒店數(shù)據(jù)的CRUD
• 聲明exchange、queue、RoutingKey
• 在hotel-admin中的增、刪、改業(yè)務(wù)中完成消息發(fā)送
• 在hotel-demo中完成消息監(jiān)聽，并更新elasticsearch中數(shù)據(jù)
• 啟動(dòng)并測(cè)試數(shù)據(jù)同步功能

1. 導(dǎo)入項(xiàng)目
   
2. 聲明exchange、queue、RoutingKey（兩類消息，兩種隊(duì)列）

導(dǎo)入amqp依賴

<!--amqp-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

yaml文件中配置rabbitmq

spring:
  rabbitmq:
    host: 10.211.55.6
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /

MqConstants.java

public class MqConstants {

    /**
     * 交換機(jī)
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";

    /**
     * 監(jiān)聽新增和修改的隊(duì)列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";

    /**
     * 監(jiān)聽刪除的隊(duì)列
     */
    public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";

    /**
     * 新增或修改的RoutingKey
     */
    public final static String HOTEL_INSERT_KEY = "hotel.insert";

    /**
     * 刪除的RoutingKey
     */
    public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

MqConfig.java

@Configuration
public class MqConfig {

    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    }

    @Bean
    public Queue insertQueue(){
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    }

    @Bean
    public Queue deleteQueue(){
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    }

    @Bean
    public Binding insertQueueBinding(){
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding(){
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    }
}

3. 在hotel-admin中的增、刪、改業(yè)務(wù)中完成消息發(fā)送

導(dǎo)入依賴，配置 yaml 文件

controller中

@RestController
@RequestMapping("hotel")
public class HotelController {

    @Autowired
    private IHotelService hotelService;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/{id}")
    public Hotel queryById(@PathVariable("id") Long id){
        return hotelService.getById(id);
    }

    @GetMapping("/list")
    public PageResult hotelList(
            @RequestParam(value = "page", defaultValue = "1") Integer page,
            @RequestParam(value = "size", defaultValue = "1") Integer size
    ){
        Page<Hotel> result = hotelService.page(new Page<>(page, size));

        return new PageResult(result.getTotal(), result.getRecords());
    }

    @PostMapping
    public void saveHotel(@RequestBody Hotel hotel){
        hotelService.save(hotel);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId());
    }

    @PutMapping()
    public void updateById(@RequestBody Hotel hotel){
        if (hotel.getId() == null) {
            throw new InvalidParameterException("id不能為空");
        }
        hotelService.updateById(hotel);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId());
    }

    @DeleteMapping("/{id}")
    public void deleteById(@PathVariable("id") Long id) {
        hotelService.removeById(id);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_DELETE_KEY, id);

    }
}

4. 在hotel-demo中完成消息監(jiān)聽，并更新elasticsearch中數(shù)據(jù)

HotelListener.java

@Component
public class HotelListener {

    @Autowired
    private IHotelService hotelService;

    /**
     * 監(jiān)聽酒店新增或修改的業(yè)務(wù)
     * @param id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id){
        hotelService.insertById(id);
    }

    /**
     * 監(jiān)聽酒店新刪除的業(yè)務(wù)
     * @param id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id){
        hotelService.deleteById(id);
    }
}

service

@Override
public void deleteById(Long id) {
    try {
        // 1. 準(zhǔn)備Request
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        // 2. 準(zhǔn)備發(fā)送請(qǐng)求
        client.delete(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

@Override
public void insertById(Long id) {
    try {
        // 0. 根據(jù)id查詢酒店數(shù)據(jù)
        Hotel hotel = getById(id);
        // 轉(zhuǎn)換為文檔類型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        // 1. 準(zhǔn)備Request對(duì)象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        // 2. 準(zhǔn)備Json文檔
        request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        // 3. 發(fā)送請(qǐng)求
        client.index(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

集群

ES集群結(jié)構(gòu)

單機(jī)的elasticsearch做數(shù)據(jù)存儲(chǔ)，必然面臨兩個(gè)問題：海量數(shù)據(jù)存儲(chǔ)問題、單點(diǎn)故障問題。

• 海量數(shù)據(jù)存儲(chǔ)問題：將索引庫(kù)從邏輯上拆分為N個(gè)分片（shard），存儲(chǔ)到多個(gè)節(jié)點(diǎn)
• 單點(diǎn)故障問題：將分片數(shù)據(jù)在不同節(jié)點(diǎn)備份（replica ）

搭建ES集群

我們會(huì)在單機(jī)上利用docker容器運(yùn)行多個(gè)es實(shí)例來模擬es集群。不過生產(chǎn)環(huán)境推薦大家每一臺(tái)服務(wù)節(jié)點(diǎn)僅部署一個(gè)es的實(shí)例。

部署es集群可以直接使用docker-compose來完成，但這要求你的Linux虛擬機(jī)至少有4G的內(nèi)存空間

創(chuàng)建es集群

首先編寫一個(gè)docker-compose文件，內(nèi)容如下：

version: '2.2'
services:
  es01:
    image: elasticsearch:7.12.1
    container_name: es01
    environment:
      - node.name=es01
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es02,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - data01:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
    networks:
      - elastic
  es02:
    image: elasticsearch:7.12.1
    container_name: es02
    environment:
      - node.name=es02
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - data02:/usr/share/elasticsearch/data
    ports:
      - 9201:9200
    networks:
      - elastic
  es03:
    image: elasticsearch:7.12.1
    container_name: es03
    environment:
      - node.name=es03
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es02
      - cluster.initial_master_nodes=es01,es02,es03
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - data03:/usr/share/elasticsearch/data
    networks:
      - elastic
    ports:
      - 9202:9200
volumes:
  data01:
    driver: local
  data02:
    driver: local
  data03:
    driver: local

networks:
  elastic:
    driver: bridge

es運(yùn)行需要修改一些linux系統(tǒng)權(quán)限，修改/etc/sysctl.conf文件

vi /etc/sysctl.conf

添加下面的內(nèi)容：

vm.max_map_count=262144

然后執(zhí)行命令，讓配置生效：

sysctl -p

通過docker-compose啟動(dòng)集群：

docker-compose up -d

集群狀態(tài)監(jiān)控

kibana可以監(jiān)控es集群，不過新版本需要依賴es的x-pack 功能，配置比較復(fù)雜。

這里推薦使用cerebro來監(jiān)控es集群狀態(tài)，官方網(wǎng)址：https://github.com/lmenezes/cerebro

解壓即可使用，非常方便。

解壓好的目錄如下：

進(jìn)入對(duì)應(yīng)的bin目錄：

雙擊其中的cerebro.bat文件即可啟動(dòng)服務(wù)。

訪問http://localhost:9000 即可進(jìn)入管理界面：

輸入你的elasticsearch的任意節(jié)點(diǎn)的地址和端口，點(diǎn)擊connect即可：

綠色的條，代表集群處于綠色（健康狀態(tài)）。

創(chuàng)建索引庫(kù)

1）利用kibana的DevTools創(chuàng)建索引庫(kù)

在DevTools中輸入指令：

PUT /itcast
{
  "settings": {
    "number_of_shards": 3, // 分片數(shù)量
    "number_of_replicas": 1 // 副本數(shù)量
  },
  "mappings": {
    "properties": {
      // mapping映射定義 ...
    }
  }
}

2）利用cerebro創(chuàng)建索引庫(kù)

利用cerebro還可以創(chuàng)建索引庫(kù)：

填寫索引庫(kù)信息：

點(diǎn)擊右下角的create按鈕：

查看分片效果

回到首頁(yè)，即可查看索引庫(kù)分片效果：

每個(gè)索引庫(kù)的分片數(shù)量、副本數(shù)量都是在創(chuàng)建索引庫(kù)時(shí)指定的，并且分片數(shù)量一旦設(shè)置以后無法修改。語(yǔ)法如下：

ES集群的節(jié)點(diǎn)角色

elasticsearch中集群節(jié)點(diǎn)有不同的職責(zé)劃分：

elasticsearch中的每個(gè)節(jié)點(diǎn)角色都有自己不同的職責(zé)，因此建議集群部署時(shí)，每個(gè)節(jié)點(diǎn)都有獨(dú)立的角色。

集群腦裂問題

默認(rèn)情況下，每個(gè)節(jié)點(diǎn)都是master eligible節(jié)點(diǎn)，因此一旦master節(jié)點(diǎn)宕機(jī)，其它候選節(jié)點(diǎn)會(huì)選舉一個(gè)成為主節(jié)點(diǎn)。當(dāng)主節(jié)點(diǎn)與其他節(jié)點(diǎn)網(wǎng)絡(luò)故障時(shí)，可能發(fā)生腦裂問題。

為了避免腦裂，需要要求選票超過 ( eligible節(jié)點(diǎn)數(shù)量 + 1 ）/ 2 才能當(dāng)選為主，因此eligible節(jié)點(diǎn)數(shù)量最好是奇數(shù)。對(duì)應(yīng)配置項(xiàng)是discovery.zen.minimum_master_nodes，在es7.0以后，已經(jīng)成為默認(rèn)配置，因此一般不會(huì)發(fā)生腦裂問題

master eligible節(jié)點(diǎn)的作用是什么？

• 參與集群選主
• 主節(jié)點(diǎn)可以管理集群狀態(tài)、管理分片信息、處理創(chuàng)建和刪除索引庫(kù)的請(qǐng)求

data節(jié)點(diǎn)的作用是什么？

• 數(shù)據(jù)的CRUD

coordinator節(jié)點(diǎn)的作用是什么？

• 路由請(qǐng)求到其它節(jié)點(diǎn)
• 合并查詢到的結(jié)果，返回給用戶

集群分布式存儲(chǔ)

當(dāng)新增文檔時(shí)，應(yīng)該保存到不同分片，保證數(shù)據(jù)均衡，那么coordinating node如何確定數(shù)據(jù)該存儲(chǔ)到哪個(gè)分片呢？
elasticsearch會(huì)通過hash算法來計(jì)算文檔應(yīng)該存儲(chǔ)到哪個(gè)分片：

說明：

• _routing默認(rèn)是文檔的id
• 算法與分片數(shù)量有關(guān)，因此索引庫(kù)一旦創(chuàng)建，分片數(shù)量不能修改！

新增文檔流程：

集群分布式查詢

elasticsearch的查詢分成兩個(gè)階段：

• scatter phase：分散階段，coordinating node會(huì)把請(qǐng)求分發(fā)到每一個(gè)分片
• gather phase：聚集階段，coordinating node匯總data node的搜索結(jié)果，并處理為最終結(jié)果集返回給用戶

分布式新增如何確定分片？

• coordinating node根據(jù)id做hash運(yùn)算，得到結(jié)果對(duì)shard數(shù)量取余，余數(shù)就是對(duì)應(yīng)的分片

分布式查詢的兩個(gè)階段

• 分散階段： coordinating node將查詢請(qǐng)求分發(fā)給不同分片
• 收集階段：將查詢結(jié)果匯總到coordinating node ，整理并返回給用戶

集群故障轉(zhuǎn)移

集群的master節(jié)點(diǎn)會(huì)監(jiān)控集群中的節(jié)點(diǎn)狀態(tài)，如果發(fā)現(xiàn)有節(jié)點(diǎn)宕機(jī)，會(huì)立即將宕機(jī)節(jié)點(diǎn)的分片數(shù)據(jù)遷移到其它節(jié)點(diǎn)，確保數(shù)據(jù)安全，這個(gè)叫做故障轉(zhuǎn)移。

故障轉(zhuǎn)移：文章來源地址http://www.zghlxwxcb.cn/news/detail-817900.html

• master宕機(jī)后，EligibleMaster選舉為新的主節(jié)點(diǎn)。
• master節(jié)點(diǎn)監(jiān)控分片、節(jié)點(diǎn)狀態(tài)，將故障節(jié)點(diǎn)上的分片轉(zhuǎn)移到正常節(jié)點(diǎn)，確保數(shù)據(jù)安全。

到了這里，關(guān)于Spring Cloud學(xué)習(xí)（十一）【深入Elasticsearch 分布式搜索引擎03】的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！