柚子快報(bào)邀請(qǐng)碼778899分享：分布式搜索引擎
柚子快報(bào)邀請(qǐng)碼778899分享：分布式搜索引擎
分布式搜索引擎03
0.學(xué)習(xí)目標(biāo)
1.數(shù)據(jù)聚合
**聚合（aggregations）**可以讓我們極其方便的實(shí)現(xiàn)對(duì)數(shù)據(jù)的統(tǒng)計(jì)、分析、運(yùn)算。例如：
什么品牌的手機(jī)最受歡迎？這些手機(jī)的平均價(jià)格、最高價(jià)格、最低價(jià)格？這些手機(jī)每月的銷售情況如何？
實(shí)現(xiàn)這些統(tǒng)計(jì)功能的比數(shù)據(jù)庫(kù)的sql要方便的多，而且查詢速度非常快，可以實(shí)現(xiàn)近實(shí)時(shí)搜索效果。
1.1.聚合的種類
聚合常見(jiàn)的有三類：
**桶（Bucket）**聚合：用來(lái)對(duì)文檔做分組
TermAggregation：按照文檔字段值分組，例如按照品牌值分組、按照國(guó)家分組Date Histogram：按照日期階梯分組，例如一周為一組，或者一月為一組 **度量（Metric）**聚合：用以計(jì)算一些值，比如：最大值、最小值、平均值等
Avg：求平均值Max：求最大值Min：求最小值Stats：同時(shí)求max、min、avg、sum等 **管道（pipeline）**聚合：其它聚合的結(jié)果為基礎(chǔ)做聚合
**注意：**參加聚合的字段必須是keyword、日期、數(shù)值、布爾類型
1.2.DSL實(shí)現(xiàn)聚合
現(xiàn)在，我們要統(tǒng)計(jì)所有數(shù)據(jù)中的酒店品牌有幾種，其實(shí)就是按照品牌對(duì)數(shù)據(jù)分組。此時(shí)可以根據(jù)酒店品牌的名稱做聚合，也就是Bucket聚合。
1.2.1.Bucket聚合語(yǔ)法
語(yǔ)法如下：
GET /hotel/_search
{
"size": 0, // 設(shè)置size為0，結(jié)果中不包含文檔，只包含聚合結(jié)果
"aggs": { // 定義聚合
"brandAgg": { //給聚合起個(gè)名字
"terms": { // 聚合的類型，按照品牌值聚合，所以選擇term
"field": "brand", // 參與聚合的字段
"size": 20 // 希望獲取的聚合結(jié)果數(shù)量
}
}
}
}
結(jié)果如圖：
1.2.2.聚合結(jié)果排序
默認(rèn)情況下，Bucket聚合會(huì)統(tǒng)計(jì)Bucket內(nèi)的文檔數(shù)量，記為_(kāi)count，并且按照_count降序排序。
我們可以指定order屬性，自定義聚合的排序方式：
GET /hotel/_search
{
"size": 0,
"aggs": {
"brandAgg": {
"terms": {
"field": "brand",
"order": {
"_count": "asc" // 按照_count升序排列
},
"size": 20
}
}
}
}
1.2.3.限定聚合范圍
默認(rèn)情況下，Bucket聚合是對(duì)索引庫(kù)的所有文檔做聚合，但真實(shí)場(chǎng)景下，用戶會(huì)輸入搜索條件，因此聚合必須是對(duì)搜索結(jié)果聚合。那么聚合必須添加限定條件。
我們可以限定要聚合的文檔范圍，只要添加query條件即可：
GET /hotel/_search
{
"query": {
"range": {
"price": {
"lte": 200
}
}
},
"size": 0,
"aggs": {
"brandAgg": {
"terms": {
"field": "brand",
"size": 20
}
}
}
}
這次，聚合得到的品牌明顯變少了：
1.2.4.Metric聚合語(yǔ)法
上節(jié)課，我們對(duì)酒店按照品牌分組，形成了一個(gè)個(gè)桶。現(xiàn)在我們需要對(duì)桶內(nèi)的酒店做運(yùn)算，獲取每個(gè)品牌的用戶評(píng)分的min、max、avg等值。
這就要用到Metric聚合了，例如stat聚合：就可以獲取min、max、avg等結(jié)果。
語(yǔ)法如下：
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
}
}
}
}
}
}
這次的score_stats聚合是在brandAgg的聚合內(nèi)部嵌套的子聚合。因?yàn)槲覀冃枰诿總€(gè)桶分別計(jì)算。
另外，我們還可以給聚合結(jié)果做個(gè)排序，例如按照每個(gè)桶的酒店平均分做排序：
1.2.5.小結(jié)
aggs代表聚合，與query同級(jí)，此時(shí)query的作用是？
限定聚合的的文檔范圍
聚合必須的三要素：
聚合名稱聚合類型聚合字段
聚合可配置屬性有：
size：指定聚合結(jié)果數(shù)量order：指定聚合結(jié)果排序方式field：指定聚合字段
1.3.RestAPI實(shí)現(xiàn)聚合
1.3.1.API語(yǔ)法
聚合條件與query條件同級(jí)別，因此需要使用request.source()來(lái)指定聚合條件。
聚合條件的語(yǔ)法：
聚合的結(jié)果也與查詢結(jié)果不同，API也比較特殊。不過(guò)同樣是JSON逐層解析：
1.3.2.業(yè)務(wù)需求
需求：搜索頁(yè)面的品牌、城市等信息不應(yīng)該是在頁(yè)面寫死，而是通過(guò)聚合索引庫(kù)中的酒店數(shù)據(jù)得來(lái)的：
分析：
目前，頁(yè)面的城市列表、星級(jí)列表、品牌列表都是寫死的，并不會(huì)隨著搜索結(jié)果的變化而變化。但是用戶搜索條件改變時(shí)，搜索結(jié)果會(huì)跟著變化。
例如：用戶搜索“東方明珠”，那搜索的酒店肯定是在上海東方明珠附近，因此，城市只能是上海，此時(shí)城市列表中就不應(yīng)該顯示北京、深圳、杭州這些信息了。
也就是說(shuō)，搜索結(jié)果中包含哪些城市，頁(yè)面就應(yīng)該列出哪些城市；搜索結(jié)果中包含哪些品牌，頁(yè)面就應(yīng)該列出哪些品牌。
如何得知搜索結(jié)果中包含哪些品牌？如何得知搜索結(jié)果中包含哪些城市？
使用聚合功能，利用Bucket聚合，對(duì)搜索結(jié)果中的文檔基于品牌分組、基于城市分組，就能得知包含哪些品牌、哪些城市了。
因?yàn)槭菍?duì)搜索結(jié)果聚合，因此聚合是限定范圍的聚合，也就是說(shuō)聚合的限定條件跟搜索文檔的條件一致。
查看瀏覽器可以發(fā)現(xiàn)，前端其實(shí)已經(jīng)發(fā)出了這樣的一個(gè)請(qǐng)求：
請(qǐng)求參數(shù)與搜索文檔的參數(shù)完全一致。
返回值類型就是頁(yè)面要展示的最終結(jié)果：
結(jié)果是一個(gè)Map結(jié)構(gòu)：
key是字符串，城市、星級(jí)、品牌、價(jià)格value是集合，例如多個(gè)城市的名稱
1.3.3.業(yè)務(wù)實(shí)現(xiàn)
在cn.itcast.hotel.web包的HotelController中添加一個(gè)方法，遵循下面的要求：
請(qǐng)求方式：POST請(qǐng)求路徑：/hotel/filters請(qǐng)求參數(shù)：RequestParams，與搜索文檔的參數(shù)一致返回值類型：Map
代碼：
@PostMapping("filters")
public Map
return hotelService.getFilters(params);
}
這里調(diào)用了IHotelService中的getFilters方法，尚未實(shí)現(xiàn)。
在cn.itcast.hotel.service.IHotelService中定義新方法：
Map
在cn.itcast.hotel.service.impl.HotelService中實(shí)現(xiàn)該方法：
@Override
public Map
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
Aggregations aggregations = response.getAggregations();
// 4.1.根據(jù)品牌名稱，獲取品牌結(jié)果
List
result.put("brand", brandList);
// 4.2.根據(jù)品牌名稱，獲取品牌結(jié)果
List
result.put("city", cityList);
// 4.3.根據(jù)品牌名稱，獲取品牌結(jié)果
List
result.put("starName", starList);
return result;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
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("starName")
.size(100)
);
}
private List
// 4.1.根據(jù)聚合名稱獲取聚合結(jié)果
Terms brandTerms = aggregations.get(aggName);
// 4.2.獲取buckets
List buckets = brandTerms.getBuckets();
// 4.3.遍歷
List
for (Terms.Bucket bucket : buckets) {
// 4.4.獲取key
String key = bucket.getKeyAsString();
brandList.add(key);
}
return brandList;
}
2.自動(dòng)補(bǔ)全
當(dāng)用戶在搜索框輸入字符時(shí)，我們應(yīng)該提示出與該字符有關(guān)的搜索項(xiàng)，如圖：
這種根據(jù)用戶輸入的字母，提示完整詞條的功能，就是自動(dòng)補(bǔ)全了。
因?yàn)樾枰鶕?jù)拼音字母來(lái)推斷，因此要用到拼音分詞功能。
2.1.拼音分詞器
要實(shí)現(xiàn)根據(jù)字母做補(bǔ)全，就必須對(duì)文檔按照拼音分詞。在GitHub上恰好有elasticsearch的拼音分詞插件。地址：https://github.com/medcl/elasticsearch-analysis-pinyin
課前資料中也提供了拼音分詞器的安裝包：
安裝方式與IK分詞器一樣，分三步：
? ①解壓
? ②上傳到虛擬機(jī)中，elasticsearch的plugin目錄
? ③重啟elasticsearch
? ④測(cè)試
詳細(xì)安裝步驟可以參考IK分詞器的安裝過(guò)程。
測(cè)試用法如下：
POST /_analyze
{
"text": "如家酒店還不錯(cuò)",
"analyzer": "pinyin"
}
結(jié)果：
2.2.自定義分詞器
默認(rèn)的拼音分詞器會(huì)將每個(gè)漢字單獨(dú)分為拼音，而我們希望的是每個(gè)詞條形成一組拼音，需要對(duì)拼音分詞器做個(gè)性化定制，形成自定義分詞器。
elasticsearch中分詞器（analyzer）的組成包含三部分：
character filters：在tokenizer之前對(duì)文本進(jìn)行處理。例如刪除字符、替換字符tokenizer：將文本按照一定的規(guī)則切割成詞條（term）。例如keyword，就是不分詞；還有ik_smarttokenizer filter：將tokenizer輸出的詞條做進(jìn)一步處理。例如大小寫轉(zhuǎn)換、同義詞處理、拼音處理等
文檔分詞時(shí)會(huì)依次由這三部分來(lái)處理文檔：
聲明自定義分詞器的語(yǔ)法如下：
PUT /test
{
"settings": {
"analysis": {
"analyzer": { // 自定義分詞器
"my_analyzer": { // 分詞器名稱
"tokenizer": "ik_max_word",
"filter": "py"
}
},
"filter": { // 自定義tokenizer filter
"py": { // 過(guò)濾器名稱
"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", # 保存文檔內(nèi)容時(shí)，使用自定義分詞器-》寫
"search_analyzer": "ik_smart" # 搜索時(shí)使用id_smart ---》讀
}
}
}
}
參數(shù)詳細(xì)說(shuō)明：
keep_first_letter:這個(gè)參數(shù)會(huì)將詞的第一個(gè)字母全部拼起來(lái).例如:劉德華->ldh.默認(rèn)為:true
keep_separate_first_letter:這個(gè)會(huì)將第一個(gè)字母一個(gè)個(gè)分開(kāi).例如:劉德華->l,d,h.默認(rèn)為:flase.如果開(kāi)啟,可能導(dǎo)致查詢結(jié)果太過(guò)于模糊,準(zhǔn)確率太低.
limit_first_letter_length:設(shè)置最大keep_first_letter結(jié)果的長(zhǎng)度,默認(rèn)為:16
keep_full_pinyin:如果打開(kāi),它將保存詞的全拼,并按字分開(kāi)保存.例如:劉德華> [liu,de,hua],默認(rèn)為:true
keep_joined_full_pinyin:如果打開(kāi)將保存詞的全拼.例如:劉德華> [liudehua],默認(rèn)為:false
keep_none_chinese:將非中文字母或數(shù)字保留在結(jié)果中.默認(rèn)為:true
keep_none_chinese_together:保證非中文在一起.默認(rèn)為: true, 例如: DJ音樂(lè)家 -> DJ,yin,yue,jia, 如果設(shè)置為:false, 例如: DJ音樂(lè)家 -> D,J,yin,yue,jia, 注意: keep_none_chinese應(yīng)該先開(kāi)啟.
keep_none_chinese_in_first_letter:將非中文字母保留在首字母中.例如: 劉德華AT2016->ldhat2016, 默認(rèn)為:true
keep_none_chinese_in_joined_full_pinyin:將非中文字母保留為完整拼音. 例如: 劉德華2016->liudehua2016, 默認(rèn)為: false
none_chinese_pinyin_tokenize:如果他們是拼音,切分非中文成單獨(dú)的拼音項(xiàng). 默認(rèn)為:true,例如: liudehuaalibaba13zhuanghan -> liu,de,hua,a,li,ba,ba,13,zhuang,han, 注意: keep_none_chinese和keep_none_chinese_together需要先開(kāi)啟.
keep_original:是否保持原詞.默認(rèn)為:false
lowercase:小寫非中文字母.默認(rèn)為:true
trim_whitespace:去掉空格.默認(rèn)為:true
remove_duplicated_term:保存索引時(shí)刪除重復(fù)的詞語(yǔ).例如: de的>de, 默認(rèn)為: false, 注意:開(kāi)啟可能會(huì)影響位置相關(guān)的查詢.
ignore_pinyin_offset:在6.0之后,嚴(yán)格限制偏移量,不允許使用重疊的標(biāo)記.使用此參數(shù)時(shí),忽略偏移量將允許使用重疊的標(biāo)記.請(qǐng)注意,所有與位置相關(guān)的查詢或突出顯示都將變?yōu)殄e(cuò)誤,您應(yīng)使用多個(gè)字段并為不同的字段指定不同的設(shè)置查詢目的.如果需要偏移量,請(qǐng)將其設(shè)置為false。默認(rèn)值:true
測(cè)試：
總結(jié)：
如何使用拼音分詞器？
①下載pinyin分詞器 ②解壓并放到elasticsearch的plugin目錄 ③重啟即可
如何自定義分詞器？
①創(chuàng)建索引庫(kù)時(shí)，在settings中配置，可以包含三部分 ②character filter ③tokenizer ④filter
拼音分詞器注意事項(xiàng)？
為了避免搜索到同音字，搜索時(shí)不要使用拼音分詞器
2.3.自動(dòng)補(bǔ)全查詢
elasticsearch提供了Completion Suggester查詢來(lái)實(shí)現(xiàn)自動(dòng)補(bǔ)全功能。這個(gè)查詢會(huì)匹配以用戶輸入內(nèi)容開(kāi)頭的詞條并返回。為了提高補(bǔ)全查詢的效率，對(duì)于文檔中字段的類型有一些約束：
參與補(bǔ)全查詢的字段必須是completion類型。 字段的內(nèi)容一般是用來(lái)補(bǔ)全的多個(gè)詞條形成的數(shù)組。
比如，一個(gè)這樣的索引庫(kù)：
PUT test
{
"mappings": {
"properties": {
"title":{
"type": "completion"
}
}
}
}
然后插入下面的數(shù)據(jù)：
POST test/_doc
{
"title": ["Sony", "WH-1000XM3"]
}
POST test/_doc
{
"title": ["SK-II", "PITERA"]
}
POST test/_doc
{
"title": ["Nintendo", "switch"]
}
查詢的DSL語(yǔ)句如下：
// 自動(dòng)補(bǔ)全查詢
GET /test/_search
{
"suggest": {
"title_suggest": {
"text": "s", // 關(guān)鍵字
"completion": {
"field": "title", // 補(bǔ)全查詢的字段
"skip_duplicates": true, // 跳過(guò)重復(fù)的
"size": 10 // 獲取前10條結(jié)果
}
}
}
}
2.4.實(shí)現(xiàn)酒店搜索框自動(dòng)補(bǔ)全
現(xiàn)在，我們的hotel索引庫(kù)還沒(méi)有設(shè)置拼音分詞器，需要修改索引庫(kù)中的配置。但是我們知道索引庫(kù)是無(wú)法修改的，只能刪除然后重新創(chuàng)建。
另外，我們需要添加一個(gè)字段，用來(lái)做自動(dòng)補(bǔ)全，將brand、suggestion、city等都放進(jìn)去，作為自動(dòng)補(bǔ)全的提示。
因此，總結(jié)一下，我們需要做的事情包括：
修改hotel索引庫(kù)結(jié)構(gòu)，設(shè)置自定義拼音分詞器 修改索引庫(kù)的name、all字段，使用自定義分詞器 索引庫(kù)添加一個(gè)新字段suggestion，類型為completion類型，使用自定義的分詞器 給HotelDoc類添加suggestion字段，內(nèi)容包含brand、business 重新導(dǎo)入數(shù)據(jù)到hotel庫(kù)
2.4.1.修改酒店映射結(jié)構(gòu)
代碼如下：
// 酒店數(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"
}
}
}
}
2.4.2.修改HotelDoc實(shí)體
HotelDoc中要添加一個(gè)字段，用來(lái)做自動(dòng)補(bǔ)全，內(nèi)容可以是酒店品牌、城市、商圈等信息。按照自動(dòng)補(bǔ)全字段的要求，最好是這些字段的數(shù)組。
因此我們?cè)贖otelDoc中添加一個(gè)suggestion字段，類型為L(zhǎng)ist
代碼如下：
package cn.itcast.hotel.pojo;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@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;
private Object distance;
private Boolean isAD;
private List
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();
// 組裝suggestion
if(this.business.contains("/")){
// business有多個(gè)值，需要切割
String[] arr = this.business.split("/");
// 添加元素
this.suggestion = new ArrayList<>();
this.suggestion.add(this.brand);
Collections.addAll(this.suggestion, arr);
}else {
this.suggestion = Arrays.asList(this.brand, this.business);
}
}
}
2.4.3.重新導(dǎo)入
重新執(zhí)行之前編寫的導(dǎo)入數(shù)據(jù)功能，可以看到新的酒店數(shù)據(jù)中包含了suggestion：
2.4.4.自動(dòng)補(bǔ)全查詢的JavaAPI
之前我們學(xué)習(xí)了自動(dòng)補(bǔ)全查詢的DSL，而沒(méi)有學(xué)習(xí)對(duì)應(yīng)的JavaAPI，這里給出一個(gè)示例：
而自動(dòng)補(bǔ)全的結(jié)果也比較特殊，解析的代碼如下：
2.4.5.實(shí)現(xiàn)搜索框自動(dòng)補(bǔ)全
查看前端頁(yè)面，可以發(fā)現(xiàn)當(dāng)我們?cè)谳斎肟蜴I入時(shí)，前端會(huì)發(fā)起ajax請(qǐng)求：
返回值是補(bǔ)全詞條的集合，類型為L(zhǎng)ist
1）在cn.itcast.hotel.web包下的HotelController中添加新接口，接收新的請(qǐng)求：
@GetMapping("suggestion")
public List
return hotelService.getSuggestions(prefix);
}
2）在cn.itcast.hotel.service包下的IhotelService中添加方法：
List
3）在cn.itcast.hotel.service.impl.HotelService中實(shí)現(xiàn)該方法：
@Override
public List
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
// 4.3.遍歷
List
for (CompletionSuggestion.Entry.Option option : options) {
String text = option.getText().toString();
list.add(text);
}
return list;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
3.數(shù)據(jù)同步
elasticsearch中的酒店數(shù)據(jù)來(lái)自于mysql數(shù)據(jù)庫(kù)，因此mysql數(shù)據(jù)發(fā)生改變時(shí)，elasticsearch也必須跟著改變，這個(gè)就是elasticsearch與mysql之間的數(shù)據(jù)同步。
3.1.思路分析
常見(jiàn)的數(shù)據(jù)同步方案有三種：
同步調(diào)用異步通知監(jiān)聽(tīng)binlog
3.1.1.同步調(diào)用
方案一：同步調(diào)用
基本步驟如下：
hotel-demo對(duì)外提供接口，用來(lái)修改elasticsearch中的數(shù)據(jù)酒店管理服務(wù)在完成數(shù)據(jù)庫(kù)操作后，直接調(diào)用hotel-demo提供的接口，
3.1.2.異步通知
方案二：異步通知
流程如下：
hotel-admin對(duì)mysql數(shù)據(jù)庫(kù)數(shù)據(jù)完成增、刪、改后，發(fā)送MQ消息hotel-demo監(jiān)聽(tīng)MQ，接收到消息后完成elasticsearch數(shù)據(jù)修改
3.1.3.監(jiān)聽(tīng)binlog
方案三：監(jiān)聽(tīng)binlog
流程如下：
給mysql開(kāi)啟binlog功能mysql完成增、刪、改操作都會(huì)記錄在binlog中hotel-demo基于canal監(jiān)聽(tīng)binlog變化，實(shí)時(shí)更新elasticsearch中的內(nèi)容
3.1.4.選擇
方式一：同步調(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)聽(tīng)binlog
優(yōu)點(diǎn)：完全解除服務(wù)間耦合缺點(diǎn)：開(kāi)啟binlog增加數(shù)據(jù)庫(kù)負(fù)擔(dān)、實(shí)現(xiàn)復(fù)雜度高
3.2.實(shí)現(xiàn)數(shù)據(jù)同步
3.2.1.思路
利用課前資料提供的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)聽(tīng)，并更新elasticsearch中數(shù)據(jù) 啟動(dòng)并測(cè)試數(shù)據(jù)同步功能
3.2.2.導(dǎo)入demo
導(dǎo)入課前資料提供的hotel-admin項(xiàng)目：
運(yùn)行后，訪問(wèn) http://localhost:8099
其中包含了酒店的CRUD功能：
3.2.3.聲明交換機(jī)、隊(duì)列
MQ結(jié)構(gòu)如圖：
1）引入依賴
在hotel-admin、hotel-demo中引入rabbitmq的依賴：
2）聲明隊(duì)列交換機(jī)名稱
在hotel-admin和hotel-demo中的cn.itcast.hotel.constatnts包下新建一個(gè)類MqConstants：
package cn.itcast.hotel.constatnts;
public class MqConstants {
/**
* 交換機(jī)
*/
public final static String HOTEL_EXCHANGE = "hotel.topic";
/**
* 監(jiān)聽(tīng)新增和修改的隊(duì)列
*/
public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
/**
* 監(jiān)聽(tīng)刪除的隊(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";
}
3）聲明隊(duì)列交換機(jī)
在hotel-admin中，定義配置類，聲明隊(duì)列、交換機(jī)：
package cn.itcast.hotel.config;
import cn.itcast.hotel.constants.MqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@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);
}
}
docker中要啟動(dòng)mq容器：
docker start mq
# 如果想設(shè)置開(kāi)機(jī) 啟動(dòng)mq
docker update --restart always mq
hotel-admin 及 hotel-demo工程中 引入mq的連接參數(shù)配置
spring:
rabbitmq:
virtual-host: /
port: 5672
host: 192.168.200.130
username: itcast
password: 123321
3.2.4.發(fā)送MQ消息
在hotel-admin中的增、刪、改業(yè)務(wù)中分別發(fā)送MQ消息：
3.2.5.接收MQ消息
hotel-demo接收到MQ消息要做的事情包括：
新增消息：根據(jù)傳遞的hotel的id查詢hotel信息，然后新增一條數(shù)據(jù)到索引庫(kù)刪除消息：根據(jù)傳遞的hotel的id刪除索引庫(kù)中的一條數(shù)據(jù)
1）首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、刪除業(yè)務(wù)
void deleteById(Long id);
void insertById(Long id);
2）給hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中實(shí)現(xiàn)業(yè)務(wù)：
@Override
public void deleteById(Long id) {
try {
// 1.準(zhǔn)備Request
DeleteRequest request = new DeleteRequest("hotel", id.toString());
// 2.發(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);
}
}
3）編寫監(jiān)聽(tīng)器
在hotel-demo中的cn.itcast.hotel.mq包新增一個(gè)類：
package cn.itcast.hotel.mq;
import cn.itcast.hotel.constants.MqConstants;
import cn.itcast.hotel.service.IHotelService;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class HotelListener {
@Autowired
private IHotelService hotelService;
/**
* 監(jiān)聽(tīng)酒店新增或修改的業(yè)務(wù)
* @param id 酒店id
*/
@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
public void listenHotelInsertOrUpdate(Long id){
hotelService.insertById(id);
}
/**
* 監(jiān)聽(tīng)酒店刪除的業(yè)務(wù)
* @param id 酒店id
*/
@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
public void listenHotelDelete(Long id){
hotelService.deleteById(id);
}
}
4.集群
單機(jī)的elasticsearch做數(shù)據(jù)存儲(chǔ)，必然面臨兩個(gè)問(wèn)題：海量數(shù)據(jù)存儲(chǔ)問(wèn)題、單點(diǎn)故障問(wèn)題。
海量數(shù)據(jù)存儲(chǔ)問(wèn)題：將索引庫(kù)從邏輯上拆分為N個(gè)分片（shard），存儲(chǔ)到多個(gè)節(jié)點(diǎn)單點(diǎn)故障問(wèn)題：將分片數(shù)據(jù)在不同節(jié)點(diǎn)備份（replica ）
ES集群相關(guān)概念:
集群（cluster）：一組擁有共同的 cluster name 的 節(jié)點(diǎn)。 節(jié)點(diǎn)（node) ：集群中的一個(gè) Elasticearch 實(shí)例 分片（shard）：索引可以被拆分為不同的部分進(jìn)行存儲(chǔ)，稱為分片。在集群環(huán)境下，一個(gè)索引的不同分片可以拆分到不同的節(jié)點(diǎn)中 解決問(wèn)題：數(shù)據(jù)量太大，單點(diǎn)存儲(chǔ)量有限的問(wèn)題。
此處，我們把數(shù)據(jù)分成3片：shard0、shard1、shard2
主分片（Primary shard）：相對(duì)于副本分片的定義。 副本分片（Replica shard）每個(gè)主分片可以有一個(gè)或者多個(gè)副本，數(shù)據(jù)和主分片一樣。 ?
數(shù)據(jù)備份可以保證高可用，但是每個(gè)分片備份一份，所需要的節(jié)點(diǎn)數(shù)量就會(huì)翻一倍，成本實(shí)在是太高了！
為了在高可用和成本間尋求平衡，我們可以這樣做：
首先對(duì)數(shù)據(jù)分片，存儲(chǔ)到不同節(jié)點(diǎn)然后對(duì)每個(gè)分片進(jìn)行備份，放到對(duì)方節(jié)點(diǎn)，完成互相備份
這樣可以大大減少所需要的服務(wù)節(jié)點(diǎn)數(shù)量，如圖，我們以3分片，每個(gè)分片備份一份為例：
現(xiàn)在，每個(gè)分片都有1個(gè)備份，存儲(chǔ)在3個(gè)節(jié)點(diǎn)：
node0：保存了分片0和1node1：保存了分片0和2node2：保存了分片1和2
4.1.搭建ES集群
參考課前資料的文檔：
其中的第四章節(jié)：
4.2.集群腦裂問(wèn)題
4.2.1.集群職責(zé)劃分
elasticsearch中集群節(jié)點(diǎn)有不同的職責(zé)劃分：
默認(rèn)情況下，集群中的任何一個(gè)節(jié)點(diǎn)都同時(shí)具備上述四種角色。
但是真實(shí)的集群一定要將集群職責(zé)分離：
master節(jié)點(diǎn)：對(duì)CPU要求高，但是內(nèi)存要求低data節(jié)點(diǎn)：對(duì)CPU和內(nèi)存要求都高coordinating節(jié)點(diǎn)：對(duì)網(wǎng)絡(luò)帶寬、CPU要求高
職責(zé)分離可以讓我們根據(jù)不同節(jié)點(diǎn)的需求分配不同的硬件去部署。而且避免業(yè)務(wù)之間的互相干擾。
一個(gè)典型的es集群職責(zé)劃分如圖：
4.2.2.腦裂問(wèn)題
腦裂是因?yàn)榧褐械墓?jié)點(diǎn)失聯(lián)導(dǎo)致的。
例如一個(gè)集群中，主節(jié)點(diǎn)與其它節(jié)點(diǎn)失聯(lián)：
此時(shí)，node2和node3認(rèn)為node1宕機(jī)，就會(huì)重新選主：
當(dāng)node3當(dāng)選后，集群繼續(xù)對(duì)外提供服務(wù)，node2和node3自成集群，node1自成集群，兩個(gè)集群數(shù)據(jù)不同步，出現(xiàn)數(shù)據(jù)差異。
當(dāng)網(wǎng)絡(luò)恢復(fù)后，因?yàn)榧褐杏袃蓚€(gè)master節(jié)點(diǎn)，集群狀態(tài)的不一致，出現(xiàn)腦裂的情況：
解決腦裂的方案是，要求選票超過(guò) ( 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ā)生腦裂問(wèn)題
例如：3個(gè)節(jié)點(diǎn)形成的集群，選票必須超過(guò) （3 + 1） / 2 ，也就是2票。node3得到node2和node3的選票，當(dāng)選為主。node1只有自己1票，沒(méi)有當(dāng)選。集群中依然只有1個(gè)主節(jié)點(diǎn)，沒(méi)有出現(xiàn)腦裂。
4.2.3.小結(jié)
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é)果，返回給用戶
4.3.集群分布式存儲(chǔ)
當(dāng)新增文檔時(shí)，應(yīng)該保存到不同分片，保證數(shù)據(jù)均衡，那么coordinating node如何確定數(shù)據(jù)該存儲(chǔ)到哪個(gè)分片呢？
4.3.1.分片存儲(chǔ)測(cè)試
插入三條數(shù)據(jù)：
測(cè)試可以看到，三條數(shù)據(jù)分別在不同分片：
結(jié)果：
4.3.2.分片存儲(chǔ)原理
elasticsearch會(huì)通過(guò)hash算法來(lái)計(jì)算文檔應(yīng)該存儲(chǔ)到哪個(gè)分片：
說(shuō)明：
_routing默認(rèn)是文檔的id算法與分片數(shù)量有關(guān)，因此索引庫(kù)一旦創(chuàng)建，分片數(shù)量不能修改！
新增文檔的流程如下：
解讀：
1）新增一個(gè)id=1的文檔2）對(duì)id做hash運(yùn)算，假如得到的是2，則應(yīng)該存儲(chǔ)到shard-23）shard-2的主分片在node3節(jié)點(diǎn)，將數(shù)據(jù)路由到node34）保存文檔5）同步給shard-2的副本replica-2，在node2節(jié)點(diǎn)6）返回結(jié)果給coordinating-node節(jié)點(diǎn)
過(guò)程細(xì)節(jié)描述：
集群寫入時(shí)，會(huì)先隨機(jī)選取一個(gè)節(jié)點(diǎn)（node），該節(jié)點(diǎn)可以稱之為“協(xié)調(diào)節(jié)點(diǎn)”。 新文檔寫入前，es會(huì)對(duì)其id做hash取模，來(lái)確定該文檔會(huì)分布在哪個(gè)分片上。 當(dāng)分片位置確定好后，es會(huì)判圖當(dāng)前“協(xié)調(diào)節(jié)點(diǎn)”上是否有該主分片。如果有，直接寫；如果沒(méi)有，則會(huì)將數(shù)據(jù)路由到包含該主分片的節(jié)點(diǎn)上。 整個(gè)寫入過(guò)程是，es會(huì)將文檔先寫入主分片上（如p0），寫完后再將數(shù)據(jù)同步一份到副本上（如r0） 待副本數(shù)據(jù)也寫完后，副本節(jié)點(diǎn)會(huì)通知協(xié)調(diào)節(jié)點(diǎn)，最后協(xié)調(diào)節(jié)點(diǎn)告知客戶端，文檔寫入結(jié)束。
4.4.集群分布式查詢
elasticsearch的查詢分成兩個(gè)階段：
scatter phase：分散階段，coordinating node會(huì)把請(qǐng)求分發(fā)到每一個(gè)分片 gather phase：聚集階段，coordinating node匯總data node的搜索結(jié)果，并處理為最終結(jié)果集返回給用戶
4.5.集群故障轉(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)移。
1）例如一個(gè)集群結(jié)構(gòu)如圖：
現(xiàn)在，node1是主節(jié)點(diǎn)，其它兩個(gè)節(jié)點(diǎn)是從節(jié)點(diǎn)。
2）突然，node1發(fā)生了故障：
宕機(jī)后的第一件事，需要重新選主，例如選中了node2：
node2成為主節(jié)點(diǎn)后，會(huì)檢測(cè)集群監(jiān)控狀態(tài)，發(fā)現(xiàn)：shard-1、shard-0沒(méi)有副本節(jié)點(diǎn)。因此需要將node1上的數(shù)據(jù)遷移到node2、node3：
柚子快報(bào)邀請(qǐng)碼778899分享：分布式搜索引擎
好文閱讀
本文內(nèi)容根據(jù)網(wǎng)絡(luò)資料整理，出于傳遞更多信息之目的，不代表金鑰匙跨境贊同其觀點(diǎn)和立場(chǎng)。
轉(zhuǎn)載請(qǐng)注明，如有侵權(quán)，聯(lián)系刪除。
Kaufland歐洲優(yōu)選倉(cāng)
