公開課-canal數(shù)據(jù)同步終極解決方案資料文檔canal-實現(xiàn)數(shù)據(jù)同步_h

1、 數(shù)據(jù)同步解決方案Canal傳智播客北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090課程目標目標一:數(shù)據(jù)同步方案分析目標二:Canal 的介紹及工作原理目標三:Canal 的部署及配置目標四:使用 Canal 完成 solr 的數(shù)據(jù)同步1需求在現(xiàn)代的系統(tǒng)開發(fā)中, 為了提高搜索效率 , 以及搜索的精準度, 會大量的使用 redis ,memcache 等 nosql 系統(tǒng)的數(shù)據(jù)庫 , 以及 solr , elasticsearch 類似的全文檢索服如何將實時務; 那么這個時候, 就又有一個問題需要我們來考慮, 就是數(shù)據(jù)同步的問題,變化的數(shù)據(jù)庫中的數(shù)據(jù)同步到 solr 的索引庫

2、中或者 redis 中呢?2數(shù)據(jù)同步方案2.1方案一: 業(yè)務代碼中同步在增加、修改、刪除之后，執(zhí)行操作 solr 索引庫的邏輯代碼。 北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090優(yōu)點 : 操作簡便缺點 :1)2)業(yè)務耦合度高執(zhí)行效率變低2.2方案二: 定時任務同步在執(zhí)行完增加、修改、刪除，操作數(shù)據(jù)庫中的數(shù)據(jù)變更之后 ，通過定時任務定時的將數(shù)據(jù)庫的數(shù)據(jù)同步到 solr 的索引庫中。 定時任務技術 ： SpringTask ， Quartz優(yōu)點：同步solr 索引庫操作與業(yè)務代碼完全解耦。缺點：數(shù)據(jù)的實時性并不高。 2.3方案三: 通過 MQ 實現(xiàn)同步在執(zhí)行完增加、修改、刪

3、除之后， 往 MQ 中發(fā)送一條消息隊列中獲取消息，然后執(zhí)行同步 solr 索引庫的邏輯。 ；同步程序作為 MQ 中的消費者，從消息 北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090優(yōu)點：業(yè)務代碼解耦， 并且可以做到準實時 缺點：需要在業(yè)務代碼中加入發(fā)送消息到 MQ 中的代碼 ， API 耦合 2.4方案四: 通過 Canal 實現(xiàn)實時同步通過 Canal 來解析數(shù)據(jù)庫的日志信息， 來檢測數(shù)據(jù)庫中表結(jié)構(gòu)的數(shù)據(jù)變化，從而更新 solr 索引庫。優(yōu)點：業(yè)務代碼完全解耦，API 完全解耦，可以做到準實時。 缺點：無 3Canal 介紹3.1Canal 概述阿里巴巴 mysql 數(shù)據(jù)

4、庫 binlog 的增量訂閱 & 消費組件。名稱：canal knl譯意：水道 / 管道 / 溝渠語言：純 java 開發(fā)定位：基于數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱 & 消費，目前主要支持了mysql關鍵詞： mysql binlog parser / real-time / queue&topic北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-90903.2Canal 下載官網(wǎng): /alibaba/canal這里我們選擇了 Canal 的 1.0.24 版本.canal.deployer-1.0.24.tar.gz : 這個是 canalSe

5、rver 的部署包canal.example-1.0.24.tar.gz : 這個是樣例Source code(zip) : 是 canal 的源碼包4Canal 工作原理4.1mysql 主從同步實現(xiàn)原理:北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090從上層來看，主從復制分成三步：1. master 將改變記錄到二進制日志 (binary log) 中（這些記錄叫做二進制日志，binary log events ，可以通過 show binlog events 進行查看）；2. slave 將 master 的 binary log events 拷貝到它的中繼日志(r

6、elaylog);3. slave 重做中繼日志中的將改變反映它自己的數(shù)據(jù)。4.2Canal 內(nèi)部原理原理圖:原理相對比較簡單：北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-90901. canal 模擬 mysql slave 的交互協(xié)議，自己為 mysqlslave ，向 mysqlmaster 發(fā)送 dump 協(xié)議。log給slave(也就是 2. mysql master 收到 dump 請求， 開始推送 binarycanal) 。3. canal 解析 binary log 對象 ( 原始為 byte 流 ) 。4.3Canal 內(nèi)部結(jié)構(gòu)說明：1)2)Server :

