使用canal分析binlog(二) canal源码分析

时间:2022-02-17 10:17:54

在能够跑通example后有几个疑问

1. canal的server端对于已经读取的binlog,client已经ack的position,是否持久化,保存在哪里

2. 即使不启动zookeeper,canal也可以正常运行,canal使用zookeeper或者不使用有什么影响

从github上下载源码,https://github.com/alibaba/canal

我使用的版本是1.0.22,照着两位的博客看着源码学习一下,版本上有些出入,但了解思想和整体架构够了

博客-杨武兵-开源社区

canal DevGuide - agapple - ITeye技术网站

deployer项目为发布的server端程序,使用的日志系统为logback,默认日志不输出到控制台,为方便查看日志信息,将日志输出到控制台,修改logback.xml

 <logger name="com.alibaba.otter.canal.instance" additivity="false">
<level value="INFO" />
<appender-ref ref="STDOUT"/>
<appender-ref ref="CANAL-ROOT" />
</logger>
<logger name="com.alibaba.otter.canal.deployer" additivity="false">
<level value="INFO" />
<appender-ref ref="STDOUT"/>
<appender-ref ref="CANAL-ROOT" />
</logger>
<logger name="com.alibaba.otter.canal.meta.FileMixedMetaManager" additivity="false">
<level value="INFO" />
<appender-ref ref="STDOUT"/>
<appender-ref ref="CANAL-META" />
</logger> <root level="WARN">
<appender-ref ref="STDOUT"/>
<appender-ref ref="CANAL-ROOT" />
</root>

为canal创建mysql用户canal权限为 replication slave,replication client,canal使用这个mysql用户启动server

这时遇到问题:server启动报错:

