Canal使用手册

Posted on | In

1.Canal同步介绍

项目地址：

https://github.com/alibaba/canal

组件图：

组件图

2.MySQL端配置

canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能，建议配置binlog模式为row.

1
2
3
4
[mysqld]
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限.针对已有的账户可直接通过grant

1
2
3
4
CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

3.Canal Server端配置

canal server端为解析bin-log日志，并将解析结果格式化发给canal client端，server端不需要修改，只需要部署配置即可。下载 canal.deployer-1.0.17.tar.gz包.

instance配置，打开conf/example/instance.properties配置文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
## mysql serverId
canal.instance.mysql.slaveId = 1234
#position info，需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =


#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =


#username/password，需要改成自己的数据库信息
canal.instance.dbUsername = canal

canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8


#table regex
canal.instance.filter.regex = .\..

说明：

canal.instance.connectionCharset 代表数据库的编码方式对应到java中的编码类型，比如UTF-8，GBK , ISO-8859-1
如果系统是1个cpu，需要将canal.instance.parser.parallel设置为false.

canal server配置：打开conf/canal.properties配置文件：

1
2
3
4
5
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =

启动server

1
sh bin/startup.sh

关闭server

1
sh bin/stop.sh

日志查看：

1
2
3
tail -f logs/canal/canal.log

tail -f logs/example/example.log

4.Canal Client端

canal client端为使用数据同步者自己开发的程序，在client中需要针对server端发送过来的不同日志类型进行数据的处理。开发步骤如下:

1.添加项目依赖

1
2
3
4
5
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.0</version>
</dependency>

2.Client代码开发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                                                                                        11111), "example", "", "");
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();

//获取服务端消息
int batchSize = 1000;
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据

//获取消息内容
long batchId = message.getId();
int size = message.getEntries().size();
List<Entry> contexts = message.getEntries();

//处理数据类型
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
    continue;
}
RowChange rowChage =RowChange.parseFrom(entry.getStoreValue());
EventType eventType = rowChage.getEventType();

//解析类型
for (RowData rowData : rowChage.getRowDatasList()) {
    if (eventType == EventType.DELETE) {
        printColumn(rowData.getBeforeColumnsList());
    } else if (eventType == EventType.INSERT) {
        printColumn(rowData.getAfterColumnsList());
    } else {
        System.out.println("-------&gt; before");
        printColumn(rowData.getBeforeColumnsList());
        System.out.println("-------&gt; after");
        printColumn(rowData.getAfterColumnsList());
    }
}

//显示变更内容
for (Column column : columns) {
        System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
}

5.Canal适配器

canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能:

客户端启动器
同步管理REST接口
日志适配器, 作为DEMO
关系型数据库的数据同步(表对表同步), ETL功能
HBase的数据同步(表对表同步), ETL功能
(后续支持) ElasticSearch多表数据同步,ETL功能

具体使用方法见：https://github.com/alibaba/canal/wiki/ClientAdapter

6.数据同步中间件比较

canal,otter,yugong,datax比较：http://www.importnew.com/22294.html