7、 代表一個 canal 運行實例，對應于一個 jvmtance : 對應于一個數(shù)據(jù)隊列 （1 個 server 對應 1.n 個 tance)tance 下的子模塊：eventParser : (數(shù)據(jù)源接入，模擬 slave 協(xié)議和 master 進行交互，協(xié)議解析)eventSink: (Parser 和 Store 鏈接器，進行數(shù)據(jù)過濾，加工，分發(fā)的工作)北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090eventStore: (數(shù)據(jù)存儲)metaManager : (增量訂閱&消費信息管理器)5Canal 環(huán)境準備5.1Mysql 數(shù)據(jù)庫root 遠程訪問grantfl

8、ushall privileges on *.* to root % identified by 2143;privileges;5.2Mysql 配置mysql的binlogcanal 的原理是基于 mysqlbinlog 技術，所以這里一定需要開啟寫入功能，建議配置 binlog模式為 row。查看方式：SHOW VARIABLES LIKEbinlog_format;修改配置：修改以下配置項：mysqldlog-bin=mysql-bin#添加這一行就 okbinlog_format=ROW#選擇 row 模式server_id=1#配置 mysql replaction 需要定義，不能

9、與 canal 的 slaveId 重復 北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090注:修改完成之后,需要重啟Mysql 服務北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090知識小貼士 :1) Row日志中會記錄成每一行數(shù)據(jù)被修改的形式，然后在 slave 端再對相同的數(shù)據(jù)進行修改。 優(yōu)點：在 row 模式下，bin-log 中可以不記錄執(zhí)行的 SQL 語句的上下文相關的信息，僅僅只需要記錄那一條記錄被修改了，修改成什么樣了。所以 row 的日志內(nèi)容會非常清楚的記錄下每一行數(shù)據(jù)修改的細節(jié)，非常容易理解。而且不會出現(xiàn)某些特定情況下的存儲過程或 fun

10、ction ，以及 trigger 的調(diào)用和觸發(fā)無法被正確復制的問題。 2) Statement每一條會修改數(shù)據(jù)的 SQL 都會記錄到 master 的 bin-log 中。slave 在復制的時候 SQL 進程會解析成和原來 master 端執(zhí)行過的相同的 SQL 再次執(zhí)行。 優(yōu)點：在 statement 模式下，首先就是解決了 row 模式的缺點，不需要記錄每一行數(shù)據(jù)的變化， 減少了 bin-log 日志量，節(jié)省 I/O 以及存儲資源，提高性能。因為他只需要記錄在 master 上所執(zhí)行的語句的細節(jié)，以及執(zhí)行語句時候的上下文的信息。 缺點：在 statement 模式下，由于他是記錄的執(zhí)行

11、語句，所以，為了讓這些語句在 slave 端也能正確執(zhí)行，那么他還必須記錄每條語句在執(zhí)行的時候的一些相關信息，也就是上下文信息，以保證所有語句在 slave 端杯執(zhí)行的時候能夠得到和在 master 端執(zhí)行時候相同的結(jié)果。另外就是，由于 MySQL 現(xiàn)在發(fā)展比較快，很多的新功能不斷的加入，使 MySQL 的復制遇到了不小的挑戰(zhàn)，自然復制的時候涉及到越復雜的內(nèi)容，bug 也就越容易出現(xiàn)。在 statement 中，目前已經(jīng)發(fā)現(xiàn)的就有不少情況會造成 MySQL的復制出現(xiàn)問題，主要是修改數(shù)據(jù)的時候使用了某些特定的函數(shù)或者功能的時候會出現(xiàn)，比如：sleep()函數(shù)在有些版本中就不能被正確復制，在存儲過

12、程中使用了 last_ ert_id() 函數(shù)，可能會使 5.3Mysql 創(chuàng)建用戶授權canal 的原理是模擬自己為 mysql slave ，所以這里一定需要做為 mysql slave的 相關權限。 創(chuàng)建一個主從同步的賬戶，并且賦予權限：CREATE USER canallocalhost IDENTIFIED BY canal;GRANT SELECT, REPLICATION SLAVE, REPLICATIONCLIENTON*.*TOcanallocalhost;FLUSH PRIVILEGES;6Canal 部署安裝6.1上傳解壓解壓后的目錄如下：目錄介紹： bin: 存儲的是

13、可執(zhí)行腳本conf ：存放 canal 的配置文件lib：存放 canal 的 lib 目錄logs ：存放的是日志文件 北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090slave 和 master 上得到不一致的 id 等等。由于 row 是基于每一行來記錄的變化，所以不會出現(xiàn)類 似的問題。 6.2配置編輯 canal/conf/example/perties :選項含義:1)canal.tance.mysql.slaveId :mysql 集群配置中的 serverId 概念，需要保證和當前 mysql 集群中 id 唯一;2)3)4)5)6)7)ca

