目錄

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

1.1、什么是數(shù)據(jù)同步

1.2、解決數(shù)據(jù)同步面臨的問題

1.3、解決辦法

1.3.1、同步調(diào)用

1.3.2、異步通知（推薦）

1.3.3、監(jiān)聽 binlog

1.3、基于 RabbitMQ 實(shí)現(xiàn)數(shù)據(jù)同步

1.3.1、需求

1.3.2、在“酒店搜索服務(wù)”中 聲明 exchange、queue、routingKey，同時開啟監(jiān)聽

1.3.3、在“酒店管理服務(wù)”中發(fā)布消息

1.3.4、啟動微服務(wù)并測試

---

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

---

1.1、什么是數(shù)據(jù)同步

我們知道 elasticsearch 的數(shù)據(jù)是來源于 數(shù)據(jù)庫（比如 mysql）.? 當(dāng)我們在寫了代碼將 mysql 中的數(shù)據(jù)導(dǎo)入 es 中，那么這次導(dǎo)入之后 mysql 的數(shù)據(jù)并不會一成不變，將來我們的業(yè)務(wù)中會有 crud，數(shù)據(jù)庫中的數(shù)據(jù)就和有新增、修改、刪除，那么 mysql 數(shù)據(jù)一定那發(fā)生變化， es 如果不跟著變化，就會出現(xiàn)問題.

比如在一個商城系統(tǒng)中，到了 雙11 ，數(shù)據(jù)庫中的商品的價格下降，而 es 還是老價格，那么用戶搜索時看到的商品的價格還是沒有變化，用戶可能就得考慮換軟件了~??

因此，我們要保證 mysql 數(shù)據(jù)變化的時候 es 也能跟著同步變化，這就是數(shù)據(jù)同步.

Ps：實(shí)際上不光是 es 存在數(shù)據(jù)同步問題，凡是涉及到數(shù)據(jù)庫雙寫的情況，比如 redis 和 mysql，都會存在數(shù)據(jù)同步問題.

1.2、解決數(shù)據(jù)同步面臨的問題

如果你現(xiàn)在是一個單體式的項(xiàng)目，所有業(yè)務(wù)都寫在一個項(xiàng)目中，那就比較好辦，無非就是在及逆行新增、修改、刪除業(yè)務(wù)的時候，同時把 es 也一起更新就? ok.

但是如果我們是一個 微服務(wù) 架構(gòu)的項(xiàng)目上，不同的業(yè)務(wù)往往會在不同的微服務(wù)上，比如 “商品數(shù)據(jù)管理業(yè)務(wù)” 和 “商品數(shù)據(jù)搜索業(yè)務(wù)” 肯定會在兩個不同的微服務(wù)上，那么跨微服務(wù)的項(xiàng)目就沒辦法直接操作了.

1.3、解決辦法

1.3.1、同步調(diào)用

假設(shè)我們現(xiàn)在有兩個微服務(wù)，一個是 酒店數(shù)據(jù)管理服務(wù)，另一個是酒店數(shù)據(jù)搜索服務(wù). 假設(shè)這兩個服務(wù)之間互相不能訪問對方的數(shù)據(jù)庫，也就是說，酒店管理服務(wù)只能訪問 mysql，而 酒店搜索服務(wù) 只能訪問 es，這也符合 微服務(wù) 里的標(biāo)準(zhǔn)和規(guī)范.

這里有一種辦法是同步調(diào)用，步驟如下：

1.比如用戶做新增操作時，首先把數(shù)據(jù)寫到數(shù)據(jù)庫里.

2. 數(shù)據(jù)庫寫完了以后緊接著調(diào)用 酒店搜索服務(wù)中的 “更新索引庫” 的接口.

3. 更新 es.

最后更新完了 es 就把響應(yīng)反饋給搜索服務(wù)，然后搜索服務(wù)才會把把響應(yīng)反饋給? 管理服務(wù)，然后再反饋給用戶.? 這整個過程依次執(zhí)行，因此也叫同步調(diào)用.

缺陷：

