基于docker的canal-server部署作业
1. 目标
启动一个canal-server实例监听目标mysql的操作
2. 准备工作
- Ubuntu下安装Docker，并配置国内镜像
- 部署一个mysql实例
- mysql实例也可以使用docker进行启动(参考菜鸟教程：Docker 安装 MySQL)
$ docker run -p 3306:3306 --name mysql-master -e MYSQL_ROOT_PASSWORD=123456 -d mysql - 配置mysql (参见canal Quickstart)
- 启用binlog
[mysqld] log-bin=mysql-bin binlog-format=ROW #选择row模式 server_id=9527 #配置mysql replaction需要定义，不能和canal的slaveId重复- 添加slave权限
在canal Quickstart文档中，并没有要求添加SHOW VIEW权限，但是本作业在执行中会产生如下异常：CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, SHOW VIEW, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
因为canal-server在启动时，会去dump表结构并缓存。Caused by: java.io.IOException: ErrorPacket [errorNumber=1142, fieldCount=-1, message=SHOW VIEW command denied to user 'canal'@'172.17.0.1' for table 'host_summary', sqlState=42000, sqlStateMarker=#] with command: show create table `sys`.`host_summary`; ...
所以首次启动成功后，再将 SHOW VIEW 权限回收并不会影响正常作业（表结果不变的情况下）。这也为我在作业中重现异常带来了一点小麻烦。为了快速跑通QuickStart，遇到问题后我第一时间给canal账号赋予了所有权限：/*com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta*/ /** * 初始化的时候dump一下表结构 */ private boolean dumpTableMeta(MysqlConnection connection, final CanalEventFilter filter) { try { ResultSetPacket packet = connection.query("show databases"); List<String> schemas = new ArrayList<String>(); for (String schema : packet.getFieldValues()) { schemas.add(schema); } for (String schema : schemas) { packet = connection.query("show tables from `" + schema + "`"); List<String> tables = new ArrayList<String>(); for (String table : packet.getFieldValues()) { String fullName = schema + "." + table; if (blackFilter == null || !blackFilter.filter(fullName)) { if (filter == null || filter.filter(fullName)) { tables.add(table); } } } if (tables.isEmpty()) { continue; } StringBuilder sql = new StringBuilder(); for (String table : tables) { sql.append("show create table `" + schema + "`.`" + table + "`;"); } List<ResultSetPacket> packets = connection.queryMulti(sql.toString()); for (ResultSetPacket onePacket : packets) { if (onePacket.getFieldValues().size() > 1) { String oneTableCreateSql = onePacket.getFieldValues().get(1); memoryTableMeta.apply(INIT_POSITION, schema, oneTableCreateSql, null); } } } return true; } catch (IOException e) { throw new CanalParseException(e); } }
然后重启canal-server，作业正常。然后回收权限，但异常并不能重现，但canal-server工作正常。GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES;REVOKE ALL PRIVILEGES ON *.* FROM 'canal'@'%' ; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
3. 部署canal-server
官网文档 Canal Docker QuickStart。 文档中使用canal提供的run.sh启动，脚本的工作就是为执行docker run准备参数，在本作业的目标是启动一个canal-server实例，大多参数都使用canal默认配置就可以，所以为了简单，直接使用docker命令进行启动。配置相应的master地址和用户名密码，映射必要的监听端口就足够了。详细配置可参考Canal AdminGuide。
docker run --name canal-server \
-e canal.instance.master.address=192.168.83.128:3306 \
-e canal.instance.dbUsername=canal \
-e canal.instance.dbPassword=canal \
-p 11111:11111 \
-d canal/canal-server:v1.1.0
canal-server:v1.1.0有一个小bug，如果canal.properties中没有配置canal.instance.parser.parallelThreadSize(默认情况，该配置项是被注释掉的)：
#canal.instance.parser.parallelThreadSize = 16
那么，MysqlMultiStageCoprocessor的parserThreadCount会被设为0，从而在启动线程池的时候会抛IllegalArgumentException异常。
/*com.alibaba.otter.canal.parse.inbound.AbstractEventParser*/
protected Integer parallelThreadSize = Runtime.getRuntime().availableProcessors() * 60 / 100; //60%的能力跑解析,剩余部分处理网络
public void setParallelThreadSize(Integer parallelThreadSize) {
if (parallelThreadSize != null) {
this.parallelThreadSize = parallelThreadSize;
}
}
protected MultiStageCoprocessor buildMultiStageCoprocessor() {
MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize,
parallelThreadSize,
(LogEventConvert) binlogParser,
transactionBuffer,
destination);
mysqlMultiStageCoprocessor.setEventsPublishBlockingTime(eventsPublishBlockingTime);
return mysqlMultiStageCoprocessor;
}
在canal的master分支上，该bug已经修复，保证线程池至少有一个线程：
Ensure at least 1 thread for thread pool.
/*com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor*/
int tc = parserThreadCount > 0 ? parserThreadCount : 1;
this.parserExecutor = Executors.newFixedThreadPool(tc,
new NamedThreadFactory("MultiStageCoprocessor-Parser-" + destination));
本作业是在虚拟机上进行，就分配了一个cup，因此虽然canal日志中显示已经启动：
# cat /home/admin/canal-server/logs/canal/canal.log
2018-08-31 19:36:18.970 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
但实际上example实例并不能工作。
# cat /home/admin/canal-server/logs/example/example.log
2018-08-31 19:36:37.345 [destination = example , address = /192.168.83.128:3306 , EventParser] ERROR c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - dump address /192.168.83.128:3306 has an error, retrying. caused by
java.lang.IllegalArgumentException: null
at java.util.concurrent.ThreadPoolExecutor.<init>(ThreadPoolExecutor.java:1314) ~[na:1.8.0_181]
at java.util.concurrent.ThreadPoolExecutor.<init>(ThreadPoolExecutor.java:1237) ~[na:1.8.0_181]
at java.util.concurrent.Executors.newFixedThreadPool(Executors.java:151) ~[na:1.8.0_181]
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor.start(MysqlMultiStageCoprocessor.java:84) ~[canal.parse-1.1.0.jar:na]
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:238) ~[canal.parse-1.1.0.jar:na]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_181]
解决办法是更改虚拟机配置，设置成2个或以上cpu就能解决问题。通过对问题的分析，对于单cpu的物理机，无法通过修改cpu数量的情况，可以通过配置文件解决。
canal.instance.parser.parallelThreadSize = 1
4. 查看运行效果
对mysql进行update操作，就能看到日下日志：
# cat /home/admin/canal-server/logs/example/meta.log
2018-08-31 20:02:05.289 - clientId:1001 cursor:[mysql-bin.000007,3802,1535716924000,1002,] address[192.168.83.128/192.168.83.128:3306]
也可以运行ClientExample查看效果。
至此，作业完毕。