14、nal. canal. canal. canal. canal.canal.tance.master.address:mysql 主 庫 鏈 接 地 址 ;tance.dbUsername : mysql 數(shù) 據(jù) 庫 帳 號 ;tance.dbPassword : mysql 數(shù) 據(jù) 庫 ;tance.defaultDatabaseNametance.connectionCharset : mysql 鏈 接 時 默 認 數(shù) 據(jù) 庫 ; mysql 數(shù) 據(jù) 解 析 編 碼 ;tance.filter.regex : mysql 數(shù)據(jù)解析關注的表，Perl 正則表達式.6.3啟動/停止北京市昌平

15、區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090# mysql serverIdcanal.tance.mysql.slaveId = 1234#position info，需要改成自己的數(shù)據(jù)庫信息canal.tance.master.address = :3306 =canal.tance.master.position = canal.tance.master.timestamp =#canal.tance.standby.address = #canal.tance.standby.journal

16、.name = #canal.tance.standby.position = #canal.tance.standby.timestamp =#username/password，需要改成自己的數(shù)據(jù)庫信息canal.tance.dbUsername = canal canal.tance.dbPassword = canalcanal.tance.defaultDatabaseName =canaldb canal.tance.connectionCharset = UTF-8#table regexcanal.tance.filter.regex = canaldb.* #1) start

17、up.sh : 啟動腳本2) stop.sh : 停止腳本 7數(shù)據(jù)拉取測試7.1官方源碼導入在源碼目錄中,有一個工程 example, 這個工程中存放的就是一些樣例工程.7.2測試類修改可以通過其中的一個 SimpleCanalClientTest 類進行測試.需要修改 CanalServer 的 IP 地址, 及端.北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-90907.3數(shù)據(jù)變更測試7.3.1 創(chuàng)建表創(chuàng)建 tb_book 表:CREATE TABLE tb_book (id INT(11) NOT NULL AUTO_INCREMENT COMMENT 主鍵,name VA

18、RCHAR(100) NOT NULL COMMENT 書名,author VARCHAR(100) DEFAULT NULL COMMENT 作者,publishtime DATETIME DEFAULT NULL COMMENT 發(fā)行日期,price DOUBLE(10,2) DEFAULT NULL COMMENT 價格,publishgroup VARCHAR(100) DEFAULT NULL COMMENT 發(fā)版社, PRIMARY KEY (id) ENGINE=INNODB DEFAULT CHARSET=utf8mb47.3.2 插入數(shù)據(jù)執(zhí)行 SQL :ERT INTO tb

19、_book(NAME , author , publishtime VALUES(白帽子講安全協(xié)議,吳瀚請,NOW(),99.00,ERT INTO tb_book(NAME , author , publishtime VALUES(白帽子講安全協(xié)議 2,吳瀚請,NOW(),99.00,price , );price ,);publishgroup)publishgroup),Canal 數(shù)據(jù)監(jiān)測結(jié)果 :7.3.3 更新數(shù)據(jù)執(zhí)行 SQL 語句:UPDATE tb_bookSET NAME = 白帽子講安全協(xié)議第二版WHERE id =2;Canal 數(shù)據(jù)監(jiān)測結(jié)果:北京市昌平區(qū)建材城西路金燕龍

20、辦公樓一層電話：400-618-90907.3.4 刪除數(shù)據(jù)執(zhí) 行 SQL :DELETE FROM tb_book WHEREid=1;Canal 數(shù)據(jù)監(jiān)測結(jié)果:8數(shù)據(jù)同步實現(xiàn)8.1需求描述將數(shù)據(jù)庫數(shù)據(jù)的變化, 通過 canal 解析 binlog 日志,實時更新到 solr 的索引庫中.8.2Solr 環(huán)境的搭建略8.3同步程序8.3.1 引入依賴北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090com.alibaba.otter8.3.2 定 義 POJO北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090public class Book privat

21、e Integer id; private String name; private String author; private Date publishtime; private Double price;private String publishgroup; public Integer getId() return id;public void setId(Integer id) tocol1.0.24commons-langcommons-lang2.6org.codehaus.jacksonj

22、ackson-mapper-asl1.8.9org.apache.solrsolr-solrj4.10.3junitjunit4.9test8.3.3 定義 solr 的域與 pojo 屬性的映射關系北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090Fieldprivate Integer id;Field(book_name)private String name;Field(book_author)this.id = id;public String getName() return name;public void setName(String name) this.n

23、ame = name;public String getAuthor() return author;public void setAuthor(String author) this.author = author;public Date getPublishtime() return publishtime;public void setPublishtime(Date publishtime) this.publishtime = publishtime;public Double getPrice() return price;public void setPrice(Double p

24、rice) this.price = price;public String getPublishgroup() return publishgroup;public void setPublishgroup(String publishgroup) this.publishgroup = publishgroup;8.3.4 同步程序編寫北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090public class CanalPullData private static Logger logger = LoggerFactory.getLogger(CanalPullData

25、.class);public static void matring args) throws Exception String destination = example; String hostname = 52; Integer port = 11111;CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname, port), destination, , );connector.connect(); connector.subscri

