Canal HA原理及安裝

一、??????????????Canal HA原理

Canal一般用于實時同步數據場景，那么對于實時場景HA顯得尤為重要，Canal支持HA搭建，canal的HA分為兩部分，canal server和canal client分別有對應的HA實現。大數據中使用Canal同步數據一般同步到Kafka中，這里Kafka相當于是Canal Client，Kafka集群自帶HA屬性，所以這里我們只關注Canal Server HA。Canal Server HA主要是為了減少對mysql dump的請求，不同server上的instance（不同server上的相同instance）要求同一時間只能有一個處于running，其他的處于standby狀態（standby是instance的狀態），Canal Server HA原理如下：

Canal HA 保證步驟如下：

【資料圖】

canal server要啟動某個canal instance時都先向zookeeper_進行一次嘗試啟動判斷。創建zookeeper節點成功后，對應的canal server就啟動對應的canal instance，沒有創建成功的canal instance就會處于standby狀態。一旦zookeeper發現canal server A創建的instance節點消失后，立即通知其他的canal server再次進行步驟1的操作，重新選出一個canal server啟動instance。canal client每次進行connect時，會首先向zookeeper詢問當前是誰啟動了canal instance，然后和其建立鏈接，一旦鏈接不可用，會重新嘗試connect。

二、??????????????Canal HA 搭建

1、機器準備

運行Canal的機器：node3,node4

zookeeper地址:node3:2181,node4:2181,node5:2181

mysql地址：node2:3306

2、在node3，node4上單獨部署配置Canal

將Canal安裝包上傳到node3，node4，并解壓到“/software/canal”目錄下，修改“/software/canal/conf”下的canal.properties文件，加上zookeeper配置

#指定zookeeper集群地址canal.zkServers = node3:2181,node4:2181,node5:2181#配置spring的xml配置文件canal.instance.global.spring.xml = classpath:spring/default-instance.xml#canal將數據寫入Kafka，可配：tcp, kafka, RocketMQ，tcp就是使用canal代碼接收canal.serverMode = kafka#配置canal寫入Kafka地址canal.mq.servers = node1:9092,node2:9092,node3:9092

復制

進入“/software/canal/conf/example”目錄，修改“instance.properties”文件：

#另外一臺機器改成123457，保證slaveId不重復即可canal.instance.mysql.slaveId=123456#配置mysql master 節點及端口canal.instance.master.address=node2:3306#配置連接mysql的用戶名和密碼，就是前面復制權限的用戶名和密碼canal.instance.dbUsername=canalcanal.instance.dbPassword=canal#配置Canal將數據導入到Kafka topiccanal.mq.topic=canal_topic

復制

注意：兩臺機器上的instance目錄的名字需要保證完全一致，HA模式是依賴于instance name進行管理，同時必須都選擇default-instance.xml配置，此配置中才有關于zookeeper的設置信息。

3、啟動兩臺機器的Canal

#在node3上啟動Canal[root@node3 ~]# cd /software/canal/bin[root@node3 bin]# ./startup.sh#在node4上啟動Canal[root@node4 ~]# cd /software/canal/bin[root@node4 bin]# ./startup.sh

復制

啟動完成后，可以查看zookeeper中對應的路徑信息：

三、Canal HA 測試

默認搭建好的Canal HA 后可以通過查看Zookeeper中的“/otter/canal/destinations/examples/running”來查看Active的Canal節點：

測試Canal HA 如下：

1、向Mysql中“testdb.person”表中寫入數據

mysql> insert into person values (4,"s1",21),(5,"s2",22),(6,"s3",23);

復制

可以觀察到Kafka canal_topic中有監控到的數據如下：

{"data":[{"id":"4","name":"s1","age":"21"},{"id":"5","name":"s2","age":"22"},{"id":"6","name":"s3","age":"23"}],"database":"testdb","es":1618849974000,"id":2,"isDdl":false,"mysqlType":{"id":"int","name":"varchar(255)","age":"int"},"old":null,"pkNames":null,"sql":"","sqlType":{"id":4,"name":12,"age":4},"table":"person","ts":1618849975203,"type":"INSERT"}

復制

2、關閉active Canal Server節點，繼續向Mysql表中寫入數據

關閉node3 Canal Server：

[root@node3 ~]# cd /software/canal/bin[root@node3 bin]# ./stop.sh

復制

查看zookeeper “/otter/canal/destinations/examples/running”路徑Active的Canal節點：

繼續向MySQL中“testdb.person”表中寫入數據：

mysql> insert into person values (7,"x1",24),(8,"x2",25),(9,"x3",26);

復制

可以觀察寫入到Kafka “canal_topic”中數據如下：

{"data":[{"id":"7","name":"x1","age":"24"},{"id":"8","name":"x2","age":"25"},{"id":"9","name":"x3","age":"26"}],"database":"testdb","es":1618850233000,"id":2,"isDdl":false,"mysqlType":{"id":"int","name":"varchar(255)","age":"int"},"old":null,"pkNames":null,"sql":"","sqlType":{"id":4,"name":12,"age":4},"table":"person","ts":1618850234136,"type":"INSERT"}

復制

經過以上測試，Canal HA 生效。

注意：經過測試Canal HA 在使用zookeeper存儲binlog position時，當有一個Canal Server重新啟動并切換成Active節點時，每次都會重復讀取最后一條數據。使用非HA 本地存儲binlog position時，沒有此問題。