写点什么

使用 Docker 部署 canal 服务,实现 MySQL 数据库 binlog 日志解析

用户头像
AlwaysBeta
关注
发布于: 2020 年 12 月 13 日

canal 是阿里巴巴开源的一个项目,主要用途是基于 MySQL 数据库 binlog 日志解析,提供增量数据订阅和消费。


基于日志增量订阅和消费的业务包括:


  • 数据库镜像

  • 数据库实时备份

  • 索引构建和实时维护(拆分异构索引、倒排索引等)

  • 业务 cache 刷新

  • 带业务逻辑的增量数据处理


我这边主要在两个场景下使用:


一个是将变更数据实时同步到 Elasticsearch 和 Redis。


这里先说一下我目前的做法,一方面是全量数据定时同步,由于数据量比较大,同步时间比较长,所以数据也就不够实时。第二个方面是针对单条数据的变更,部分更新 Elasticsearch 和 Redis 的逻辑都是直接写在了业务代码中,耦合比较严重。


拆出来之后就可以实现实时增量更新,而且还可以解耦,收益还是很大的。


第二个是保存重点关注数据的历史变更。


这个目前用在了「资产管理」模块,通过记录 IP 资产的创建,变更以及删除,实现 IP 生命周期管理,方便历史信息回溯。


MySQL 配置


修改 MySQL 配置文件 my.cnf,开启 binlog 写入功能,并配置模式为 ROW。


log-bin=mysql-bin # 开启 binlogbinlog-format=ROW # 选择 ROW 模式server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复
复制代码

重启数据库,查看配置是否生效。


mysql> show variables like 'binlog_format';+---------------+-------+| Variable_name | Value |+---------------+-------+| binlog_format | ROW   |+---------------+-------+1 row in set (0.19 sec)mysql>mysql> show variables like 'log_bin';+---------------+-------+| Variable_name | Value |+---------------+-------+| log_bin       | ON    |+---------------+-------+1 row in set (0.00 sec)mysql>mysql> show master status;+------------------+----------+--------------+------------------+-------------------+| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |+------------------+----------+--------------+------------------+-------------------+| mysql-bin.000003 |     4230 |              |                  |                   |+------------------+----------+--------------+------------------+-------------------+1 row in set (0.00 sec)
复制代码

然后创建用户,并授权。


mysql> CREATE USER canal IDENTIFIED BY 'canal';mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%%';mysql> FLUSH PRIVILEGES;mysql> show grants for 'canal'@'%%';+----------------------------------------------------------------------------+| Grants for canal@%%                                                        |+----------------------------------------------------------------------------+| GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO `canal`@`%%` |+----------------------------------------------------------------------------+1 row in set (0.00 sec)
复制代码

canal 服务端


拉取镜像:


# docker pull canal/canal-server:v1.1.4
复制代码

然后用官方提供的 shell 脚本直接启动:


# sh run.sh -e canal.auto.scan=false -e canal.destinations=test -e canal.instance.master.address=127.0.0.1:3306 -e canal.instance.dbUsername=canal -e canal.instance.dbPassword=canal -e canal.instance.connectionCharset=UTF-8 -e canal.instance.tsdb.enable=true -e canal.instance.gtidon=false
复制代码

但每次都这样启动还是有点麻烦,可以写一个 docker-compose 文件,如下:


version: '3'
services: canal-server: image: canal/canal-server:v1.1.4 container_name: canal-server restart: unless-stopped network_mode: host ports: - 11111:11111 environment: - canal.auto.scan=false - canal.instance.master.address=127.0.0.1:3306 - canal.instance.dbUsername=canal - canal.instance.dbPassword=canal - canal.instance.filter.regex=.*\\..* - canal.destinations=test - canal.instance.connectionCharset=UTF-8 - canal.instance.tsdb.enable=true volumes: - /root/canal/test/log/:/home/admin/canal-server/logs/
复制代码

启动服务:


