為什么使用elasticsearch?數(shù)據(jù)很多么？

項目使用Elasticsearch是實現(xiàn)了門戶上對服務(wù)的搜索。

平臺上的服務(wù)數(shù)據(jù)是并不是很多，全國所有區(qū)域下的服務(wù)信息加一起幾千條，之所以使用Elasticsearch是因為：

1、公司架構(gòu)師在系統(tǒng)架構(gòu)時考慮幾年后的數(shù)據(jù)及對全文檢索使用的需求使用了Elasticsearch.

2、對服務(wù)信息進(jìn)行搜索使用的是全文檢索方式，雖然MySQL也支持全文檢索但是我們這個接口是面向 C端用戶且對接口性能有要求，所以使用了ES。

3、雖然現(xiàn)在數(shù)據(jù)量不大考慮幾年后的數(shù)據(jù)量增長問題，我們使用了Elasticsearch。

4、在項目中除了通過關(guān)鍵字搜索服務(wù)信息，還有根據(jù)地理坐標(biāo)進(jìn)行搜索，使用Elasticsearch也考慮了這一點。

實現(xiàn)目標(biāo)：

對于服務(wù)類型和服務(wù)項的名稱進(jìn)行索引，當(dāng)搜索服務(wù)項名稱時，展示出該服務(wù)項，當(dāng)搜索服務(wù)類型時，展示出該類型下的所有服務(wù)項。

方案一

?因為操作es和數(shù)據(jù)庫 是分布式事務(wù)，無法控制一致性。

方案二

使用Canal+MQ

Canal可與很多數(shù)據(jù)源進(jìn)行對接，將數(shù)據(jù)由MySQL同步到ES、MQ、DB等各個數(shù)據(jù)源。

Canal的意思是水道/管道/溝渠，它相當(dāng)于一個數(shù)據(jù)管道，通過解析MySQL的binlog日志完成數(shù)據(jù)同步工作。

要理解上圖中Canal的工作原理需要首先要知道MySQL主從數(shù)據(jù)同步的原理，如下圖：

?MySQL主從數(shù)據(jù)同步的原理

MySQL主從集群由MySQL主服務(wù)器(master)和MySQL從服務(wù)器(slave)組成，MySQL主從數(shù)據(jù)同步是一種數(shù)據(jù)庫復(fù)制技術(shù)，進(jìn)行寫數(shù)據(jù)會先向主服務(wù)器寫，寫成功后將數(shù)據(jù)同步到從服務(wù)器，流程如下：

1、主服務(wù)器將所有寫操作（INSERT、UPDATE、DELETE）以二進(jìn)制日志（binlog）的形式記錄下來。

2、從服務(wù)器連接到主服務(wù)器，發(fā)送dump 協(xié)議,請求獲取主服務(wù)器上的binlog日志。

MySQL的dump協(xié)議是MySQL復(fù)制協(xié)議中的一部分。

3、MySQL master 收到 dump 請求，開始推送 binary log 給 slave

4、從服務(wù)器解析日志，根據(jù)日志內(nèi)容更新從服務(wù)器的數(shù)據(jù)庫，完成從服務(wù)器的數(shù)據(jù)保持與主服務(wù)器同步。

理解了MySQL主從同步的原理，Canal在整個過程充當(dāng)什么角色呢？

如下圖：

工作流程如下：

1、Canal模擬 MySQL slave 的交互協(xié)議，偽裝自己為 MySQL slave ，向 MySQL master 發(fā)送dump 協(xié)議

MySQL的dump協(xié)議是MySQL復(fù)制協(xié)議中的一部分。

2、MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal )

。一旦連接建立成功，Canal會一直等待并監(jiān)聽來自MySQL主服務(wù)器的binlog事件流，當(dāng)有新的數(shù)據(jù)庫變更發(fā)生時MySQL master主服務(wù)器發(fā)送binlog事件流給Canal。

3、Canal會及時接收并解析這些變更事件并解析 binary log

通過以上流程可知Canal和MySQL master主服務(wù)器之間建立了長連接。

