目錄

一、數(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 實現(xiàn)數(shù)據(jù)同步

1.3.1、需求

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

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

1.3.4、啟動微服務并測試

---

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

---

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

我們知道 elasticsearch 的數(shù)據(jù)是來源于 數(shù)據(jù)庫（比如 mysql）.? 當我們在寫了代碼將 mysql 中的數(shù)據(jù)導入 es 中，那么這次導入之后 mysql 的數(shù)據(jù)并不會一成不變，將來我們的業(yè)務中會有 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：實際上不光是 es 存在數(shù)據(jù)同步問題，凡是涉及到數(shù)據(jù)庫雙寫的情況，比如 redis 和 mysql，都會存在數(shù)據(jù)同步問題.

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

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

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

1.3、解決辦法

1.3.1、同步調(diào)用

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

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

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

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

3. 更新 es.

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

缺陷：

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

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

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

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

1.3.2、異步通知（推薦）

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

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

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

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

缺陷：

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

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

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

1.3.3、監(jiān)聽 binlog

mysql 默認情況下 binlog 時關閉的，一旦開啟，那么每次 mysql 在做增刪改的時候，都會記錄相應的操作到 binlog 中.

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

優(yōu)勢：

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

劣勢：

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

2.引入新的中間件.

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

1.3.1、需求

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

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

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

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

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

public class MqConstants {//主題交換機public static final String EXCHANGE_TOPIC = "hotel.topic";//增加 or 修改酒店 隊列public static final String INSERT_QUEUE = "hotel.insert.queue";//刪除酒店 隊列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）的包必須與啟動類同級，否則聲明交換機和隊列失敗.

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

@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 來管理），否則聲明的交換機和隊列無效.

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

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

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

    @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、啟動微服務并測試

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

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