1. 數(shù)據(jù)耦合，業(yè)務(wù)耦合：原本只是寫數(shù)據(jù)庫，寫完就結(jié)束了，然后現(xiàn)在還需要再寫完數(shù)據(jù)庫的代碼后面再加上 調(diào)用 “更新索引庫” 接口的代碼，而且這調(diào)用 這個接口的業(yè)務(wù)跟我新增業(yè)務(wù)顯然沒有關(guān)系啊，現(xiàn)在業(yè)務(wù)耦合在一起，將來必然也會影響性能.

2. 影響性能（耦合帶來的問題）：原本數(shù)據(jù)庫寫完了，比如耗時 50ms，但是現(xiàn)在寫完數(shù)據(jù)庫，你還得等待后臺調(diào)用這個接口返回的響應(yīng)，而這個接口又要等待 es 這里的響應(yīng)，假如這里也耗時 50ms，那么總的耗時不就是這個三個步驟相加的，達(dá)到 100 ms.

3. 牽一發(fā)而動全身（耦合帶來的問題）：如果 步驟 2 和 步驟 3 任意一個位置出現(xiàn)了異常，就會導(dǎo)致整個業(yè)務(wù)也出現(xiàn)崩潰.

同步調(diào)用這么多問題，那么就需要考慮別的方案了.

1.3.2、異步通知（推薦）

這里就需要使用到 mq 來實(shí)現(xiàn)了，步驟如下：

1. 當(dāng)有人做新增操作時，先去寫數(shù)據(jù)庫.

2. 寫完之后，不調(diào)用任何服務(wù)的接口，而是向 mq 發(fā)送一個消息，通知一下 其他服務(wù)：“我這里數(shù)據(jù)新增了啊~”，整個步驟到這里結(jié)束.

那么至于誰來監(jiān)聽這個消息，監(jiān)聽了以后做什么，跟我有關(guān)系嗎？沒關(guān)系，這樣一來，業(yè)務(wù)的耦合就解除了；? 至于其他服務(wù)耗時多少秒，跟我也沒關(guān)系，我寫完數(shù)據(jù)庫，發(fā)完消息就結(jié)束了，因此性能也提升了；再者，就算其他服務(wù)出現(xiàn)異常了，跟我這里也沒關(guān)系.

缺陷：

1. 這樣一來，比較依賴 mq 消息的可靠性.

2. 引入新的中間件，實(shí)現(xiàn)的復(fù)雜度也會有一定的上升.

不過這些缺陷，跟同步調(diào)用比起來，算不了什么，因此這也是比較推薦的方案.

1.3.3、監(jiān)聽 binlog

mysql 默認(rèn)情況下 binlog 時關(guān)閉的，一旦開啟，那么每次 mysql 在做增刪改的時候，都會記錄相應(yīng)的操作到 binlog 中.

那么可以使用類似于 canal 這樣的中間件來監(jiān)聽 binlog，一旦發(fā)現(xiàn)變化，立馬通知對應(yīng)的微服務(wù)，這個時候就知道數(shù)據(jù)發(fā)生變更，就可以進(jìn)行更新了.

優(yōu)勢：

這種方案他既不給任何中間件發(fā)消息，也不去調(diào)用任何接口，因此耦合度是最低的.

劣勢：

1.開啟 binlog，對 mysql 的壓力就增加了.

2.引入新的中間件.

1.3、基于 RabbitMQ 實(shí)現(xiàn)數(shù)據(jù)同步

1.3.1、需求

現(xiàn)在有兩個微服務(wù)：“酒店管理服務(wù)” 和 "酒店搜索服務(wù)"

用戶操作?“酒店管理服務(wù)” 進(jìn)行增刪改數(shù)據(jù)、要求對 "酒店搜索服務(wù)" 中的 es 的數(shù)據(jù)也要完成相同的數(shù)據(jù)更改操作.

這里使用 MQ 的異步方式實(shí)現(xiàn)數(shù)據(jù)同步.

1.3.2、在“酒店搜索服務(wù)”中 聲明 exchange、queue、routingKey，同時開啟監(jiān)聽

