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时，没有此问题。