參考：https://www.2cto.com/database/201609/547661.html
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對應的目錄下建立同名的文件
比如：
|
1
|
canal.destinations = example1,example2
|
這時需要創建example1和example2兩個目錄，每個目錄里各自有一份instance.properties.
ps. canal自帶了一份instance.properties demo，可直接復制conf/example目錄進行配置修改
|
1
2
|
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目錄結果：
|
1
2
3
4
5
6
|
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介紹：
|
1
2
3
4
5
|
所有的組件(parser , sink , store)都選擇了內存版模式，記錄位點的都選擇了memory模式，重啟后又會回到初始位點進行解析
特點：速度最快，依賴最少(不需要zookeeper)
場景：一般應用在quickstart，或者是出現問題后，進行數據分析的場景，不應該將其應用於生產環境
|
file-instance.xml介紹：
|
1
2
3
4
5
|
所有的組件(parser , sink , store)都選擇了基於file持久化模式，注意，不支持HA機制.
特點：支持單機持久化
場景：生產環境，無HA需求，簡單可用.
|
default-instance.xml介紹：
|
1
2
3
4
5
|
所有的組件(parser , sink , store)都選擇了持久化模式，目前持久化的方式主要是寫入zookeeper，保證數據集群共享.
特點：支持HA
場景：生產環境，集群化部署.
|
group-instance.xml介紹：
|
1
2
3
|
主要針對需要進行多庫合並時，可以將多個物理instance合並為一個邏輯instance，提供客戶端訪問。
場景：分庫業務。 比如產品數據拆分了4個庫，每個庫會有一個instance，如果不用
group
，業務上要消費數據時，需要啟動4個客戶端，分別鏈接4個instance實例。使用
group
后，可以在canal server上合並為一個邏輯instance，只需要啟動1個客戶端，鏈接這個邏輯instance即可.
|
instance.xml設計初衷：
|
1
|
允許進行自定義擴展，比如實現了基於數據庫的位點管理后，可以自定義一份自己的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配置
|
1
2
|
canal.zkServers=10.20.144.51:2181
canal.instance.
global
.spring.xml = classpath:spring/
default
-instance.xml
|
創建example目錄，並修改instance.properties
|
1
2
|
canal.instance.mysql.slaveId = 1234 ##另外一台機器改成1235，保證slaveId不重復即可
canal.instance.master.address = 10.20.144.15:3306
|
注意： 兩台機器上的instance目錄的名字需要保證完全一致，HA模式是依賴於instance name進行管理，同時必須都選擇default-instance.xml配置
啟動兩台機器的canal
|
1
2
3
4
5
6
|
-------
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
|
1
2
3
4
|
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
|
1
2
|
[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節點，獲取當前服務的工作節點，然后與其建立鏈接：
|
1
|
CanalConnector connector = CanalConnectors.newClusterConnector(
"10.20.144.51:2181"
,
"example"
,
""
,
""
);
|
b. 鏈接成功后，canal server會記錄當前正在工作的canal client信息，比如客戶端ip，鏈接的端口信息等 (聰明的你，應該也可以發現，canal client也可以支持HA功能)
|
1
2
|
[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時，會從這最后一個位點繼續進行消費)
|
1
2
|
[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
|
1
2
|
ssh 10.20.144.51
sh bin/stop.sh
|
這時10.20.144.22會立馬啟動example instance，提供新的數據服務
|
1
2
|
[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.
|
1
|
{
"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
|
1
2
3
|
[mysqld]
xxxxx ##其他正常master/slave配置
log_slave_updates=
true
##這個配置一定要打開
|
canal instance配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
# 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機器上