26、be();(Canal Server + hostname + : + port + 連接成功);Integer batchSize = 5*1024;while (true)Message message = connector.getWithoutAck(batchSize);long messageId = message.getId();int size = message.getEntries().size();if(messageId = -1 | size = 0)try TimeUnit.SECONDS.sleep(1); catch (Interrupt

27、edException e) e.printStackTrace();connector.ack(messageId);elseprivate String author;Field(book_publishtime)private Date publishtime;Field(book_price)private Double price;Field(book_publishgroup)private String publishgroup;(binlog 分析開始);List entryList = CanalDataParser.convertToInnerBinl

28、ogEntry(message);syncDataToSolr(entryList);(檢測到修改的 Entry 數(shù)量size=+entryList.size()+);private static void syncDataToSolr(List entryList) throws Exception SolrServer solrServer = new HttpSolrServer(52:8080/solr);if(entryList != null)for (InnerBinlogEntry innerBinlogEntry :

29、 entryList) Book book = new Book();if(innerBinlogEntry.getEventType() = CanalEntry.EventType.innerBinlogEntry.getEventType() = CanalEntry.EventType.UPDATE)ERT |ListMap mapList = innerBinlogEntry.getRows();if(mapList != null)for (Map valueMap : mapList) BinlogValue idValue = valueMap.get(id); BinlogV

