canal的是如何HA机制
什么是HA机制
所谓HA（High Available），即高可用（7*24小时不中断服务）。
通过配置Active/Standby两个NameNodes实现在集群中对NameNode的热备来解决上述问题。
- 有一台节点是Active模式，也就是工作模式，其它的节点是 Standby（备用模式）；
- 干活的（Active模式的节点）如果挂了，就从备用模式的节点中选出一台顶上去。
基于zookeeper
通过watcher和EPHEMERAL
1.watcher节点，客户端像zookeeper提供目录订阅，当目录发生改变，通知所有订阅的客户端
2.EPHEMERAL 临时节点，临时节点只存在创建节点的客户端建立连接的过程中。连接释放自动移除
1.canal server 要启动某个 canal instance 时都先向 zookeeper 进行一次尝试启动判断 (实现：创建 EPHEMERAL 节点，谁创建成功就将谁加入running）；
2.创建 zookeeper 节点成功后，对应的 canal server 就启动对应的 canal instance，没有创建成功的 canal instance 就会处于 standby 状态；
3.一旦 zookeeper 发现 canal server A 创建的节点消失后，立即通知其他的 canal server 再次进行步骤1的操作，重新选出一个 canal server 启动instance；
4.canal client 每次进行connect时，会首先向 zookeeper 询问当前是谁启动了canal instance，然后和其建立链接，一旦链接不可用，会重新尝试connect。
注意
为了减少对mysql dump的请求，不同server上的instance要求同一时间只能有一个处于running，其他的处于standby状态。
canal client实现流程
- canal client 的方式和 canal server 方式类似，也是利用 zookeeper 的抢占EPHEMERAL 节点的方式进行控制
- 为了保证有序性，一份 instance 同一时间只能由一个 canal client 进行get/ack/rollback操作，否则客户端接收无法保证有序。
搭建步骤
配置文件修改
1.修改/conf/canal.properties
1.1修改各个server的这2个端口保证不重复冲突
canal.port = 11111
canal.metrics.pull.port = 11112
1.2.注释默认配置改为HA(高可用的配置)
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml #HA模式必须使用该xml，需要将相关数据写入zookeeper,保证数据集群共享 canal.instance.global.spring.xml = classpath:spring/default-instance.xml
2.修改/conf/example/instance.properties
2.1配置zookeeper地址
#zookeeper集群地址 多个,号隔开
canal.zkServers =192.168.20.4:2181
2.2.修改保证各个server不冲突
canal.instance.mysql.slaveId=1
启动Server
我编辑了2个启动脚本
allStart.sh
cd canal.deployer-1.1.4-cluster1/bin ./stop.sh ./startup.sh cd ../../ cd canal.deployer-1.1.4-cluster2/bin ./stop.sh ./startup.sh
allStop.sh
cd canal.deployer-1.1.4-cluster1/bin ./stop.sh cd ../../ cd canal.deployer-1.1.4-cluster2/bin ./stop.sh
canalZookeeper目录
ls /otter/canal:canal的根目录
ls /otter/canal/cluster:整个canal server的集群列表
ls /otter/canal/destinations:destination的根目录
get /otter/canal/destinations/{客户端订阅的destination}/running:服务端当前正在提供服务的running节点
ls /otter/canal/destinations/{客户端订阅的destination}/cluster:针对某个destination的工作集群列表
get /otter/canal/destinations/{客户端订阅的destination}/{clientid}/running:客户端当前正在读取的running节点
get /otter/canal/destinations/{客户端订阅的destination/{clientId}/running 针对某个destination当前消费的客户端
get /otter/canal/destinations/{客户端订阅的destination}/{clientId}/cursor:客户端读取的position信息
注意:因为一个canal-server的各个destination只能有一个客户端订阅 所以clientId当前版本写死的是1001
单机非zookeeper存储格式
{"clientDatas":[{"clientIdentity":{"clientId":,"destination":"example","filter":"merge_test.pro_brand"},"cursor":{"identity":{"slaveId":-1,"sourceAddress":{"address":"rm-2zeqc826391dt0lk2.mysql.rds.aliyuncs.com","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mysql-bin.005121","position":72729280,"serverId":2539862954,"timestamp":1618291064000}}}],"destination":"example"}
zookeeper查看
1.连接zookeeper
./zkCli.sh -server 192.168.20.4:2181
2.查看可用节点
[zk: 192.168.20.4:2181(CONNECTED) 14] ls /otter/canal/cluster
[192.168.2.1:11111, 192.168.2.1:11113]
3.查看当前active节点
[zk: 192.168.20.4:2181(CONNECTED) 9] get /otter/canal/destinations/example/running {"active":true,"address":"192.168.2.1:11113"}
4.停掉11113这个server
可用节点也只有11111
[zk: 192.168.20.4:2181(CONNECTED) 23] ls /otter/canal/cluster
[192.168.2.1:11111]
可以发现running变成了11111这个节点
[zk: 192.168.20.4:2181(CONNECTED) 22] get /otter/canal/destinations/example/running {"active":true,"address":"192.168.2.1:11111"}
当然客户端也正常消费