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

时间  2019-11-10
标签 使用 canal 分析 binlog 源码 繁體版
原文   原文链接

在可以跑通example后有几个疑问html

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

2. 即便不启动zookeeper，canal也能够正常运行，canal使用zookeeper或者不使用有什么影响mysql

 

 从github上下载源码，https://github.com/alibaba/canalgit

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

博客-杨武兵-开源社区sql

canal DevGuide - agapple - ITeye技术网站json

 

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启动报错：app

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

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程