Canal之配置，多庫監聽
配置說明
介紹配置之前，先了解下canal的配置載入方式：
Spring配置
spring配置的原理是將整個配置抽象為兩部分：
xxxx-instance.xml (canal元件的配置定義，可以在多個instance配置中共享) xxxx.properties (每個instance通道都有各自一份定義，因為每個mysql的ip，帳號，密碼等資訊不會相同)
通過spring的PropertyPlaceholderConfigurer通過機制將其融合，生成一份instance例項物件，每個instance對應的元件都是相互獨立的，互不影響
properties配置檔案
properties配置分為兩部分：
canal.properties (系統根配置檔案) instance.properties (instance級別的配置檔案，每個instance一份)
canal.properties介紹：
canal配置主要分為兩部分定義：
1. instance列表定義 (列出當前server上有多少個instance，每個instance的載入方式是spring/manager等)
| 引數名字 | 引數說明 | 預設值 |
|---|---|---|
| canal.destinations | 當前server上部署的instance列表 | 無 |
| canal.conf.dir | conf/目錄所在的路徑 | ../conf |
| canal.auto.scan | 開啟instance自動掃描 如果配置為true，canal.conf.dir目錄下的instance配置變化會自動觸發： a. instance目錄新增： 觸發instance配置載入，lazy為true時則自動啟動 b. instance目錄刪除：解除安裝對應instance配置，如已啟動則進行關閉 c. instance.properties檔案變化：reload instance配置，如已啟動自動進行重啟操作 |
true |
| canal.auto.scan.interval | instance自動掃描的間隔時間，單位秒 | 5 |
| canal.instance.global.mode | 全域性配置載入方式 | spring |
| canal.instance.global.lazy | 全域性lazy模式 | false |
| canal.instance.global.manager.address | 全域性的manager配置方式的連結資訊 | 無 |
| canal.instance.global.spring.xml | 全域性的spring配置方式的元件檔案 | classpath:spring/file-instance.xml (spring目錄相對於canal.conf.dir) |
| canal.instance.example.mode canal.instance.example.lazy canal.instance.example.spring.xml ….. |
instance級別的配置定義，如有配置，會自動覆蓋全域性配置定義模式 命名規則：canal.instance.{name}.xxx |
無 |
2. common引數定義，比如可以將instance.properties的公用引數，抽取放置到這裡，這樣每個instance啟動的時候就可以共享. 【instance.properties配置定義優先順序高於canal.properties】
| 引數名字 | 引數說明 | 預設值 |
|---|---|---|
| canal.id | 每個canal server例項的唯一標識，暫無實際意義 | 1 |
| canal.ip | canal server繫結的本地IP資訊，如果不配置，預設選擇一個本機IP進行啟動服務 | 無 |
| canal.port | canal server提供socket服務的埠 | 11111 |
| canal.zkServers | canal server連結zookeeper叢集的連結資訊 例子：127.0.0.1:2181,127.0.0.1:2182 |
無 |
| canal.zookeeper.flush.period | canal持久化資料到zookeeper上的更新頻率，單位毫秒 | 1000 |
| canal.file.data.dir | canal持久化資料到file上的目錄 | ../conf (預設和instance.properties為同一目錄，方便運維和備份) |
| canal.file.flush.period | canal持久化資料到file上的更新頻率，單位毫秒 | 1000 |
| canal.instance.memory.batch.mode | canal記憶體store中資料快取模式 1. ITEMSIZE : 根據buffer.size進行限制，只限制記錄的數量 2. MEMSIZE : 根據buffer.size * buffer.memunit的大小，限制快取記錄的大小 |
MEMSIZE |
| canal.instance.memory.buffer.size | canal記憶體store中可快取buffer記錄數，需要為2的指數 | 16384 |
| canal.instance.memory.buffer.memunit | 記憶體記錄的單位大小，預設1KB，和buffer.size組合決定最終的記憶體使用大小 | 1024 |
| canal.instance.transactionn.size | 最大事務完整解析的長度支援超過該長度後，一個事務可能會被拆分成多次提交到canal store中，無法保證事務的完整可見性 | 1024 |
| canal.instance.fallbackIntervalInSeconds | canal發生mysql切換時，在新的mysql庫上查詢binlog時需要往前查詢的時間，單位秒 說明：mysql主備庫可能存在解析延遲或者時鐘不統一，需要回退一段時間，保證資料不丟 |
60 |
| canal.instance.detecting.enable | 是否開啟心跳檢查 | false |
| canal.instance.detecting.sql | 心跳檢查sql | insert into retl.xdual values(1,now()) on duplicate key update x=now() |
| canal.instance.detecting.interval.time | 心跳檢查頻率，單位秒 | 3 |
| canal.instance.detecting.retry.threshold | 心跳檢查失敗重試次數 | 3 |
| canal.instance.detecting.heartbeatHaEnable | 心跳檢查失敗後，是否開啟自動mysql自動切換 說明：比如心跳檢查失敗超過閥值後，如果該配置為true，canal就會自動鏈到mysql備庫獲取binlog資料 |
false |
| canal.instance.network.receiveBufferSize | 網路連結引數，SocketOptions.SO_RCVBUF | 16384 |
| canal.instance.network.sendBufferSize | 網路連結引數，SocketOptions.SO_SNDBUF | 16384 |
| canal.instance.network.soTimeout | 網路連結引數，SocketOptions.SO_TIMEOUT | 30 |
| canal.instance.filter.query.dcl | 是否忽略DCL的query語句，比如grant/create user等 | false |
| canal.instance.filter.query.dml | 是否忽略DML的query語句，比如insert/update/delete table.(mysql5.6的ROW模式可以包含statement模式的query記錄) | false |
| canal.instance.filter.query.ddl | 是否忽略DDL的query語句，比如create table/alater table/drop table/rename table/create index/drop index. (目前支援的ddl型別主要為table級別的操作，create databases/trigger/procedure暫時劃分為dcl型別) | false |
| canal.instance.get.ddl.isolation | ddl語句是否隔離傳送，開啟隔離可保證每次只返回傳送一條ddl資料，不和其他dml語句混合返回.(otter ddl同步使用) | false |
instance.properties介紹：
a. 在canal.properties定義了canal.destinations後，需要在canal.conf.dir對應的目錄下建立同名的檔案
比如：
canal.destinations = example1,example2這時需要建立example1和example2兩個目錄，每個目錄裡各自有一份instance.properties.
ps. canal自帶了一份instance.properties demo，可直接複製conf/example目錄進行配置修改
cp -R example example1/
cp -R example example2/
b. 如果canal.properties未定義instance列表，但開啟了canal.auto.scan時
server第一次啟動時，會自動掃描conf目錄下，將檔名做為instance name，啟動對應的instance server執行過程中，會根據canal.auto.scan.interval定義的頻率，進行掃描
發現目錄有新增，啟動新的instance 發現目錄有刪除，關閉老的instance 發現對應目錄的instance.properties有變化，重啟instance
一個標準的conf目錄結果：
jianghang@jianghang-laptop:~/work/canal/deployer/target/canal$ ls -l conf/ 總用量 8 -rwxrwxrwx 1 jianghang jianghang 1677 2013-03-19 15:03 canal.properties ##系統配置 drwxr-xr-x 2 jianghang jianghang 88 2013-03-19 15:03 example ## instance配置 -rwxrwxrwx 1 jianghang jianghang 1840 2013-03-19 15:03 logback.xml ## 日誌檔案 drwxr-xr-x 2 jianghang jianghang 168 2013-03-19 17:04 spring ## spring instance模板
| 數名字 | 引數說明 | 預設值 |
|---|---|---|
| canal.instance.mysql.slaveId | mysql叢集配置中的serverId概念，需要保證和當前mysql叢集中id唯一 | 1234 |
| canal.instance.master.address | mysql主庫連結地址 | 127.0.0.1:3306 |
| canal.instance.master.journal.name | mysql主庫連結時起始的binlog檔案 | 無 |
| canal.instance.master.position | mysql主庫連結時起始的binlog偏移量 | 無 |
| canal.instance.master.timestamp | mysql主庫連結時起始的binlog的時間戳 | 無 |
| canal.instance.dbUsername | mysql資料庫帳號 | canal |
| canal.instance.dbPassword | mysql資料庫密碼 | canal |
| canal.instance.defaultDatabaseName | mysql連結時預設schema | |
| canal.instance.connectionCharset mysql 資料解析編碼 | UTF-8 | |
| canal.instance.filter.regex | mysql 資料解析關注的表，Perl正則表示式.多個正則之間以逗號(,)分隔，轉義符需要雙斜槓(\\) 常見例子： 1. 所有表：.* or .*\\..* 2. canal schema下所有表： canal\\..* 3. canal下的以canal打頭的表：canal\\.canal.* 4. canal schema下的一張表：canal.test1 5. 多個規則組合使用：canal\\..*,mysql.test1,mysql.test2 (逗號分隔) 注意：此過濾條件只針對row模式的資料有效(ps. mixed/statement因為不解析sql，所以無法準確提取tableName進行過濾) |
.*\\..* |
幾點說明：
mysql連結時的起始位置
canal.instance.master.journal.name + canal.instance.master.position : 精確指定一個binlog位點，進行啟動 canal.instance.master.timestamp : 指定一個時間戳，canal會自動遍歷mysql binlog，找到對應時間戳的binlog位點後，進行啟動不指定任何資訊：預設從當前資料庫的位點，進行啟動。(show master status)
mysql解析關注表定義
標準的Perl正則，注意轉義時需要雙斜槓：\\
mysql連結的編碼
目前canal版本僅支援一個數據庫只有一種編碼，如果一個庫存在多個編碼，需要通過filter.regex配置，將其拆分為多個canal instance，為每個instance指定不同的編碼
instance.xml配置檔案
目前預設支援的instance.xml有以下幾種：
spring/memory-instance.xml spring/file-instance.xml spring/default-instance.xml spring/group-instance.xml
在介紹instance配置之前，先了解一下canal如何維護一份增量訂閱&消費的關係資訊：
解析位點 (parse模組會記錄，上一次解析binlog到了什麼位置，對應元件為：CanalLogPositionManager) 消費位點 (canal server在接收了客戶端的ack後，就會記錄客戶端提交的最後位點，對應的元件為：CanalMetaManager)
對應的兩個位點元件，目前都有幾種實現：
memory (memory-instance.xml中使用) zookeeper mixed file (file-instance.xml中使用，集合了file+memory模式，先寫記憶體，定時重新整理資料到本地file上) period (default-instance.xml中使用，集合了zookeeper+memory模式，先寫記憶體，定時重新整理資料到zookeeper上)
memory-instance.xml介紹：
所有的元件(parser , sink , store)都選擇了記憶體版模式，記錄位點的都選擇了memory模式，重啟後又會回到初始位點進行解析
特點：速度最快，依賴最少(不需要zookeeper)
場景：一般應用在quickstart，或者是出現問題後，進行資料分析的場景，不應該將其應用於生產環境
file-instance.xml介紹：
所有的元件(parser , sink , store)都選擇了基於file持久化模式，注意，不支援HA機制.
特點：支援單機持久化
場景：生產環境，無HA需求，簡單可用.
default-instance.xml介紹：
所有的元件(parser , sink , store)都選擇了持久化模式，目前持久化的方式主要是寫入zookeeper，保證資料叢集共享.
特點：支援HA
場景：生產環境，叢集化部署.
group-instance.xml介紹：
主要針對需要進行多庫合併時，可以將多個物理instance合併為一個邏輯instance，提供客戶端訪問。
場景：分庫業務。 比如產品資料拆分了4個庫，每個庫會有一個instance，如果不用group，業務上要消費資料時，需要啟動4個客戶端，分別連結4個instance例項。使用group後，可以在canal server上合併為一個邏輯instance，只需要啟動1個客戶端，連結這個邏輯instance即可.
instance.xml設計初衷：
允許進行自定義擴充套件，比如實現了基於資料庫的位點管理後，可以自定義一份自己的instance.xml，整個canal設計中最大的靈活性在於此HA模式配置
機器準備
執行canal的機器： 10.20.144.22 , 10.20.144.51. zookeeper地址為10.20.144.51:2181 mysql地址：10.20.144.15:3306
按照部署和配置，在單臺機器上各自完成配置，演示時instance name為example
修改canal.properties，加上zookeeper配置
canal.zkServers=10.20.144.51:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
建立example目錄，並修改instance.properties
canal.instance.mysql.slaveId = 1234 ##另外一臺機器改成1235，保證slaveId不重複即可
canal.instance.master.address = 10.20.144.15:3306
注意： 兩臺機器上的instance目錄的名字需要保證完全一致，HA模式是依賴於instance name進行管理，同時必須都選擇default-instance.xml配置
啟動兩臺機器的canal
-------
ssh 10.20.144.51
sh bin/startup.sh
--------
ssh 10.20.144.22
sh bin/startup.sh
啟動後，你可以檢視logs/example/example.log，只會看到一臺機器上出現了啟動成功的日誌。
比如我這裡啟動成功的是10.20.144.51
2013-03-19 18:18:20.590 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties] 2013-03-19 18:18:20.596 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties] 2013-03-19 18:18:20.831 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 2013-03-19 18:18:20.845 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful...
檢視一下zookeeper中的節點資訊，也可以知道當前工作的節點為10.20.144.51:11111
[zk: localhost:2181(CONNECTED) 15] get /otter/canal/destinations/example/running
{"active":true,"address":"10.20.144.51:11111","cid":1}
客戶端連結, 消費資料
a. 可以直接指定zookeeper地址和instance name，canal client會自動從zookeeper中的running節點，獲取當前服務的工作節點，然後與其建立連結：
CanalConnector connector = CanalConnectors.newClusterConnector("10.20.144.51:2181", "example", "", "");
b. 連結成功後，canal server會記錄當前正在工作的canal client資訊，比如客戶端ip，連結的埠資訊等 (聰明的你，應該也可以發現，canal client也可以支援HA功能)
[zk: localhost:2181(CONNECTED) 17] get /otter/canal/destinations/example/1001/running
{"active":true,"address":"10.12.48.171:50544","clientId":1001}
c. 資料消費成功後，canal server會在zookeeper中記錄下當前最後一次消費成功的binlog位點. (下次你重啟client時，會從這最後一個位點繼續進行消費)
[zk: localhost:2181(CONNECTED) 16] get /otter/canal/destinations/example/1001/cursor
{"@type":"com.alibaba.otter.canal.protocol.position.LogPosition","identity":{"slaveId":-1,"sourceAddress":{"address":"10.20.144.15","port":3306}},"postion":{"included":false,"journalName":"mysql-bin.002253","position":2574756,"timestamp":1363688722000}}
重啟一下canal server
停止正在工作的10.20.144.51的canal server
ssh 10.20.144.51 sh bin/stop.sh
這時10.20.144.22會立馬啟動example instance，提供新的資料服務
[zk: localhost:2181(CONNECTED) 19] get /otter/canal/destinations/example/running
{"active":true,"address":"10.20.144.22:11111","cid":1}
與此同時，客戶端也會隨著canal server的切換，通過獲取zookeeper中的最新地址，與新的canal server建立連結，繼續消費資料，整個過程自動完成
觸發HA自動切換場景 (server/client HA模式都有效)
正常場景
正常關閉canal server(會釋放instance的所有資源，包括刪除running節點) 平滑切換(gracefully)
操作：更新對應instance的running節點內容，將”active”設定為false，對應的running節點收到訊息後，會主動釋放running節點，讓出控制權但自己jvm不退出，gracefully.
{"active":false,"address":"10.20.144.22:11111","cid":1}
異常場景
canal server對應的jvm異常crash，running節點的釋放會在對應的zookeeper session失效後，釋放running節點(EPHEMERAL節點)
ps. session過期時間預設為zookeeper配置檔案中定義的tickTime的20倍，如果不改動zookeeper配置，那預設就是40秒
canal server所在的網路出現閃斷，導致zookeeper認為session失效，釋放了running節點，此時canal server對應的jvm並未退出，(一種假死狀態，非常特殊的情況)
ps. 為了保護假死狀態的canal server，避免因瞬間runing失效導致instance重新分佈，所以做了一個策略：canal server在收到running節點釋放後，延遲一段時間搶佔running，原本running節點的擁有者可以不需要等待延遲，優先取得running節點，可以保證假死狀態下儘可能不無謂的釋放資源。 目前延遲時間的預設值為5秒，即running節點針對假死狀態的保護期為5秒.
mysql多節點解析配置(parse解析自動切換)
mysql機器準備
準備兩臺mysql機器，配置為M-M模式，比如ip為：10.20.144.25:3306，10.20.144.29:3306
[mysqld] xxxxx ##其他正常master/slave配置 log_slave_updates=true ##這個配置一定要開啟
canal instance配置
# position info canal.instance.master.address = 10.20.144.25:3306 canal.instance.master.journal.name = canal.instance.master.position = canal.instance.master.timestamp = canal.instance.standby.address = 10.20.144.29:3306 canal.instance.standby.journal.name = canal.instance.standby.position = canal.instance.standby.timestamp = ## detecing config canal.instance.detecting.enable = true ## 需要開啟心跳檢查 canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() ##心跳檢查sql，也可以選擇類似select 1的query語句 canal.instance.detecting.interval.time = 3 ##心跳檢查頻率 canal.instance.detecting.retry.threshold = 3 ## 心跳檢查失敗次數閥值，超過該閥值後會觸發mysql連結切換，比如切換到standby機器上繼續消費binlog canal.instance.detecting.heartbeatHaEnable = true ## 心跳檢查超過失敗次數閥值後，是否開啟master/standby的切換.
注意：
填寫master/standby的地址和各自的起始binlog位置，目前配置只支援一個standby配置. 發生master/standby的切換的條件：(heartbeatHaEnable = true) && (失敗次數>=retry.threshold). 多引入一個heartbeatHaEnable的考慮：開啟心跳sql有時候是為client檢測canal server是否正常工作，如果定時收到了心跳語句，那說明整個canal server工作正常
啟動 & 測試
比如關閉一臺機器的mysql , /etc/init.d/mysql stop 。在經歷大概 interval.time * retry.threshold時間後，就會切換到standby機器上