通过zk来保证集群中的canal实例只会有一个实例来工作，并将binlog解析进度保存在zk集群中，从而保证集群中某个实例出现故障时，及时切换为其他的实例，但是并不是实时保存而是延迟一秒保存，因此会出现binlog在主备切换时出现binlog重复解析的问题。

修改canal配置文件

修改2个配置

  1. [root@master1 canal]# vi canal/conf/canal.properties
  2. # 用逗号隔开且不留空格 
  3. canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
  4. canal.instance.global.spring.xml = classpath:spring/default-instance.xml

拷贝canal到另一台机器并解压缩(作为从节点)

  1. [root@master2 ~]# tar -zxvf canal.deployer-1.1.5.tar.gz -C /root/canal

修改slave配置文件

  1. [root@master2 canal]# vi /root/canal/conf/example/instance.properties
  2. ## mysql serverId修改与master不同即可
  3. canal.instance.mysql.slaveId = 1235
  4. # position info修改成mysql地址即可
  5. canal.instance.master.address = 10.105.10.121:3306
  7. [root@master2 canal]# vi /root/canal/conf/canal.properties
  8. # 用逗号隔开且不留空格 
  9. canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
  10. canal.instance.global.spring.xml = classpath:spring/default-instance.xml

两台机器分别启动canal

  1. [root@master2 canal]#  sh /root/canal/bin/startup.sh

查看canal在zk中的状态

  1. [zk: localhost:2181(CONNECTED) 11] get /otter/canal/destinations/example/running
  2. {"active":true,"address":"10.105.10.121:11111","cid":1}
  3. cZxid = 0x250000f037
  4. ctime = Thu Apr 13 13:42:04 CST 2017
  5. mZxid = 0x250000f037
  6. mtime = Thu Apr 13 13:42:04 CST 2017
  7. pZxid = 0x250000f037
  8. cversion = 0
  9. dataVersion = 0
  10. aclVersion = 0
  11. ephemeralOwner = 0x15b5b11e9cd16e1
  12. dataLength = 55
  13. numChildren = 0

10.105.10.121为active，所以10.105.10.122位stangby节点

测试HA是否配置成功

stop掉10.105.10.121这台机器

  1. [root@master1 canal]# sh /root/canal/bin/stop.sh
  2. master1: stopping canal 6107 ... 
  3. Oook! cost:1
  1. [zk: localhost:2181(CONNECTED) 28] get /otter/canal/destinations/example/running
  2. {"active":true,"address":"10.105.10.122:11111","cid":1}
  3. cZxid = 0x250000f0a7
  4. ctime = Thu Apr 13 13:46:10 CST 2017
  5. mZxid = 0x250000f0a7
  6. mtime = Thu Apr 13 13:46:10 CST 2017
  7. pZxid = 0x250000f0a7
  8. cversion = 0
  9. dataVersion = 0
  10. aclVersion = 0
  11. ephemeralOwner = 0x35b5b11e9d816cb
  12. dataLength = 55

很明显active已经切换到10.105.10.122上了

客户端代码

  1. CanalConnector connector = CanalConnectors.newClusterConnector("10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181","example", "", "");