# docker-compose upRecreating canal-server ... doneAttaching to canal-servercanal-server    | DOCKER_DEPLOY_TYPE=VMcanal-server    | ==> INIT /alidata/init/02init-sshd.shcanal-server    | ==> EXIT CODE: 0canal-server    | ==> INIT /alidata/init/fix-hosts.pycanal-server    | ==> EXIT CODE: 0canal-server    | ==> INIT DEFAULTcanal-server    | Generating SSH1 RSA host key: [  OK  ]canal-server    | Starting sshd: [  OK  ]canal-server    | Starting crond: [  OK  ]canal-server    | ==> INIT DONEcanal-server    | ==> RUN /home/admin/app.shcanal-server    | ==> START ...canal-server    | start canal ...canal-server    | start canal successfulcanal-server    | ==> START SUCCESSFUL ...
复制代码

canal Python 客户端


直接 Copy 官方提供的客户端代码:


import time
from canal.client import Clientfrom canal.protocol import EntryProtocol_pb2from canal.protocol import CanalProtocol_pb2
client = Client()client.connect(host='127.0.0.1', port=11111)client.check_valid(username=b'', password=b'')client.subscribe(client_id=b'1001', destination=b'test', filter=b'.*\\..*')
while True: message = client.get(100) entries = message['entries'] for entry in entries: entry_type = entry.entryType if entry_type in [EntryProtocol_pb2.EntryType.TRANSACTIONBEGIN, EntryProtocol_pb2.EntryType.TRANSACTIONEND]: continue row_change = EntryProtocol_pb2.RowChange() row_change.MergeFromString(entry.storeValue) event_type = row_change.eventType header = entry.header database = header.schemaName table = header.tableName event_type = header.eventType for row in row_change.rowDatas: format_data = dict() if event_type == EntryProtocol_pb2.EventType.DELETE: for column in row.beforeColumns: format_data = { column.name: column.value } elif event_type == EntryProtocol_pb2.EventType.INSERT: for column in row.afterColumns: format_data = { column.name: column.value } else: format_data['before'] = format_data['after'] = dict() for column in row.beforeColumns: format_data['before'][column.name] = column.value for column in row.afterColumns: format_data['after'][column.name] = column.value data = dict( db=database, table=table, event_type=event_type, data=format_data, ) print(data) time.sleep(1)
client.disconnect()
复制代码

功能验证


首先在 MySQL 里边创建一张测试表,然后再增删改几条测试数据:


