canal集群部署与java接入 - my201110lc的博客 - CSDN博客 - 关键信息 CanalSample.printEntry(message.getEntries()); 没了
Canal应用
文档总体上分5部分
1. 应用场景；
2. 如何部署环境；
3. 应用端如何使用；
4. 踩坑；
5. 稳定性和实时性；
应用场景：canal的原理是伪装为rds的一个slave来监听binlog；所以canal可以用在：
1. 数据同步，比如：做在/离线数据库之间的数据同步操作；
2. 数据消费，比如：需要根据关注的数据库表的变化，做搜索增量；
3. 数据脱敏，比如：需要将线上动态数据导入到其它地方，做数据脱敏；
如何部署：
配置：
集群最低配置zookeeper X3 + canal X2，
单台机器：4C8G
增加机器数量可以提高稳定性，但是对消费速度不影响；
步骤：
1. 先搭个zk环境；zookeeper\zookeeper-3.4.6.tar.gz
2. 在三台机器上下载zookeeper安装包，3.4.6或3.4.9是稳定版本；
3. 三台机器解压安装包
4. 进入conf中复制zoo_sample.cfg 到 zoo.cfg
5. 配置zoo.cfg，
tickTime：leader-follower通信心跳时间，也就是每个 tickTime 时间就会发送一个心跳，单位ms；
initLimit：Leader-Follower初始通信时限，初始连接时能容忍的最多心跳数（tickTime的数量），比如这里 initLimit=10就是10次心跳（2000ms X 10）无应答，则认为连接失败；
syncLimit：Leader-Follower同步通信时限，请求和应答之间能容忍的最多心跳数，比如这里syncLimit=5 就是5次心跳（2000ms X 10）无应答，则认为同步失败；
dataDir：数据文件目录，默认情况下保存数据和日志；可以把myid也放这个目录下；
clientPort：zookeeper客户端的连接端口；
server.1,server.2,server.3：这里的编号1，2，3对应集群中机器的myid设的数值，必须一一对应；
2888端口号是zookeeper服务之间通信的端口。
3888端口是zookeeper与其他应用程序通信的端口。
6. 在配置好的dataDir下面新建myid，里面配置server对应的数字编号；
7. 三台机器都配置好后，启动./bin/zkServer.sh start
附常用命令：
停机：zkServer.sh stop，
查看状态：zkServer.sh status，
进入zk客户端：./bin/zkCli.sh -server 127.0.0.1:2181(如果默认端口可直接进入./bin/zkCli.sh)
手动操作节点:
查看子目录: ls /node;
查看节点内容:get /node；
删除根节点: delete /node；
递归删除节点及其子节点：rmr /node；
创建节点：create /node2 “”;
8. 如果连接客户端抛异常:Unable to read additional data from server sessionid xxx，则检查三台机器配置是否完成，myid是否对应server后面的编号，是否都已成功启动；
接下来开始搭建canal环境：
前提：rds已经打开binlog功能，并配置binlog模式为row
1. 下载最新canal安装包https://github.com/alibaba/canal/releases
2. 解压后进入conf目录：cd canal/conf
3. 把example复制出来，命名为自己连rds的一个实例：cp -r example dev_db；
4. 进入dev_db改配置，如下：
slaveId：保证canal集群中slaveId不同即可；
address：rds连接地址；
dbUsername和dbPassword：需要在rds上有repication权限；
defaultDatabaseName：配置默认监听的数据库
5. 改conf/canal.properties
canal.id：保证每台canal机器id不同；
canal.port：客户端访问端口；
canal.zkServers：zk集群地址；
canal.instance.global.spring.xml：集群模式下配置为上图；
其它默认配置即可；
6. 启动：./bin/startup.sh；停止：./bin/stop.sh；
7. 启动后，可以在zk集群节点上查看canal连接状态；
8. 查看偏移位点；
问题排查：
1.位点不匹配
ERROR c.a.otter.canal.parse.inbound.mysql.MysqlEventParser - dump address /xxxxxx:3306 has an error, retrying. caused by
com.alibaba.otter.canal.parse.exception.CanalParseException: can't find start position for example
原因分析：canal异常停止或者rds做过切换，导致的canal消费rds的binlog位点不一致；
解决：
单机版删除conf/dev_db实例下的meta.dat；
集群版删除zk上节点：/otter/canal/destinations/test_db/1001/cursor
重启canal，恢复正常；
2.账号权限和rds配置问题
[EventParser] ERROR com.alibaba.otter.canal.common.alarm.LogAlarmHandler - destination:example[com.alibaba.otter.canal.parse.exception.CanalParseException: command : 'show master status' has an error!
解决：再次确认rds，binlog是否正常，canal实例配置的账号是否有repication权限，若权限恢复，重启canal解决问题；
稳定性：
在两台canal都接入到zookeeper集群的前提下，停掉一台canal，zk会自动切换到另外一个canal继续处理binlog，因为位点都是保存在zk节点上，所以另一台canal也可以继续这个位点消费；
实时性：
可以在接收端，设置每次取数据的间隔时间，数据大小；按业务场景确定数值；
效率：
经测，在2C4G配置，测试环境网络带宽，消息接收速度大于2000条/秒；
下面是java应用端接入：
1. pom引入
<!-- canal -->
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.0.24</version>
</dependency>
2. Demo接入
private static final Logger LOGGER = LoggerFactory._getLogger_(CanalSample.class);
private static final int _CANAL_BATCH_SIZE_ = 1000;
public static void main(String args[]) {
String zkHost = "127.0.0.1:2181,127.0.0.1:2181,127.0.0.1:2181";
String confName = "dev_db";
String canalUsername = "";
String canalPassword = "";
// 每次数据的偏移量
long batchId = 0;
// 创建链接
CanalConnector connector = CanalConnectors._newClusterConnector_(zkHost, confName, canalUsername, canalPassword);
// 外层死循环：
// 在canal节点宕机后，抛出异常，
// 等待zookeeper对canal处理机的切换，
// 切换完后，继续创建连接处理数据
while (true) {
try {
connector.connect();
connector.subscribe(".\..");
connector.rollback();
// 内层死循环:
// 按频率实时监听数据变化，
// 一旦收到变化数据，立即做消费处理，并ack
// 考虑消费速度，可以考虑异步处理，并ack
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(CANAL_BATCH_SIZE);
batchId = message.getId();
int size = message.getEntries().size();
// 偏移量不等于-1 或者 获取的数据条数不为0 时，认为拿到消息，并处理
if (batchId == -1 || size == 0) {
// 200ms 拉一次变动数据
Thread._sleep_(200);
connector.ack(batchId); // 提交确认
} else {
CanalSample._printEntry_(message.getEntries());
connector.ack(batchId); // 提交确认
}
}
} catch (Exception e) {
LOGGER.error("read canal message error , exception : ", e);
// 处理失败, 按偏移量回滚数据
connector.rollback(batchId);
} finally {
// 关闭连接
connector.disconnect();
}
}
}
Original url: Access
Created at: 2019-09-10 18:35:51
Category: default
Tags: none
未标明原创文章均为采集，版权归作者所有，转载无需和我联系，请注明原出处，南摩阿彌陀佛，知识，不只知道，要得到
最新评论