canal(基于mysql数据库binlog的增量订阅&
2017-04-13 本文已影响10063人
献给记性不好的自己
一、背景
早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。
ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)
基于日志增量订阅&消费支持的业务：
数据库镜像
数据库实时备份
多级索引 (卖家和买家各自分库索引)
search build
业务cache刷新
价格变化等重要业务消息
二、项目介绍
名称：canal [kə'næl]
译意： 水道/管道/沟渠
语言： 纯java开发
定位： 基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了mysql
三、工作原理
1、mysql主备复制实现
Paste_Image.png
2、pom.xml增加依赖
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.0.12</version>
</dependency>
3、编写ClientSample类订阅binlog
>package com.chainfin.canal;
>
>import java.net.InetSocketAddress;
>import java.util.List;
>
>import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
>
/**
*
* @author jinxiaoxin
*
*/
public class ClientSample {
public static void main(String[] args) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("10.105.10.121", 11111), "example", "",
"");// AddressUtils.getHostIp(),
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
connector.subscribe("test\\..*");// .*代表database，..*代表table
connector.rollback();//
int totalEmptyCount = 120;
while (emptyCount < totalEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
// System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n",
// batchId, size);
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
>
private static void printEntry(List<Entry> entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
|| entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
RowChange rowChage = null;
try {
rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException(
"ERROR ## parser of eromanga-event has an error,data:"
+ entry.toString(), e);
}
EventType eventType = rowChage.getEventType();
System.out
.println(String
.format("================> binlog[%s:%s] ,name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry
.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry
.getHeader().getTableName(),
eventType));
for (RowData rowData : rowChage.getRowDatasList()) {
if (eventType == EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
>
private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue()
+ " update=" + column.getUpdated());
}
}
}
>
4、MYSQL进行增删改查
mysql> CREATE TABLE `xdual` (
-> `ID` int(11) NOT NULL AUTO_INCREMENT,
-> `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
-> PRIMARY KEY (`ID`)
-> ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;
Query OK, 0 rows affected (0.01 sec)
mysql> insert into xdual(id,x) values(1,now());
Query OK, 1 row affected (0.01 sec)
5、eclipse控制台可以看到如下输出
Paste_Image.png
六、canal集群搭建
1、需要安装zookeeper
自行了解安装及其使用，此处略过
2、修改canal配置文件
修改两个配置
[root@master1 canal]# vi /root/canal/conf/canal.properties
# 用逗号隔开且不留空格
canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
3、拷贝canal到另一台机器并解压缩(作为从节点)
[root@master1 canal]# scp canal.deployer-1.0.23.tar.gz root@10.105.10.122:/root
canal.deployer-1.0.23.tar.gz 100% 9586KB 9.4MB/s 00:00
[root@master2 ~]# mkdir canal
[root@master2 ~]# tar -zxvf canal.deployer-1.0.23.tar.gz -C /root/canal
4、修改slave配置文件
[root@master2 canal]# vi /root/canal/conf/example/instance.properties
## mysql serverId修改与master不同即可
canal.instance.mysql.slaveId = 1235
# position info修改成mysql地址即可
canal.instance.master.address = 10.105.10.121:3306
[root@master2 canal]# vi /root/canal/conf/canal.properties
# 用逗号隔开且不留空格
canal.zkServers=10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
5、两台机器分别启动canal
[root@master2 canal]# sh /root/canal/bin/startup.sh
6、查看canal在zk中的状态
[zk: localhost:2181(CONNECTED) 11] get /otter/canal/destinations/example/running
{"active":true,"address":"10.105.10.121:11111","cid":1}
cZxid = 0x250000f037
ctime = Thu Apr 13 13:42:04 CST 2017
mZxid = 0x250000f037
mtime = Thu Apr 13 13:42:04 CST 2017
pZxid = 0x250000f037
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15b5b11e9cd16e1
dataLength = 55
numChildren = 0
10.105.10.121为active，所以10.105.10.122位stangby节点
7、测试HA是否配置成功
stop掉10.105.10.121这台机器
[root@master1 canal]# sh /root/canal/bin/stop.sh
master1: stopping canal 6107 ...
Oook! cost:1
[zk: localhost:2181(CONNECTED) 28] get /otter/canal/destinations/example/running
{"active":true,"address":"10.105.10.122:11111","cid":1}
cZxid = 0x250000f0a7
ctime = Thu Apr 13 13:46:10 CST 2017
mZxid = 0x250000f0a7
mtime = Thu Apr 13 13:46:10 CST 2017
pZxid = 0x250000f0a7
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x35b5b11e9d816cb
dataLength = 55
很明显active已经切换到10.105.10.122上了
8、客户端代码
//创建connector 为集群connector
CanalConnector connector = CanalConnectors.newClusterConnector("10.105.10.123:2181,10.105.10.124:2181,10.105.10.125:2181","example", "", "");
启动代码
Paste_Image.png
9、错误解决
当这个配置的ip不配成127.0.0.1时(也就是本地)
canal.instance.master.address = 10.105.10.121:3306
查看canal配置文件，会报如下错误导致，无法读取binlog
[root@master1 example]# tail -f example.log
2017-04-13 14:29:23.646 [destination = example , address = master1/10.105.10.121:3306 , EventParser] ERROR c.a.o.canal.parse.inbound.mysql.dbsync.DirectLogFetcher - I/O error while reading from client socket
java.io.IOException: Received error packet: errno = 1236, sqlstate = HY000 errmsg = Could not find first log file name in binary log index file
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.fetch(DirectLogFetcher.java:95) ~[canal.parse-1.0.23.jar:na]
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.dump(MysqlConnection.java:121) [canal.parse-1.0.23.jar:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:209) [canal.parse-1.0.23.jar:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]
解决办法：
登录mysql
mysql> show variables like "%sum%";
+---------------------------+-------+
| Variable_name | Value |
+---------------------------+-------+
| binlog_checksum | CRC32 |
| innodb_checksum_algorithm | crc32 |
| innodb_checksums | ON |
| innodb_log_checksums | ON |
| master_verify_checksum | OFF |
| slave_sql_verify_checksum | ON |
+---------------------------+-------+
6 rows in set (0.00 sec)
查询资料发现mysql版本为5.6时，
这个错误一般出现在master5.6,slave在低版本的情况下。这是由于5.6使用了crc32做binlog的checksum；
当一个event被写入binary log(二进制日志)的时候，checksum也同时写入binary log，然后在event通过网络传输到从服务器(slave)之后，再在从服务器中对其进行验证并写入从服务器的relay log.
由于每一步都记录了event和checksum，所以我们可以很快地找出问题所在。
在master1中设置binlog_checksum =none；
mysql> set global binlog_checksum='NONE';
Query OK, 0 rows affected (0.00 sec)
mysql> show variables like "%sum%";
+---------------------------+-------+
| Variable_name | Value |
+---------------------------+-------+
| binlog_checksum | NONE |
| innodb_checksum_algorithm | crc32 |
| innodb_checksums | ON |
| innodb_log_checksums | ON |
| master_verify_checksum | OFF |
| slave_sql_verify_checksum | ON |
+---------------------------+-------+
6 rows in set (0.01 sec)
这样错误就可以解决了
七、properties配置文件参数解释
properties配置分为两部分：
canal.properties (系统根配置文件)
instance.properties (instance级别的配置文件，每个instance一份)
1、canal.properties
canal.destinations #当前server上部署的instance列表
canal.conf.dir #conf/目录所在的路径
canal.auto.scan #开启instance自动扫描
#如果配置为true，canal.conf.dir目录下的instance配置变化会自动触发：
#a. instance目录新增： 触发instance配置载入，lazy为true时则自动启动
#b. instance目录删除：卸载对应instance配置，如已启动则进行关闭
#c. instance.properties文件变化：reload instance配置，如已启动自动进行重启操作
canal.auto.scan.interval #instance自动扫描的间隔时间，单位秒
canal.instance.global.mode #全局配置加载方式
canal.instance.global.lazy #全局lazy模式
canal.instance.global.manager.address #全局的manager配置方式的链接信息 无
canal.instance.global.spring.xml #全局的spring配置方式的组件文件
canal.instance.example.mode
canal.instance.example.lazy
canal.instance.example.spring.xml
#instance级别的配置定义，如有配置，会自动覆盖全局配置定义模式
#命名规则：canal.instance.{name}.xxx 无
2、instance.properties
canal.id #每个canal server实例的唯一标识，暂无实际意义
canal.ip #canal server绑定的本地IP信息，如果不配置，默认选择一个本机IP进行启动服务
canal.port #canal server提供socket服务的端口
canal.zkServers #canal server链接zookeeper集群的链接信息
#例子：10.20.144.22:2181,10.20.144.51:2181
canal.zookeeper.flush.period #canal持久化数据到zookeeper上的更新频率，单位毫秒
canal.instance.memory.batch.mode #canal内存store中数据缓存模式
#1. ITEMSIZE : 根据buffer.size进行限制，只限制记录的数量
#2. MEMSIZE : 根据buffer.size * buffer.memunit的大小，限制缓存记录的大小
canal.instance.memory.buffer.size #canal内存store中可缓存buffer记录数，需要为2的指数
canal.instance.memory.buffer.memunit #内存记录的单位大小，默认1KB，和buffer.size组合决定最终的内存使用大小
canal.instance.transactionn.size 最大事务完整解析的长度支持
超过该长度后，一个事务可能会被拆分成多次提交到canal store中，无法保证事务的完整可见性
canal.instance.fallbackIntervalInSeconds #canal发生mysql切换时，在新的mysql库上查找binlog时需要往前查找的时间，单位秒
#说明：mysql主备库可能存在解析延迟或者时钟不统一，需要回退一段时间，保证数据不丢
canal.instance.detecting.enable #是否开启心跳检查
canal.instance.detecting.sql #心跳检查sql insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.interval.time #心跳检查频率，单位秒
canal.instance.detecting.retry.threshold #心跳检查失败重试次数
canal.instance.detecting.heartbeatHaEnable #心跳检查失败后，是否开启自动mysql自动切换
#说明：比如心跳检查失败超过阀值后，如果该配置为true，canal就会自动链到mysql备库获取binlog数据
canal.instance.network.receiveBufferSize #网络链接参数，SocketOptions.SO_RCVBUF
canal.instance.network.sendBufferSize #网络链接参数，SocketOptions.SO_SNDBUF
canal.instance.network.soTimeout #网络链接参数，SocketOptions.SO_TIMEOUT
