使用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类实现。

相关文章