mysql> create database test;mysql> use test;mysql> CREATE TABLE `role` (   `id` int unsigned NOT NULL AUTO_INCREMENT,   `role_name` varchar(255)DEFAULT NULL,   PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;mysql> insert into role (id, role_name) values (10, 'admin');Query OK, 1 row affected (0.01 sec)
mysql> update role set role_name='hh' where id = 10;Query OK, 1 row affected (0.01 sec)Rows matched: 1 Changed: 1 Warnings: 0
mysql> delete from role where id = 10;Query OK, 1 row affected (0.01 sec)
复制代码

客户端打印输出:


$ python canal_client.pyconnected to 127.0.0.1:11111Auth succedSubscribe succed

header { version: 1 logfileName: "mysql-bin.000003" logfileOffset: 5497 serverId: 1 serverenCode: "UTF-8" executeTime: 1607843285000 sourceType: MYSQL eventLength: 75}entryType: TRANSACTIONBEGINstoreValue: " \217\001"
header { version: 1 logfileName: "mysql-bin.000003" logfileOffset: 5630 serverId: 1 serverenCode: "UTF-8" executeTime: 1607843285000 sourceType: MYSQL schemaName: "test" tableName: "role" eventLength: 47 eventType: INSERT props { key: "rowsCount" value: "1" }}entryType: ROWDATAstoreValue: "\010\322\001\020\001P\000bN\022 \010\000\020\004\032\002id \001(\0010\000B\00210R\014int unsigned\022*\010\001\020\014\032\trole_name \000(\0010\000B\005adminR\014varchar(255)"
{'db': 'test', 'table': 'role', 'event_type': 1, 'data': {'role_name': 'admin'}}header { version: 1 logfileName: "mysql-bin.000003" logfileOffset: 5677 serverId: 1 serverenCode: "UTF-8" executeTime: 1607843285000 sourceType: MYSQL eventLength: 31}entryType: TRANSACTIONENDstoreValue: "\022\003440"
复制代码

变更一条数据,输出内容分三部分,分别是:TRANSACTIONBEGIN,ROWDATA 和 TRANSACTIONEND。然后我们比较关注的内容都在 ROWDATA 中,解析出来之后就是我们需要的,包括数据库名,表名和变更内容。


其中 event_type 字段 1 表示新增,2 表示更新,3 表示删除。


update 对应输出:


{'db': 'test', 'table': 'role', 'event_type': 2, 'data': {'before': {'id': '10', 'role_name': 'hh'}, 'after': {'id': '10', 'role_name': 'hh'}}}
复制代码

delete 对应输出:


{'db': 'test', 'table': 'role', 'event_type': 3, 'data': {'role_name': 'hh'}}
复制代码

canal 服务端启动之后,在 /home/admin/canal-server/logs/test 目录下会生成两个日志文件,分别是:meta.log 和 test.log,可以查看服务是不是正常,有没有报错信息。其中 test 是启动 Docker 时 canal.destinations 设置的名称。


# cat meta.log2020-12-13 14:55:18.051 - clientId:1001 cursor:[mysql-bin.000003,4805,1607842360000,1,] address[/127.0.0.1:3306]2020-12-13 14:55:33.051 - clientId:1001 cursor:[mysql-bin.000003,5096,1607842531000,1,] address[127.0.0.1:3306]2020-12-13 14:57:07.051 - clientId:1001 cursor:[mysql-bin.000003,5387,1607842625000,1,] address[127.0.0.1:3306]
# cat test.log2020-12-13 14:55:09.067 [main] INFO c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....2020-12-13 14:55:09.144 [destination = test , address = /127.0.0.1:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position2020-12-13 14:55:09.144 [destination = test , address = /127.0.0.1:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status2020-12-13 14:55:09.693 [destination = test , address = /127.0.0.1:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=mysql-bin.000003,position=4699,serverId=1,gtid=,timestamp=1607842360000] cost : 538ms , the next step is binlog dump
复制代码

踩坑记录


在我自己搭建的测试环境一切正常,但放到项目 beta 环境上还是遇到了一个问题:


[fetch failed by table meta:schemeName.tableName]


查了一下说是由于表删除,或者表结构变更引起的解析错误,增加一条配置就可以解决:


canal.instance.filter.table.error=true
复制代码

加上之后,报错信息的确都没有了,但消费出来的数据没有 ROWDATA,这个地方确实困扰了我很长时间。


说实话,有的时候调试程序,并不怕碰到报错,怕的是没有报错,然后程序还不正常。


后来,我把忽略表错误的配置删除,又仔细看了一遍日志,发现还有一个报错:


Caused by: java.io.IOException: ErrorPacket [errorNumber=1142, fieldCount=-1, message=SHOW command denied to user


这明显就是权限不够嘛,问了一下我们的 DBA,果然如此,我们的 binlog 账号默认是没有 select 权限的,加上之后,问题就成功解决了。


静下心来仔细看日志是多么重要。


以上,下篇会说说对接 MQ 的事。


参考文档:


https://github.com/alibaba/canal


https://github.com/haozi3156666/canal-python

发布于: 2020 年 12 月 13 日阅读数: 1152
用户头像

AlwaysBeta

关注

微信公众号:AlwaysBeta 2017.11.30 加入

后端开发,技术栈:Python,Django,Redis,RabbitMQ,Kafka,Elasticsearch,MySQL

评论 (1 条评论)

发布
用户头像
写的太简单了
2020 年 12 月 14 日 08:50
回复
没有更多了
使用 Docker 部署 canal 服务,实现 MySQL 数据库 binlog 日志解析