流程：運(yùn)營端人員修改信息，canal監(jiān)聽到mysql的binlog日志并進(jìn)行解析，發(fā)送到mq,mq監(jiān)聽到消息?，同步簡歷索引

MQ技術(shù)方案

目標(biāo)：

能說出如何保證MQ消息的可靠性？

1）保證生產(chǎn)消息可靠性

上述技術(shù)方案中有一個關(guān)鍵點，首先數(shù)據(jù)增刪改的信息是保證寫入binlog的，Canal解析出增刪改的信息后寫入MQ，同步程序從MQ去讀取消息，如果MQ中的消息丟失了數(shù)據(jù)將無法進(jìn)行同步。

如何保證MQ消息的可靠性？

保證MQ消息的可靠性分兩個方面：保證生產(chǎn)消息的可靠性、保證消費(fèi)消息的可靠性。

保證生產(chǎn)消息的可靠性：

RabbitMQ提供生產(chǎn)者確認(rèn)機(jī)制保證生產(chǎn)消息的可靠性，技術(shù)方案如下?

• 首先發(fā)送消息的方法如果執(zhí)行失敗會進(jìn)行重試，重試次數(shù)耗盡記錄失敗消息

• 

如果重試失敗，則將數(shù)據(jù)保存到數(shù)據(jù)庫中

• 通過MQ的提供的生產(chǎn)者確認(rèn)機(jī)制保證生產(chǎn)消息的可靠性

使用生產(chǎn)者確認(rèn)機(jī)制需要給每個消息指定一個唯一ID，生產(chǎn)者確認(rèn)機(jī)制通過異步回調(diào)的方式進(jìn)行，包括ConfirmCallback和Return回調(diào)。

ConfirmCallback：消息發(fā)送到Broker會有一個結(jié)果返回給發(fā)送者表示消息是否處理成功：

1）消息成功投遞到交換機(jī)，返回ack

2）消息未投遞到交換機(jī)，返回nack

在發(fā)送消息時指定回調(diào)對象

回調(diào)類中回調(diào)方法源代碼：

如果沒有返回ack則將消息記錄到失敗消息表，如果經(jīng)過重試后返回了ack說明消息發(fā)送成功，此時將消息從失敗消息表刪除。

Return回調(diào)：如果消息發(fā)送到交換機(jī)成功了但是并沒有到達(dá)隊列，此時會調(diào)用ReturnCallback回調(diào)方法，在回調(diào)方法中我們可以收到失敗的消息存入失敗消息表以便進(jìn)行補(bǔ)償。

要使用Return回調(diào)需要開啟設(shè)置：

首先在shared-rabbitmq.yaml中配置rabbitMQ參數(shù)，如下：

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true說明：
publish-confirm-type：開啟publisher-confirm，這里支持兩種類型：
simple：同步等待confirm結(jié)果，直到超時
correlated：異步回調(diào)，定義ConfirmCallback，MQ返回結(jié)果時會回調(diào)這個ConfirmCallback
publish-returns：開啟publish-return功能，同樣是基于callback機(jī)制，不過是定義ReturnCallback
template.mandatory：定義消息路由失敗時的策略。true，則調(diào)用ReturnCallback；false：則直接丟棄消息

保證消費(fèi)消息可靠性

首先設(shè)置消息持久化，保證消息發(fā)送到MQ消息不丟失。具體需要設(shè)置交換機(jī)和隊列支持持久化，發(fā)送消息設(shè)置deliveryMode=2。

RabbitMQ是通過消費(fèi)者回執(zhí)來確認(rèn)消費(fèi)者是否成功處理消息的：消費(fèi)者獲取消息后，應(yīng)該向RabbitMQ發(fā)送ACK回執(zhí)，表明自己已經(jīng)處理完成消息，RabbitMQ收到ACK后刪除消息。

spring:rabbitmq:....listener:simple:acknowledge-mode: auto #，出現(xiàn)異常時返回nack，消息回滾到mq；沒有異常，返回ackretry:enabled: true # 開啟消費(fèi)者失敗重試initial-interval: 1000 # 初識的失敗等待時長為1秒multiplier: 10 # 失敗的等待時長倍數(shù)，下次等待時長 = multiplier * last-intervalmax-attempts: 3 # 最大重試次數(shù)stateless: true # true無狀態(tài)；false有狀態(tài)。如果業(yè)務(wù)中包含事務(wù)，這里改為false

