一敏晤、背景

早期，阿里巴巴B2B公司因?yàn)榇嬖诤贾莺兔绹p機(jī)房部署缅茉，存在跨機(jī)房同步的業(yè)務(wù)需求嘴脾。不過早期的數(shù)據(jù)庫同步業(yè)務(wù)，主要是基于trigger的方式獲取增量變更蔬墩，不過從2010年開始译打，阿里系公司開始逐步的嘗試基于數(shù)據(jù)庫的日志解析，獲取增量變更進(jìn)行同步筹我，由此衍生出了增量訂閱&消費(fèi)的業(yè)務(wù)扶平，從此開啟了一段新紀(jì)元帆离。
ps. 目前內(nèi)部版本已經(jīng)支持mysql和oracle部分版本的日志解析蔬蕊，當(dāng)前的canal開源版本支持5.7及以下的版本(阿里內(nèi)部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)

基于日志增量訂閱&消費(fèi)支持的業(yè)務(wù)：

數(shù)據(jù)庫鏡像
數(shù)據(jù)庫實(shí)時(shí)備份
多級(jí)索引 (賣家和買家各自分庫索引)
search build
業(yè)務(wù)cache刷新
價(jià)格變化等重要業(yè)務(wù)消息

二、項(xiàng)目介紹

名稱：canal [k?'n?l]
譯意： 水道/管道/溝渠
語言： 純java開發(fā)
定位： 基于數(shù)據(jù)庫增量日志解析哥谷，提供增量數(shù)據(jù)訂閱&消費(fèi)岸夯，目前主要支持了mysql

三、工作原理

1们妥、mysql主備復(fù)制實(shí)現(xiàn)

2、canal的工作原理：

四鱼冀、canal安裝

1报破、下載并解壓縮

[root@master1 canal]#wget https://github.com/alibaba/canal/releases/download/v1.0.23/canal.deployer-1.0.23.tar.gz
[root@master1 canal]#mkdir /root/canal
[root@master1 canal]#tar zxvf canal.deployer-1.0.23.tar.gz  -C /root/canal

2、創(chuàng)建mysql的canal用戶

mysql> CREATE USER 'canal'@'localhost' IDENTIFIED BY 'canal';
Query OK, 0 rows affected (0.00 sec)
mysql> GRANT ALL PRIVILEGES ON *.* TO 'canal'@'localhost' WITH GRANT OPTION;
Query OK, 0 rows affected (0.01 sec)
mysql> CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
Query OK, 0 rows affected (0.00 sec)
mysql> GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' WITH GRANT OPTION;
Query OK, 0 rows affected (0.00 sec)
mysql> flush privileges;
Query OK, 0 rows affected (0.00 sec)

3千绪、修改canal配置文件(如果是訪問本機(jī)充易，并且用戶密碼都為canal則不需要修改配置文件)

[root@master1 canal]# vi /root/canal/conf/example/instance.properties
#################################################
## mysql serverId
canal.instance.mysql.slaveId = 1234
# position info
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
# username/password
#修改這兩項(xiàng)
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
# table regex
canal.instance.filter.regex = .*\\..*
# table black regex
canal.instance.filter.black.regex =
#################################################

4、啟動(dòng)canal

[root@master1 canal]# sh /root/canal/bin/startup.sh
cd to /root/canal/bin for workaround relative path
LOG CONFIGURATION : /root/canal/bin/../conf/logback.xml
canal conf : /root/canal/bin/../conf/canal.properties
CLASSPATH :/root/canal/bin/../conf:/root/canal/bin/../lib/zookeeper-3.4.5.jar:/root/canal/bin/../lib/zkclient-0.1.jar:/root/canal/bin/../lib/spring-2.5.6.jar:/root/canal/bin/../lib/slf4j-api-1.7.12.jar:/root/canal/bin/../lib/protobuf-java-2.6.1.jar:/root/canal/bin/../lib/oro-2.0.8.jar:/root/canal/bin/../lib/netty-3.2.5.Final.jar:/root/canal/bin/../lib/logback-core-1.1.3.jar:/root/canal/bin/../lib/logback-classic-1.1.3.jar:/root/canal/bin/../lib/log4j-1.2.14.jar:/root/canal/bin/../lib/jcl-over-slf4j-1.7.12.jar:/root/canal/bin/../lib/guava-18.0.jar:/root/canal/bin/../lib/fastjson-1.1.35.jar:/root/canal/bin/../lib/commons-logging-1.1.1.jar:/root/canal/bin/../lib/commons-lang-2.6.jar:/root/canal/bin/../lib/commons-io-2.4.jar:/root/canal/bin/../lib/commons-beanutils-1.8.2.jar:/root/canal/bin/../lib/canal.store-1.0.23.jar:/root/canal/bin/../lib/canal.sink-1.0.23.jar:/root/canal/bin/../lib/canal.server-1.0.23.jar:/root/canal/bin/../lib/canal.protocol-1.0.23.jar:/root/canal/bin/../lib/canal.parse.driver-1.0.23.jar:/root/canal/bin/../lib/canal.parse.dbsync-1.0.23.jar:/root/canal/bin/../lib/canal.parse-1.0.23.jar:/root/canal/bin/../lib/canal.meta-1.0.23.jar:/root/canal/bin/../lib/canal.instance.spring-1.0.23.jar:/root/canal/bin/../lib/canal.instance.manager-1.0.23.jar:/root/canal/bin/../lib/canal.instance.core-1.0.23.jar:/root/canal/bin/../lib/canal.filter-1.0.23.jar:/root/canal/bin/../lib/canal.deployer-1.0.23.jar:/root/canal/bin/../lib/canal.common-1.0.23.jar:/root/canal/bin/../lib/aviator-2.2.1.jar:.:/usr/java/jdk1.8.0_121/lib
cd to /root/canal for continue

5荸型、關(guān)閉canal

[root@master1 canal]# sh /root/canal/bin/stop.sh
master1: stopping canal 16062 ... 
Oook! cost:1

6盹靴、相關(guān)日志位置

[root@master1 canal]# cat /root/canal/logs/canal/canal.log
[root@master1 canal]# cat /root/canal/logs/example/example.log

五、編寫代碼讀取binlog

1、創(chuàng)建maven項(xiàng)目

Paste_Image.png

2鹉究、pom.xml增加依賴

        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.0.12</version>
        </dependency>

3宇立、編寫ClientSample類訂閱binlog

>package com.chainfin.canal;
>
>import java.net.InetSocketAddress;
>import java.util.List;
>
>import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
>
/**
 * 
 * @author jinxiaoxin
 *
 */
public class ClientSample {
    public static void main(String[] args) {
        // 創(chuàng)建鏈接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("10.105.10.121", 11111), "example", "",
                "");// AddressUtils.getHostIp(),
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe("test\\..*");// .*代表database，..*代表table
            connector.rollback();//
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數(shù)量的數(shù)據(jù)
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    // System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n",
                    // batchId, size);
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交確認(rèn)
                // connector.rollback(batchId); // 處理失敗, 回滾數(shù)據(jù)
            }
            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }
