前言

如今大型的IT系統中，都會使用分布式的方式，同時會有非常多的中間件，如redis、消息隊列、大數據存儲等，但是實際核心的數據存儲依然是存儲在數據庫，作為使用最廣泛的數據庫，如何將mysql的數據與中間件的數據進行同步，既能確保數據的一致性、及時性，也能做到代碼無侵入的方式呢？如果有這樣的一個需求，數據修改后，需要及時的將mysql中的數據更新到elasticsearch,我們會怎么進行實現呢？

數據同步方案選擇

針對上文的需求，經過思考，初步有如下的一些方案：

• 代碼實現
• 針對代碼中進行數據庫的增刪改操作時，同時進行elasticsearch的增刪改操作。
• mybatis實現
• 通過mybatis plugin進行實現，截取sql語句進行分析， 針對insert、update、delete的語句進行處理。顯然，這些操作如果都是單條數據的操作，是很容易處理的。但是，實際開發中，總是會有一些批量的更新或者刪除操作，這時候，就很難進行處理了。
• Aop實現
• 不管是通過哪種Aop方式，根據制定的規則，如規范方法名，注解等進行切面處理，但依然還是會出現無法處理批量操作數據的問題。
• logstash
• logstash類似的同步組件提供的文件和數據同步的功能，可以進行數據的同步，只需要簡單的配置就能將mysql數據同步到elasticsearch，但是logstash的原理是每秒進行一次增量數據查詢，將結果同步到elasticsearch，實時性要求特別高的，可能無法滿足要求。且此方案的性能不是很好，造成資源的浪費。

代碼實現：技術難度低，侵入性強，實時性高。

基于mybatis：有一定的技術難度，但是無法覆蓋所有的場景

Aop實現：技術難度低，半侵入性，需要規范代碼，依然無法覆蓋所有的場景

logstash：技術難度低，無侵入性，無需開發，但會造成資源浪費。

那么是否有什么更好的方式進行處理嗎？mysql binlog同步，實時性強，對于應用無任何侵入性，且性能更好，不會造成資源浪費，那么就有了我今天的主角——canal

canal

介紹

canal 是阿里巴巴的一個開源項目，基于java實現，整體已經在很多大型的互聯網項目生產環境中使用，包括阿里、美團等都有廣泛的應用，是一個非常成熟的數據庫同步方案，基礎的使用只需要進行簡單的配置即可。

canal是通過模擬成為mysql 的slave的方式，監聽mysql 的binlog日志來獲取數據，binlog設置為row模式以后，不僅能獲取到執行的每一個增刪改的腳本，同時還能獲取到修改前和修改后的數據，基于這個特性，canal就能高性能的獲取到mysql數據數據的變更。

使用

canal的介紹在官網有非常詳細的說明，如果想了解更多，大家可以移步官網(https://github.com/alibaba/canal)了解。我這里補充下使用中不太容易理解部分。

canal的部署主要分為server端和client端。

server端部署好以后，可以直接監聽mysql binlog,因為server端是把自己模擬成了mysql slave，所以，只能接受數據，沒有進行任何邏輯的處理，具體的邏輯處理，需要client端進行處理。

client端一般是需要大家進行簡單的開發。https://github.com/alibaba/canal/wiki/ClientAPI 有一個簡單的示例，很容易理解。

canal Adapter

為了便于大家的使用，官方做了一個獨立的組件Adapter，Adapter是可以將canal server端獲取的數據轉換成幾個常用的中間件數據源，現在支持kafka、rocketmq、hbase、elasticsearch，針對這幾個中間件的支持，直接配置即可，無需開發。上文中，如果需要將mysql的數據同步到elasticsearch，直接運行 canal Adapter，修改相關的配置即可。

常見問題

• 無法接收到數據，程序也沒有報錯？
• 一定要確保mysql的binlog模式為row模式，canal原理是解析Binlog文件，并且直接中文件中獲取數據的。
• Adapter 使用無法同步數據？
• 按照官方文檔，檢查配置項，如sql的大小寫，字段的大小寫可能都會有影響，如果還無法搞定，可以自己獲取代碼調試下，Adapter的代碼還是比較容易看懂的。

canal Adapter elasticsearch 改造

因為有了canal和canal Adapter這個神器，同步到elasticsearch、hbase等問題都解決了，但是自己的開發的過程中發現，Adapter使用還是有些問題，因為先使用的是elasticsearch同步功能，所以對elasticsearch進行了一些改造：

elasticsearch初始化

一個全新的elasticsearch無法使用，因為沒有創建elasticsearch index和mapping,增加了對應的功能。 elasticsearch配置文件mapping節點增加兩個參數：

enablefieldmap: true fieldmap: id: "text" BuildingId: "text" HouseNum: "text" Floors: "text" IdProjectInfo: "text" HouseDigitNum: "text" BuildingNum: "text" BuildingName: "text" Name: "text" projectid: "text" bIdProjectInfo: "text" cinitid: "text" pCommunityId: "text"

enablefieldmap 是否需要自動生成fieldmap，默認為false,如果需要啟動的時候就生成這設置為true,并且設置 fieldmap,類似elasticsearch mapping中每個字段的類型。

esconfig bug處理

代碼中獲取binlog的日志處理時，必須要獲取數據庫名，但是當獲取binlog為type query時，是無法獲取 數據庫名的，此處有bug，導致出現 "Outer adapter write failed" ,且未輸出錯誤日志，修復此bug.

后續計劃

• 增加rabbit MQ的支持
• 增加redis的支持

源碼

源碼地址：https://github.com/itmifen/canal

總結

以上是生活随笔為你收集整理的mysql获取最好成绩对应数据的其他项_开源数据同步神器——canal的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。