在 “酒店搜索服務(wù)” 中去聲明一個?exchange，用來接收 增刪改 的消息，接著聲明兩個隊(duì)列即可，一個隊(duì)列用來增改（這兩用一個隊(duì)列是因?yàn)樵?es 中，增改可以使用同一個 DSL 語句實(shí)現(xiàn)），另一個用來刪除（這里我是以 Bean 的方式注入到容器中了）.

public class MqConstants {//主題交換機(jī)public static final String EXCHANGE_TOPIC = "hotel.topic";//增加 or 修改酒店 隊(duì)列public static final String INSERT_QUEUE = "hotel.insert.queue";//刪除酒店 隊(duì)列public static final String DELETE_QUEUE = "hotel.delete.queue";//增加 or 修改酒店 routingKeypublic static final String INSERT_KEY = "hotel.insert.key";//刪除酒店 routingKeypublic static final String DELETE_KEY = "hotel.delete.key";}

@Configuration
public class MqConfig {@Beanpublic TopicExchange hotelTopicExchange() {return new TopicExchange(MqConstants.EXCHANGE_TOPIC, true, false);}@Beanpublic Queue hotelInsertQueue() {return new Queue(MqConstants.INSERT_QUEUE, true);}@Beanpublic Queue hotelDeleteQueue() {return new Queue(MqConstants.DELETE_QUEUE, true);}@Beanpublic Binding hotelInsertBinding() {return BindingBuilder.bind(hotelInsertQueue()).to(hotelTopicExchange()).with(MqConstants.INSERT_KEY);}@Beanpublic Binding hotelDeleteBinding() {return BindingBuilder.bind(hotelDeleteQueue()).to(hotelTopicExchange()).with(MqConstants.DELETE_KEY);}}

Ps：此類（MqConfig）的包必須與啟動類同級，否則聲明交換機(jī)和隊(duì)列失敗.

最后就可以使用 @RabbitListener 監(jiān)聽 隊(duì)列 了.

@Component
public class MqListener {@Autowiredprivate IHotelService hotelService;@RabbitListener(queues = MqConstants.INSERT_QUEUE)public void HotelInsertOrUpdateListener(Long id) {hotelService.insertHotelById(id);}@RabbitListener(queues = MqConstants.DELETE_QUEUE)public void HotelDeleteListener(Long id) {hotelService.deleteHotelById(id);}}

Ps：此類（MyListener，監(jiān)聽者類）上必須要有 @Component 注解（交由給 Spring 來管理），否則聲明的交換機(jī)和隊(duì)列無效.

1.3.3、在“酒店管理服務(wù)”中發(fā)布消息

酒店管理服務(wù)中，一旦用戶進(jìn)行酒店的 增刪改，就會對數(shù)據(jù)庫信息進(jìn)行修改，然后將增刪改的消息發(fā)布到 MQ 中.

Ps：這里不要發(fā)送 hotel 整體數(shù)據(jù)，太大可能會導(dǎo)致占滿隊(duì)列(Mq 是基于內(nèi)存存儲的，因此會設(shè)定隊(duì)列上限)，因此這里發(fā)送 id 即可.? es 這邊拿到 id，就可進(jìn)行相應(yīng)的增刪改.

    @PostMappingpublic void saveHotel(@RequestBody Hotel hotel){// 新增酒店hotelService.save(hotel);rabbitTemplate.convertAndSend(MqConstants.EXCHANGE_TOPIC, MqConstants.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.EXCHANGE_TOPIC, MqConstants.INSERT_KEY, hotel.getId());}@DeleteMapping("/{id}")public void deleteById(@PathVariable("id") Long id) {//刪除酒店信息hotelService.removeById(id);rabbitTemplate.convertAndSend(MqConstants.EXCHANGE_TOPIC, MqConstants.DELETE_KEY, id);}

1.3.4、啟動微服務(wù)并測試

a）在酒店管理頁面中，修改 “7天連鎖酒店（上海莘莊地鐵站店）” 為 “7天連鎖酒店（此處正在施工，請謹(jǐn)慎前往）”，如下.

b）在酒店搜索頁面中，搜索 “施工” 關(guān)鍵詞，就可以看到在 “酒店管理服務(wù)” 中更新的信息，已經(jīng)同步到了 “酒店搜索服務(wù)” 中.