>
    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException(
                        "ERROR ## parser of eromanga-event has an error,data:"
                                + entry.toString(), e);
            }
            EventType eventType = rowChage.getEventType();
            System.out
                    .println(String
                            .format("================> binlog[%s:%s] ,name[%s,%s] , eventType : %s",
                                    entry.getHeader().getLogfileName(), entry
                                            .getHeader().getLogfileOffset(),
                                    entry.getHeader().getSchemaName(), entry
                                            .getHeader().getTableName(),
                                    eventType));
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }
>
    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue()
                    + "    update=" + column.getUpdated());
        }
    }
}
>

4自赔、MYSQL進(jìn)行增刪改查

mysql> CREATE TABLE `xdual` (
    ->        `ID` int(11) NOT NULL AUTO_INCREMENT,
    ->        `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    ->     PRIMARY KEY (`ID`)
    ->      ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;
Query OK, 0 rows affected (0.01 sec)
mysql> insert into xdual(id,x) values(1,now());
Query OK, 1 row affected (0.01 sec)

5妈嘹、eclipse控制臺(tái)可以看到如下輸出

Paste_Image.png

六、canal集群搭建

1绍妨、需要安裝zookeeper

自行了解安裝及其使用润脸，此處略過

2、修改canal配置文件

修改兩個(gè)配置

[root@master1 canal]# vi /root/canal/conf/canal.properties
# 用逗號(hào)隔開且不留空格 
canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

3他去、拷貝canal到另一臺(tái)機(jī)器并解壓縮(作為從節(jié)點(diǎn))

[root@master1 canal]# scp canal.deployer-1.0.23.tar.gz root@10.105.10.122:/root
canal.deployer-1.0.23.tar.gz          100% 9586KB   9.4MB/s   00:00   
[root@master2 ~]# mkdir canal
[root@master2 ~]# tar -zxvf canal.deployer-1.0.23.tar.gz -C /root/canal                           

4毙驯、修改slave配置文件

[root@master2 canal]# vi /root/canal/conf/example/instance.properties
## mysql serverId修改與master不同即可
canal.instance.mysql.slaveId = 1235
# position info修改成mysql地址即可
canal.instance.master.address = 10.105.10.121:3306
[root@master2 canal]# vi /root/canal/conf/canal.properties
# 用逗號(hào)隔開且不留空格 
canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

5、兩臺(tái)機(jī)器分別啟動(dòng)canal

[root@master2 canal]#  sh /root/canal/bin/startup.sh

6灾测、查看canal在zk中的狀態(tài)

[zk: localhost:2181(CONNECTED) 11] get /otter/canal/destinations/example/running
{"active":true,"address":"10.105.10.121:11111","cid":1}
cZxid = 0x250000f037
ctime = Thu Apr 13 13:42:04 CST 2017
mZxid = 0x250000f037
mtime = Thu Apr 13 13:42:04 CST 2017
pZxid = 0x250000f037
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15b5b11e9cd16e1
dataLength = 55
numChildren = 0

10.105.10.121為active爆价，所以10.105.10.122位stangby節(jié)點(diǎn)

7、測(cè)試HA是否配置成功

stop掉10.105.10.121這臺(tái)機(jī)器

[root@master1 canal]# sh /root/canal/bin/stop.sh
master1: stopping canal 6107 ... 
Oook! cost:1

[zk: localhost:2181(CONNECTED) 28] get /otter/canal/destinations/example/running
{"active":true,"address":"10.105.10.122:11111","cid":1}
cZxid = 0x250000f0a7
ctime = Thu Apr 13 13:46:10 CST 2017
mZxid = 0x250000f0a7
mtime = Thu Apr 13 13:46:10 CST 2017
pZxid = 0x250000f0a7
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x35b5b11e9d816cb
dataLength = 55

很明顯active已經(jīng)切換到10.105.10.122上了

8媳搪、客戶端代碼

//創(chuàng)建connector 為集群connector 
CanalConnector connector = CanalConnectors.newClusterConnector("10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181","example", "", "");

啟動(dòng)代碼

Paste_Image.png

9铭段、錯(cuò)誤解決

當(dāng)這個(gè)配置的ip不配成127.0.0.1時(shí)(也就是本地)
canal.instance.master.address = 10.105.10.121:3306
查看canal配置文件，會(huì)報(bào)如下錯(cuò)誤導(dǎo)致秦爆，無法讀取binlog

[root@master1 example]# tail -f example.log
2017-04-13 14:29:23.646 [destination = example , address = master1/10.105.10.121:3306 , EventParser] ERROR c.a.o.canal.parse.inbound.mysql.dbsync.DirectLogFetcher - I/O error while reading from client socket
java.io.IOException: Received error packet: errno = 1236, sqlstate = HY000 errmsg = Could not find first log file name in binary log index file
    at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.fetch(DirectLogFetcher.java:95) ~[canal.parse-1.0.23.jar:na]
    at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.dump(MysqlConnection.java:121) [canal.parse-1.0.23.jar:na]
    at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:209) [canal.parse-1.0.23.jar:na]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]

解決辦法：
登錄mysql

mysql> show variables like "%sum%";
+---------------------------+-------+
| Variable_name             | Value |
+---------------------------+-------+
| binlog_checksum           | CRC32 |
| innodb_checksum_algorithm | crc32 |
| innodb_checksums          | ON    |
| innodb_log_checksums      | ON    |
| master_verify_checksum    | OFF   |
| slave_sql_verify_checksum | ON    |
+---------------------------+-------+
6 rows in set (0.00 sec)

查詢資料發(fā)現(xiàn)mysql版本為5.6時(shí)序愚，
這個(gè)錯(cuò)誤一般出現(xiàn)在master5.6,slave在低版本的情況下。這是由于5.6使用了crc32做binlog的checksum等限；
當(dāng)一個(gè)event被寫入binary log(二進(jìn)制日志)的時(shí)候爸吮，checksum也同時(shí)寫入binary log，然后在event通過網(wǎng)絡(luò)傳輸?shù)綇?a target="_blank" rel="nofollow">服務(wù)器(slave)之后望门，再在從服務(wù)器中對(duì)其進(jìn)行驗(yàn)證并寫入從服務(wù)器的relay log.
由于每一步都記錄了event和checksum形娇，所以我們可以很快地找出問題所在。
在master1中設(shè)置binlog_checksum =none怒允；

mysql> set global binlog_checksum='NONE';
Query OK, 0 rows affected (0.00 sec)
mysql>  show variables like "%sum%";
+---------------------------+-------+
| Variable_name             | Value |
+---------------------------+-------+
| binlog_checksum           | NONE  |
| innodb_checksum_algorithm | crc32 |
| innodb_checksums          | ON    |
| innodb_log_checksums      | ON    |
| master_verify_checksum    | OFF   |
| slave_sql_verify_checksum | ON    |
+---------------------------+-------+
6 rows in set (0.01 sec)

這樣錯(cuò)誤就可以解決了

七埂软、properties配置文件參數(shù)解釋

properties配置分為兩部分：

canal.properties  (系統(tǒng)根配置文件)
instance.properties  (instance級(jí)別的配置文件，每個(gè)instance一份)

1纫事、canal.properties

canal.destinations  #當(dāng)前server上部署的instance列表 
canal.conf.dir  #conf/目錄所在的路徑   
canal.auto.scan #開啟instance自動(dòng)掃描
#如果配置為true勘畔，canal.conf.dir目錄下的instance配置變化會(huì)自動(dòng)觸發(fā)：
#a. instance目錄新增： 觸發(fā)instance配置載入，lazy為true時(shí)則自動(dòng)啟動(dòng)
#b. instance目錄刪除：卸載對(duì)應(yīng)instance配置丽惶，如已啟動(dòng)則進(jìn)行關(guān)閉
#c. instance.properties文件變化：reload instance配置炫七，如已啟動(dòng)自動(dòng)進(jìn)行重啟操作
canal.auto.scan.interval    #instance自動(dòng)掃描的間隔時(shí)間，單位秒  
canal.instance.global.mode  #全局配置加載方式   
canal.instance.global.lazy  #全局lazy模式   
canal.instance.global.manager.address   #全局的manager配置方式的鏈接信息    無
canal.instance.global.spring.xml    #全局的spring配置方式的組件文件 
canal.instance.example.mode
canal.instance.example.lazy
canal.instance.example.spring.xml
#instance級(jí)別的配置定義钾唬，如有配置万哪，會(huì)自動(dòng)覆蓋全局配置定義模式
#命名規(guī)則：canal.instance.{name}.xxx 無

2侠驯、instance.properties

canal.id    #每個(gè)canal server實(shí)例的唯一標(biāo)識(shí)，暫無實(shí)際意義   
canal.ip    #canal server綁定的本地IP信息奕巍，如果不配置吟策，默認(rèn)選擇一個(gè)本機(jī)IP進(jìn)行啟動(dòng)服務(wù)   
canal.port  #canal server提供socket服務(wù)的端口  
canal.zkServers #canal server鏈接zookeeper集群的鏈接信息
#例子：10.20.144.22:2181,10.20.144.51:2181 
canal.zookeeper.flush.period    #canal持久化數(shù)據(jù)到zookeeper上的更新頻率，單位毫秒    
canal.instance.memory.batch.mode    #canal內(nèi)存store中數(shù)據(jù)緩存模式
#1. ITEMSIZE : 根據(jù)buffer.size進(jìn)行限制的止，只限制記錄的數(shù)量
#2. MEMSIZE : 根據(jù)buffer.size  * buffer.memunit的大小檩坚，限制緩存記錄的大小  
canal.instance.memory.buffer.size   #canal內(nèi)存store中可緩存buffer記錄數(shù)，需要為2的指數(shù)  
canal.instance.memory.buffer.memunit    #內(nèi)存記錄的單位大小诅福，默認(rèn)1KB匾委，和buffer.size組合決定最終的內(nèi)存使用大小  
canal.instance.transactionn.size    最大事務(wù)完整解析的長度支持
超過該長度后，一個(gè)事務(wù)可能會(huì)被拆分成多次提交到canal store中氓润，無法保證事務(wù)的完整可見性
canal.instance.fallbackIntervalInSeconds    #canal發(fā)生mysql切換時(shí)赂乐，在新的mysql庫上查找binlog時(shí)需要往前查找的時(shí)間，單位秒
#說明：mysql主備庫可能存在解析延遲或者時(shí)鐘不統(tǒng)一咖气，需要回退一段時(shí)間挨措，保證數(shù)據(jù)不丟
canal.instance.detecting.enable #是否開啟心跳檢查
canal.instance.detecting.sql    #心跳檢查sql    insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.interval.time  #心跳檢查頻率，單位秒
canal.instance.detecting.retry.threshold    #心跳檢查失敗重試次數(shù)
canal.instance.detecting.heartbeatHaEnable  #心跳檢查失敗后采章，是否開啟自動(dòng)mysql自動(dòng)切換
#說明：比如心跳檢查失敗超過閥值后运嗜，如果該配置為true，canal就會(huì)自動(dòng)鏈到mysql備庫獲取binlog數(shù)據(jù)
canal.instance.network.receiveBufferSize    #網(wǎng)絡(luò)鏈接參數(shù)悯舟，SocketOptions.SO_RCVBUF 
canal.instance.network.sendBufferSize   #網(wǎng)絡(luò)鏈接參數(shù)，SocketOptions.SO_SNDBUF 
canal.instance.network.soTimeout    #網(wǎng)絡(luò)鏈接參數(shù)砸民，SocketOptions.SO_TIMEOUT