能說出如何保證MQ冪等性？或 如何防止重復(fù)消費(fèi)？

?消費(fèi)者在消費(fèi)消息時難免出現(xiàn)重復(fù)消費(fèi)的情況，比如：消費(fèi)者沒有向MQ返回ack導(dǎo)致重復(fù)消費(fèi)，所以消費(fèi)者需要保證消費(fèi)消息冪等性。

什么是冪等性？

冪等性是指不論執(zhí)行多少次其結(jié)果是一致的。

舉例：

收到消息需要向數(shù)據(jù)新增一條記錄，如果重復(fù)消費(fèi)則會出現(xiàn)重復(fù)添加記錄的問題。

下邊根據(jù)場景分析解決方案：

1、查詢操作

本身具有冪等性。

2、添加操作

如果主鍵是自增則可能重復(fù)添加記錄。

保證冪等性可以設(shè)置數(shù)據(jù)庫的唯一約束，比如：添加學(xué)生信息，將學(xué)號字段設(shè)置為唯一索引，即使重復(fù)添加相同的學(xué)生同一個學(xué)號只會添加一條記錄。

3、更新操作

如果是更新一個固定的值，比如: update users set status =1 where id=?，本身具有冪等性。

如果只允許更新成功一次則可以使用token機(jī)制，發(fā)送消息前生成一個token寫入redis，收到消息后解析出token從redis查詢token如果成功則說明沒有消費(fèi)，此時更新成功將token從redis刪除，當(dāng)重復(fù)消費(fèi)相同 的消息時由于token已經(jīng)從redis刪除不會再執(zhí)行更新操作。

4、刪除操作

與更新操作類似，如果是刪除某個具體的記錄，比如：delete from users where id=?，本身具有冪等性。

如果只允許刪除成功一次可以采用更新操作相同的方法。

可以百分百保證MQ的消息可靠性嗎？

保證消息可靠性分兩個方面：保證生產(chǎn)消息可靠性和保證消費(fèi)消息可靠性。

保證生產(chǎn)消息可靠性：

生產(chǎn)消息可靠性是通過判斷MQ是否發(fā)送ack回執(zhí)，如果發(fā)nack表示發(fā)送消息失敗，此時會進(jìn)行重發(fā)或記錄到失敗消息表，通過定時任務(wù)進(jìn)行補(bǔ)償發(fā)送。如果Java程序并沒有收到回執(zhí)（如jvm進(jìn)程異常結(jié)束了，或斷電等因素），此時將無法保證生產(chǎn)消息的可靠性。

保證消費(fèi)消息可靠性：

保證消費(fèi)消息可靠性方案首先保證發(fā)送消息設(shè)置為持久化，其次通過MQ的消費(fèi)確認(rèn)機(jī)制保證消費(fèi)者消費(fèi)成功消息后再將消息刪除。

雖然設(shè)置了消息持久化，消息進(jìn)入MQ首先是在緩存存在，MQ會根據(jù)一定的規(guī)則進(jìn)行刷盤，比如：每隔幾毫秒進(jìn)行刷盤，如果在消息還沒有保存到磁盤時MQ進(jìn)程終止，此時將會丟失消息。雖然可以使用鏡像隊列（用于在 RabbitMQ 集群中復(fù)制隊列的消息，這樣做的目的是提高隊列的可用性和容錯性，以防止在單個節(jié)點故障時導(dǎo)致消息的丟失。）但也不能百分百保證消息不丟失。

雖然我們加了很多保證可靠性的機(jī)制，這樣也只是去提高消息的可靠性，不能百分百做的可靠，所以使用MQ的場景要考慮這種問題的存在，做好補(bǔ)償處理任務(wù)。

Canal+MQ同步流程

實現(xiàn)將MySQL的變更數(shù)據(jù)通過Canal寫入MQ。

配置Canal+MQ數(shù)據(jù)同步環(huán)境