com.alibaba.otter.canal.parse.exception.CanalParseException: parse row data failed.
Caused by: com.alibaba.otter.canal.parse.exception.CanalParseException: com.google.common.collect.ComputationException: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:`test`.`t_table`
Caused by: com.google.common.collect.ComputationException: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:`test`.`t_table`
at com.google.common.collect.MapMaker$ComputingMapAdapter.get(MapMaker.java:889) ~[guava-18.0.jar:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMeta(TableMetaCache.java:78) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.getTableMeta(LogEventConvert.java:677) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEvent(LogEventConvert.java:362) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parse(LogEventConvert.java:111) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parse(LogEventConvert.java:1) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser.parseAndProfilingIfNecessary(AbstractEventParser.java:327) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3$1.sink(AbstractEventParser.java:177) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.dump(MysqlConnection.java:129) [classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:210) [classes/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_73]
Caused by: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:`test`.`t_table`
Caused by: java.io.IOException: ErrorPacket [errorNumber=1142, fieldCount=-1, message=SELECT command denied to user 'canal'@'localhost' for table 't_terminal', sqlState=42000, sqlStateMarker=#]
with command: desc `test`.`t_table`
at com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor.query(MysqlQueryExecutor.java:60) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.query(MysqlConnection.java:73) [classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMeta0(TableMetaCache.java:105) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.access$0(TableMetaCache.java:104) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache$1.apply(TableMetaCache.java:51) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache$1.apply(TableMetaCache.java:1) ~[classes/:na]
at com.google.common.collect.ComputingConcurrentHashMap$ComputingValueReference.compute(ComputingConcurrentHashMap.java:356) ~[guava-18.0.jar:na]
at com.google.common.collect.ComputingConcurrentHashMap$ComputingSegment.compute(ComputingConcurrentHashMap.java:182) ~[guava-18.0.jar:na]
at com.google.common.collect.ComputingConcurrentHashMap$ComputingSegment.getOrCompute(ComputingConcurrentHashMap.java:151) ~[guava-18.0.jar:na]
at com.google.common.collect.ComputingConcurrentHashMap.getOrCompute(ComputingConcurrentHashMap.java:67) ~[guava-18.0.jar:na]
at com.google.common.collect.MapMaker$ComputingMapAdapter.get(MapMaker.java:885) ~[guava-18.0.jar:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMeta(TableMetaCache.java:78) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.getTableMeta(LogEventConvert.java:677) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEvent(LogEventConvert.java:362) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parse(LogEventConvert.java:111) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parse(LogEventConvert.java:1) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser.parseAndProfilingIfNecessary(AbstractEventParser.java:327) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3$1.sink(AbstractEventParser.java:177) ~[classes/:na]
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.dump(MysqlConnection.java:129) [classes/:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:210) [classes/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_73]

查看官方说明,示例中为canal配置的mysql用户还需要select权限。

汗,我是按照mysql slave配的权限,原以为不需要select。。。仔细一想,至少通过select方式判断心跳肯定需要select权限啊,不过我貌似没有开启心跳啊。 为mysql用户canal增加了select权限后,server可以正常运行。

既然都报错了,就看看canal到底select了些什么。

查看报错信息,异常发生的最近的位置:MapMaker.java:889,是第三方代码,MapMaker是一个功能强大的Map的实现,示例查看这里 。估计是在使用这个工具的时出现的异常,内部有调用回调函数之类的。

下一行报错信息 TableMetaCache.getTableMeta(TableMetaCache.java:78),查看代码

    public TableMeta getTableMeta(String schema, String table, boolean useCache) {
if (!useCache) {
tableMetaCache.remove(getFullName(schema, table));
} return tableMetaCache.get(getFullName(schema, table));  // 78行
}
查看tableMetaCache的声明,及初始化
/**
* 处理table meta解析和缓存
*
* @author jianghang 2013-1-17 下午10:15:16
* @version 1.0.0
*/
public class TableMetaCache { // 第一层tableId,第二层schema.table,解决tableId重复,对应多张表
private Map<String, TableMeta> tableMetaCache; public TableMetaCache(MysqlConnection con){
this.connection = con;
tableMetaCache = MigrateMap.makeComputingMap(new Function<String, TableMeta>() { public TableMeta apply(String name) {
try {
return getTableMeta0(name);
} catch (IOException e) {
// 尝试做一次retry操作
try {
connection.reconnect();
return getTableMeta0(name);
} catch (IOException e1) {
throw new CanalParseException("fetch failed by table meta:" + name, e1);
}
}
} });
.... // 省略
} }

当第78行调用tableMetaCache.get方法时,参数为空,就会去执行回调函数 apply();

查看在apply方法中调用的getTableMeta0();

 private TableMeta getTableMeta0(String fullname) throws IOException {
ResultSetPacket packet = connection.query("desc " + fullname);
return new TableMeta(fullname, parserTableMeta(packet));
}

执行select的就是这里,通过断点看到出错时执行的查询是 desc test.t_table。

那么是什么时候需要获取表结构信息呢,通过报错信息继续向上找LogEventConvert类的parse方法,源码太长就不贴了,LogEventConvert类的功能是是将LogEvent转化为Entry对象,对着注释还是能看的。可以确定的是在MysqlConnection在dump到binlog数据后作处理时去查询的表结构。

select的问题告一段落,回到正题。canal的工作原理可参考这里。经过测试,同步数据是会保存到 config/{destination}/meta.dat文件中( 运行deployer项目时文件保存位置为canal/config/{destination}/meta.dat )。文件内容为json格式,如下所示

{
"clientDatas": [
{
"clientIdentity": {
"clientId": 1001,
"destination": "example",
"filter": ".*\\..*"
},
"cursor": {
"identity": {
"slaveId": -1,
"sourceAddress": {
"address": "127.0.0.1",
"port": 3306
}
},
"postion": {
"included": false,
"journalName": "mysql-log.000002",
"position": 10329,
"serverId": 1,
"timestamp": 1480670799000
}
}
}
],
"destination": "example"
}

在不使用zookeeper,非HA模式下,我做了如下几个测试:

1. 删除meta.dat文件->启动server->启动client->操作mysql

测试结果:在操作mysql后创建了meta.dat文件

2. 删除meta.dat文件->操作mysql->启动server->启动client->操作mysql

测试结果:启动client时未创建meta.dat文件,第二次操作mysql创建了meta.dat文件

3. 不删除meta.dat文件->操作mysql—>启动server->启动client

测试结果:启动client后client获取到mysql的log数据

根据测试结果,大致猜测一下处理过程

1. canal server端启动时去查看meta.dat文件,如果存在则加载该destination的位置信息到内存中,根据内存中destination的journalName和position从mysql master获取binlog数据;如果不存在meta.dat文件,则获取最新发生的binlog数据,并保存在内存中(中继日志)。

2. 启动canal client,client端从server端获取binlog数据,如果之前client端有过成功的ACK,server端会记录到内存中并保存到meta.dat文件,client获取binlog数据会从上次ACK成功位置开始;如果client之前没有过ACK,则client获取binlog数据会从server保存在内存中(中继日志)开始。

接下来通过查阅源码去证实猜测的是否正确。

......

通过打断点和读源码 ,发现保存文件到本地是通过 FileMixedMetaManager类实现。