canal [kə’næl]，譯意為水道/管道/溝渠，主要用途是基於 MySQL 數據庫增量日誌解析，提供增量數據 訂閱 和 消費。應該是阿里雲DTS（Data Transfer Service）的開源版本，開源地址：
https://github.com/alibaba/canal。
canal從1.1.4版本開始引入了admin控制台，有了很多不一樣的配置方式。在搭建過程中如果僅僅按照wiki的用戶手冊，還是容易踩很多坑的。因此，將筆者在搭建過程中的步驟記錄下來，作為官方wiki的 補充，希望能有所幫助。
根據本文內容與搭建順序 ，並搭配對應的官網文檔鏈接，應該就能快速搭建完成了，enjoy~
1. 部署canal-admin
1）部署服務
官方文檔地址：
https://github.com/alibaba/canal/wiki/Canal-Admin-QuickStart
主要配置application.yml文件
server:
port: 8089
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
spring.datasource:
address: 127.0.0.1:3306
database: canal_manager
username: xxxx
assword: xxxxx
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
hikari:
maximum-pool-size: 30
minimum-idle: 1
canal:
adminUser: admin
adminPasswd: admin
這裏需要注意，canal的adminPasswd並不是登陸admin的密碼，登陸admin的密碼是設置在對應的數據庫中的，默認為123456。
另外，因為 Canal Admin 是一個管理系統，需要使用數據庫存放配置信息，只用在 MySQL 中執行 Canal Admin 提供的數據庫初始化文件即可，該文件在“conf/canal_manager.sql”路徑下面。
2）登陸瀏覽器訪問
上面的 Canal Admin 配置好了之後直接根據“/bin/startup.sh”啟動 Canal Admin 即可，在瀏覽器上面輸入 hostip:8089 即可進入到管理頁面，如果使用的默認的配置信息，用戶名入”admin”，密碼輸入”123456”即可訪問首頁。
進入到首頁點擊集群的菜單欄，然後選擇新建集群。
在裏面輸入集群的名稱以及 Zookeeper即可，這裏的集群目前還沒有任務節點，後續通過配置 Canal Server 的自動註冊功能，便可以查看該集群下面擁有的節點。
創建集群后，需要先配置集群 主配置，載入模板即可。
如果沒有載入這個模版，那麼在canal-server執行 sh bin/startup.sh local 命令，讀取canal_local.propeties配置啟動時，會報錯
1 Caused by: com.alibaba.otter.canal.common.CanalException: requestGet for canal config error: canal.properties config is empty
注意，在主配置中，還是需要記得加入zk的地址配置，跟上面配置集群名字的時候輸入的zk無關 (那個zk地址不知道有啥用)：
canal.zkServers = xxx.xx.xx.xx:2181,xxx.xx.xx.xx:2181,xxx.xx.xx.xx:2181
如果希望使用canal的集群模式（推薦使用），記得更改配置使用default-instance.xml，不要使用file-instance.xml配置，如下：
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
2. 部署canal-server
注意，建議先部署好admin，再來部署canal-server，省事不少。
官方文檔地址：
https://github.com/alibaba/canal/wiki/QuickStart
注意，不同於單機版的properties配置，使用admin后，各種配置通過admin全局管理，所以只需要配置canal_local.properties即可。canal_local.properties內容如下：
# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = xxxxxxxxxxxxxx
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
對各個參數說明一下：
- canal.register.ip：用來指定當前 Canal Server 的 IP 信息，如果主機是多網卡，可以避免 IP 信息錯亂的問題。
- canal.admin.passwd：這裏的密碼就是之前配置 Canal Admin 裏面配置的adminPasswd，只不過這裏並不是明文展示，使用 MySQL 的”select password(“admin”)”語句查詢處理過的密碼，注意查詢結果前面的”*”要去掉。
- canal.admin.register.auto：這裡是自動註冊的意思，如果沒有配置，Canal Server 啟動后需要自行在 Canal Admin 上面添加。
- canal.admin.register.cluster：這個配置如果不寫代表當前的 Canal Server 是一個單機節點，如果添加的名字在 Canal Admin 上面沒有提前註冊，Canal Server 啟動時會報錯。
啟動server，切記切記帶上參數local，這樣才會讀取canal_local.properties的配置
sh bin/startup.sh local
啟動完成后，可以在admin界面看到server的連接信息
同時，登陸zk，查看是否已經註冊成功。
3. 配置canal-server
採用admin的集群模式后，集群內的canal-server通過zk做HA，因此，canal-server的配置也只能通過集群做全局配置。
就是前面的集群配置-主配置進入。
另外，canal作為一個增量數據抓取模塊，抓到變更信息后需要投遞。
通過canal.sererMode配置
這裏也限制了，一個集群內的canal只能支持一種投遞模式。
我們目前暫時以投遞RocketMQ為例進行配置。
官方文檔地址：
https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
最終配置文件的配置如下：
#################################################
######### common argument #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
canal.user = xxxx
canal.passwd = xxxxxxxxxxxxxxxxxxx
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = xxxx
canal.admin.passwd = xxxxxxxxxxxxxx
canal.zkServers = xx.xx.xx.xx:xxxx,xx.xx.xx.xxx:xxxx,xx.xx.xx.xx:xxxx
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = RocketMQ
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true
## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false
# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size = 1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60
# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30
# binlog filter config
# 這裏可以對訂閱的消息做過濾
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
# binlog ddl isolation
canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
#################################################
######### destinations #############
#################################################
canal.destinations =
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
canal.instance.global.mode = manager
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
##################################################
######### MQ #############
##################################################
canal.mq.servers = xx.xx.xx.xx:xxxx,xx.xx.xx.xxx:xxxx,xx.xx.xx.xx:xxxx
canal.mq.retries = 0
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
canal.mq.flatMessage = true
canal.mq.compressionType = none
canal.mq.acks = all
#canal.mq.properties. =
canal.mq.producerGroup = test_canal_cluster
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
# aliyun mq namespace
#canal.mq.namespace =
##################################################
######### Kafka Kerberos Info #############
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"
4. 配置canal instance（以投遞MQ為例）
Canal Admin 提供了 Canal Instance 的管理功能。
我們嘗試通過 UI 界面添加需要監聽的數據庫，讓該 Instance 消費 binlog 並將事件發送到 MQ。
- 點擊“新建 Instance”按鈕創建 Instance
- 點擊“載入模板”，進行配置修改。
主要修改以下配置：
- ”canal.instance.mysql.slaveId”：目前1.1.4版本已經不需要配置，系統自動生成
- ”canal.instance.master.address”：配置你的數據庫地址
- canal.instance.dbUsername：數據庫用戶名
- canal.instance.dbPassword：數據庫密碼
- canal.mq.topic：mq的topic
- canal.instance.filter.regex=訂閱的庫表名單（例如：dbvtest\\..*），具體參考https://github.com/alibaba/canal/wiki/AdminGuide
保存，啟動，觀察日誌，沒有報錯即可。
同時可以從 canal-server的 操作-詳情 進入，查看可以看到正在運行的instance
這裏需要注意，如果沒有正確註冊到zk，那麼每個canal-server都會有一個正在運行的instance，會導致變更投遞多次。
如果正確註冊了zk，同一個集群下，每個instance應該只在某一個cannal-server中運行。
可以查看zk路徑
/otter/canal/destinations/{xxxx}/running
xxxx就是instance的 名稱，running節點表示它運行在哪個server上。
另外，目前對binlog的訂閱支持gtid模式和position模式，通過以下參數設置
# enable gtid use true/false
canal.instance.gtidon=false
在demo過程中，發現高可用模式下，gtid無法正常更新到zk，而position模式使用正常。
gtid模式在canal高可用模式下可能存在bug（參考issue：
https://github.com/alibaba/canal/issues/2616），暫時不建議使用。
5. 演練訂閱
在rds執行以下sql
UPDATE `album` SET `attribute`=’5′ WHERE `albumid`=’1′;
在MQ中的消息格式如下：
{
"data":[
{
"albumid":"1",
"picid":"1234",
"attribute":"5",
"lastmodified":"2020-05-15 18:13:35",
"created":"2019-09-04 18:18:51"
}
],
"database":"dbvtest",
"es":1589537615000,
"id":75,
"isDdl":false,
"mysqlType":{
"albumid":"int(10) unsigned",
"picid":"int(10) unsigned",
"attribute":"varchar(200)",
"lastmodified":"timestamp",
"created":"timestamp"
},
"old":[
{
"attribute":"2",
"lastmodified":"2020-05-15 17:42:57"
}
],
"pkNames":[
"albumid"
],
"sql":"",
"sqlType":{
"albumid":4,
"picid":4,
"attribute":12,
"lastmodified":93,
"created":93
},
"table":"album",
"ts":1589537615877,
"type":"UPDATE"
}
6. 監控告警
官方文檔地址：
https://github.com/alibaba/canal/wiki/Prometheus-QuickStart
如果你已經有一套prometheus的監控體系，我們只需要導入模板(
canal/conf/metrics/Canal_instances_tmpl.json)即可，完美～
都看到最後了，原創不易，點個關注，點個贊吧～
知識碎片重新梳理，構建Java知識圖譜： github.com/saigu/JavaK…（歷史文章查閱非常方便）
本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】
※超省錢租車方案
※別再煩惱如何寫文案,掌握八大原則!
※回頭車貨運收費標準
※教你寫出一流的銷售文案?
※產品缺大量曝光嗎?你需要的是一流包裝設計!
※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益