30、alue nameValue = valueMap.get(name);BinlogValue authorValue = valueMap.get(author);BinlogValue publishtimeValue = valueMap.get(publishtime); BinlogValue priceValue = valueMap.get(price);BinlogValue publishgroupValue = valueMap.get(publishgroup);book.setId(Integer.parseInt(idValue.getValue(); book.se

31、tName(nameValue.getValue(); book.setAuthor(authorValue.getValue(); book.setPublishtime(DateUtils.parseDate(publishtimeValue.getValue(); book.setPublishgroup(publishgroupValue.getValue();book.setPrice(Double.parseDouble(priceValue.getValue();/添加/更新數(shù)據(jù)到 solr 索引庫 (- 添加/更新 solr 索引庫 : + book);

32、solrServer.addBean(book);solrSmit();else if(innerBinlogEntry.getEventType() =CanalEntry.EventType.DELETE) ListMap rows = innerBinlogEntry.getRows(); if(rows != null)for (Map row : rows) 北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-90908.3.5 工具類CanalDataParser:用來轉(zhuǎn)換解析從 CanalServer 中獲取的 Message 對象.北京市昌平區(qū)建材城西

33、路金燕龍辦公樓一層電話：400-618-9090public class CanalDataParser protected static final String DATE_FORMAT = yyyy-MM-dd HH:mm:ss; protected static final String yyyyMMddHHmmss = yyyyMMddHHmmss; protected static final String yyyyMMdd= yyyyMMdd;protected static final String SEP= SystemUtils.LINE_SEPARATOR;protecte

34、d static String context_format= null; protected static String row_format= null; protected static String transaction_format = null; protected static String row_log = null;private static Logger logger = LoggerFactory.getLogger(CanalDataParser.class);static context_format = SEP + * + SEP; context_forma

35、t += * Batch Id: ,count : , memsize : , Time : + SEP; context_format += * Start : + SEP;context_format += * End : + SEP;context_format += * + SEP;row_format = SEP+ binlog:,name, eventType: , executeTime : , delay : ms+ SEP;transaction_format = SEP + = binlog: , executeTime : , delay : ms + SEP;row_l

36、og = schema, table;BinlogValue idValue = row.get(id); String id = idValue.getBeforeValue();/根據(jù) ID 刪除 solr 索引庫的數(shù)據(jù) (- 刪除 solr 索引庫 : + id); solrServer.deleteById(id);solrSmit();public static List convertToInnerBinlogEntry(Message message) List innerBinlogEntryList = newArrayList();l

37、(接收到空的 message; 忽略);return innerBinlogEntryList;long batchId = message.getId();int size = message.getEntries().size();if (batchId = -1 | size = 0) (接收到空的 messagesize= + size + ; 忽略);return innerBinlogEntryList;printLog(message, batchId, size);List entrys = message.getEntries();/

38、輸出日志 for (Entry entry : entrys) long executeTime = entry.getHeader().getExecuteTime();long delayTime = new Date().getTime() - executeTime;if (entry.getEntryType() = EntryType.TRANSACTIONBEGIN | entry.getEntryType() = EntryType.TRANSACTIONEND) if (entry.getEntryType() = EntryType.TRANSACTIONBEGIN) Tr

39、ansactionBegin begin = null; try begin = TransactionBegin.parseFrom(entry.getStoreValue(); catch (InvalidProtocolBufferException e) throw new RuntimeException(parse event has an error , data: + entry.toString(), e);/ 打印事務頭信息，執(zhí)行的線程 id，事務耗時 (BEGIN - Thread id: , begin.getThreadId();logger.i

40、nfo(transaction_format, new Object entry.getHeader().getLogfileName(),String.valueOf(entry.getHeader().getLogfileOffset(),String.valueOf(entry.getHeader().getExecuteTime(), String.valueOf(delayTime) ); else if (entry.getEntryType() = EntryType.TRANSACTIONEND) TransactionEnd end = null;try end = Tran

41、sactionEnd.parseFrom(entry.getStoreValue(); catch (InvalidProtocolBufferException e) throw new RuntimeException(parse event has an error , data: + entry.toString(), e);/ 打印事務提交信息，事務 id北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090(END - transaction id: , end.getTransactionId();(transaction

42、_format,new Object entry.getHeader().getLogfileName(),String.valueOf(entry.getHeader().getLogfileOffset(),String.valueOf(entry.getHeader().getExecuteTime(), String.valueOf(delayTime) );continue;/解析結(jié)果 if (entry.getEntryType() = EntryType.ROWDATA) RowChange rowChage = null;try rowChage = RowChange.par

43、seFrom(entry.getStoreValue(); catch (Exception e) throw new RuntimeException(parse event has an error , data: + entry.toString(), e);EventType eventType = rowChage.getEventType();(row_format, new Object entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset(

44、),entry.getHeader().getSchemaName(),entry.getHeader().getTableName(), eventType, String.valueOf(entry.getHeader().getExecuteTime(), String.valueOf(delayTime) );/組裝數(shù)據(jù)結(jié)果 if (eventType = EventType.ERT | eventType = EventType.DELETE | eventType = EventType.UPDATE)String schemaName = entry.getHeader().ge

45、tSchemaName();String tableName = entry.getHeader().getTableName();ListMap rows = parseEntry(entry);InnerBinlogEntry innerBinlogEntry = new InnerBinlogEntry(); innerBinlogEntry.setEntry(entry);innerBinlogEntry.setEventType(eventType);innerBinlogEntry.setSchemaName(schemaName); innerBinlogEntry.setTab

46、leName(tableName.toLowerCase();innerBinlogEntry.setRows(rows);innerBinlogEntryList.add(innerBinlogEntry); else ( 存 在ERT UPDATE 操作之外的 SQL + eventType.toString() + );ERTcontinue;北京市昌平區(qū)建材城西路金燕龍辦公樓一層電話：400-618-9090return innerBinlogEntryList;private static ListMap parseEntry(Entry entry) List

47、Map rows = new ArrayListMap(); try String schemaName = entry.getHeader().getSchemaName();String tableName = entry.getHeader().getTableName();RowChange rowChage = RowChange.parseFrom(entry.getStoreValue();EventType eventType = rowChage.getEventType();/ 處理每個 Entry 中的每行數(shù)據(jù) for (RowData rowData : rowChag

48、e.getRowDatasList() StringBuilder rowlog = new StringBuilder(rowlog schema + schemaName + , table + tableName +, event + eventType.toString() + );Map row = new HashMap(); List beforeColumns = rowData.getBeforeColumnsList(); List afterColumns = rowData.getAfterColumnsList(); beforeColumns = rowData.g

49、etBeforeColumnsList();if (eventType = EventType.DELETE) /deletefor(Column column : beforeColumns) BinlogValue binlogValue = new BinlogValue(); binlogValue.setValue(column.getValue(); binlogValue.setBeforeValue(column.getValue(); row.put(column.getName(), binlogValue); else if(eventType = EventType.U

50、PDATE) /updatefor(Column column : beforeColumns) BinlogValue binlogValue = new BinlogValue(); binlogValue.setBeforeValue(column.getValue();row.put(column.getName(), binlogValue);for(Column column : afterColumns) BinlogValue binlogValue = row.get(column.getName();if(binlogValue = null) binlogValue = newBinlogValue();binlogValue.setValue(column.getValue();row.put(column.getName(), binlogValue);北