大数据Canal(四):Canal HA原理及安装

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 保证步骤如下:

  1. canal server要启动某个canal instance时都先向zookeeper_进行一次尝试启动判断。
  2. 创建zookeeper节点成功后,对应的canal server就启动对应的canal instance,没有创建成功的canal instance就会处于standby状态。
  3. 一旦zookeeper发现canal server A创建的instance节点消失后,立即通知其他的canal server再次进行步骤1的操作,重新选出一个canal server启动instance。
  4. 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=canal
canal.instance.dbPassword=canal

#配置Canal将数据导入到Kafka topic
canal.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时,没有此问题。

正文完