參考配置鏈接

Docshttps://mx67xggunk5.feishu.cn/wiki/Yifpw51Qoim81akHF7ic5ye3npd

根據(jù)Canal+MQ同步流程，下邊進(jìn)行如下配置：

1. 配置Mysql主從同步，開啟MySQL主服務(wù)器的binlog

2. 安裝Canal并配置，保證Canal連接MySQL主服務(wù)器成功

3. 安裝RabbitMQ，并配置同步隊列。

4. 在Canal中配置RabbitMQ的連接信息，保證Canal收到binlog消息寫入MQ

索引同步?

因為涉及到好幾張表，為了降低代碼復(fù)雜度，把好幾張表中的字段都聚合到一張表中，那么canal只需要監(jiān)聽這一張表的數(shù)據(jù)就可以

上面通過配置Canal+MQ的數(shù)據(jù)同步環(huán)境實現(xiàn)了Canal從數(shù)據(jù)庫讀取binlog并且將數(shù)據(jù)寫入MQ。

下邊編寫同步程序監(jiān)聽MQ，收到消息后向ES創(chuàng)建索引。

es的版本是7.17.7，不同的版本api不一樣

?

?keyword是關(guān)鍵詞，表示不對該屬性進(jìn)行分詞，index:false表示不對其進(jìn)行索引，還有分詞方式

PUT /serve_aggregation
{"mappings" : {"properties" : {"city_code" : {"type" : "keyword"},"detail_img" : {"type" : "text","index" : false},"hot_time_stamp" : {"type" : "long"},"id" : {"type" : "keyword"},"is_hot" : {"type" : "short"},"price" : {"type" : "double"},"serve_item_icon" : {"type" : "text","index" : false},"serve_item_id" : {"type" : "keyword"},"serve_item_img" : {"type" : "text","index" : false},"serve_item_name" : {"type" : "text","analyzer": "ik_max_word","search_analyzer":"ik_smart"},"serve_item_sort_num" : {"type" : "short"},"serve_type_icon" : {"type" : "text","index" : false},"serve_type_id" : {"type" : "keyword"},"serve_type_img" : {"type" : "text","index" : false},"serve_type_name" : {"type" : "text","analyzer": "ik_max_word","search_analyzer":"ik_smart"},"serve_type_sort_num" : {"type" : "short"}}}
}

?編寫監(jiān)聽器監(jiān)聽canal的數(shù)據(jù)變化

concurrency = "1"：表示消費(fèi)線程數(shù)為1。

在同步程序中需要根據(jù)業(yè)務(wù)需求編寫同步方法，當(dāng)服務(wù)下架時會刪除索引需要重寫抽象類中的batchDelete(List<Long> ids)方法，此方法是當(dāng)刪除Serve_sync表的記錄時 對索引執(zhí)行刪除操作。

當(dāng)服務(wù)上架后需要添加索引，當(dāng)服務(wù)信息修改時需要修改索引，需要重寫抽象類中的batchSave(List<ServeSync> data)方法，此方法是當(dāng)向Serve_sync表新增或修改記錄時對索引執(zhí)行添加及修改操作。

mq監(jiān)聽到消息后解析消息

/*** 服務(wù)信息同步程序** @author itcast* @create 2023/8/15 18:14**/
@Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {@Resourceprivate ElasticSearchTemplate elasticSearchTemplate;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "canal-mq-jzo2o-foundations"),exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),key = "canal-mq-jzo2o-foundations"),concurrency = "1")public void onMessage(Message message) throws Exception {parseMsg(message);}@Overridepublic void batchSave(List<ServeSync> data) {Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException("同步失敗");}}@Overridepublic void batchDelete(List<Long> ids) {Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException("同步失敗");}}
}

?啟動jzo2o-foundations服務(wù)。

啟動成功,jzo2o-foundations服務(wù)作為MQ的消費(fèi)者和MQ建立通道，進(jìn)入canal-mq-jzo2o-foundations隊列的管理界面，查看是否建立 了監(jiān)聽通道。

?解析完數(shù)據(jù)之后判斷是否是讀多個數(shù)據(jù)，多個數(shù)據(jù)就執(zhí)行批量方法，

?然后判斷是否是保存新增或者刪除。

?

此時運(yùn)行程序，看是否能正常接收到消息。?

如何保證Canal+MQ同步消息的順序性？

場景：

如下圖：

首先明確Canal解析binlog日志信息按順序發(fā)到MQ的隊列中，現(xiàn)在是要保證消費(fèi)端如何按順序消費(fèi)隊列中的消息。

生產(chǎn)中同一個jzo2o-foundations服務(wù)會啟動多個jvm進(jìn)程，每個進(jìn)程作為canal-mq-jzo2o-foundations的消費(fèi)者,如下圖：

?

?

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "canal-mq-jzo2o-foundations",arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean") }),exchange = @Exchange(name="exchange.canal-jzo2o",type = ExchangeTypes.TOPIC),key="canal-mq-jzo2o-foundations"),concurrency="1")public void onMessage(Message message) throws Exception{parseMsg(message);}

?

管理同步表

通過測試Canal+MQ同步流程，只有當(dāng)serve_sync表變化時才會觸發(fā)同步，serve_sync表什么時候變化 ？

/*** 新增服務(wù)同步數(shù)據(jù)** @param serveId 服務(wù)id*/
private void addServeSync(Long serveId) {//服務(wù)信息Serve serve = baseMapper.selectById(serveId);//區(qū)域信息Region region = regionMapper.selectById(serve.getRegionId());//服務(wù)項信息ServeItem serveItem = serveItemMapper.selectById(serve.getServeItemId());//服務(wù)類型ServeType serveType = serveTypeMapper.selectById(serveItem.getServeTypeId());ServeSync serveSync = new ServeSync();serveSync.setServeTypeId(serveType.getId());serveSync.setServeTypeName(serveType.getName());serveSync.setServeTypeIcon(serveType.getServeTypeIcon());serveSync.setServeTypeImg(serveType.getImg());serveSync.setServeTypeSortNum(serveType.getSortNum());serveSync.setServeItemId(serveItem.getId());serveSync.setServeItemIcon(serveItem.getServeItemIcon());serveSync.setServeItemName(serveItem.getName());serveSync.setServeItemImg(serveItem.getImg());serveSync.setServeItemSortNum(serveItem.getSortNum());serveSync.setUnit(serveItem.getUnit());serveSync.setDetailImg(serveItem.getDetailImg());serveSync.setPrice(serve.getPrice());serveSync.setCityCode(region.getCityCode());serveSync.setId(serve.getId());serveSync.setIsHot(serve.getIsHot());serveSyncMapper.insert(serveSync);
}

?

省略 修改 服務(wù)類型，服務(wù)項時 更改 sync表的代碼?

搜索接口

目標(biāo)：開發(fā)搜索接口。

@RestController("consumerServeController")
@RequestMapping("/customer/serve")
@Api(tags = "用戶端 - 首頁服務(wù)查詢接口")
public class FirstPageServeController {
...@GetMapping("/search")@ApiOperation("首頁服務(wù)搜索")@ApiImplicitParams({@ApiImplicitParam(name = "cityCode", value = "城市編碼", required = true, dataTypeClass = String.class),@ApiImplicitParam(name = "serveTypeId", value = "服務(wù)類型id", dataTypeClass = Long.class),@ApiImplicitParam(name = "keyword", value = "關(guān)鍵詞", dataTypeClass = String.class)})public List<ServeSimpleResDTO> findServeList(@RequestParam("cityCode") String cityCode,@RequestParam(value = "serveTypeId", required = false) Long serveTypeId,@RequestParam(value = "keyword", required = false) String keyword) {return null;}

首先通過ES的查詢語言進(jìn)行查詢，如下：

GET /serve_aggregation/_search
{"query" : {"bool" : {"must" : [{"term" : {"city_code" : {"value" : "010"}}},{"multi_match" : {"fields" : [ "serve_item_name", "serve_type_name" ],"query" : "保潔"}}]}},"sort" : [{"serve_item_sort_num" : {"order" : "asc"}}]
}

?

?