canal-server | ==> EXIT CODE: 0

canal-server | ==> INIT /alidata/init/fix-hosts.py

canal-server | ==> EXIT CODE: 0

canal-server | ==> INIT DEFAULT

canal-server | Generating SSH1 RSA host key: [ OK ]

canal-server | Starting sshd: [ OK ]

canal-server | Starting crond: [ OK ]

canal-server | ==> INIT DONE

canal-server | ==> RUN /home/admin/app.sh

canal-server | ==> START …

canal-server | start canal …

canal-server | start canal successful

canal-server | ==> START SUCCESSFUL …

canal Python 客户端

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

import time

from canal.client import Client

from canal.protocol import EntryProtocol_pb2

from 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.py

connected to 127.0.0.1:11111

Auth succed

Subscribe succed

header {

version: 1

logfileName: “mysql-bin.000003”

logfileOffset: 5497

serverId: 1

serverenCode: “UTF-8”

executeTime: 1607843285000

sourceType: MYSQL

eventLength: 75

}

entryType: TRANSACTIONBEGIN

storeValue: " \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: ROWDATA

storeValue: “\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: TRANSACTIONEND

storeValue: “\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.log

2020-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.log

2020-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 position

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 - prepare to find start position just show master status

2020-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,这个地方确实困扰了我很长时间。

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

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

最后

针对最近很多人都在面试,我这边也整理了相当多的面试专题资料,也有其他大厂的面经。希望可以帮助到大家。

下面的面试题答案都整理成文档笔记。也还整理了一些面试资料&最新2021收集的一些大厂的面试真题(都整理成文档,小部分截图)

在这里插入图片描述

最新整理电子书

在这里插入图片描述

l.instance.filter.table.error=true

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

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

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

最后

针对最近很多人都在面试,我这边也整理了相当多的面试专题资料,也有其他大厂的面经。希望可以帮助到大家。

下面的面试题答案都整理成文档笔记。也还整理了一些面试资料&最新2021收集的一些大厂的面试真题(都整理成文档,小部分截图)

[外链图片转存中…(img-iJEsybdk-1720027236345)]

最新整理电子书

[外链图片转存中…(img-OAbmUfEO-1720027236346)]

GitHub 加速计划 / ca / canal
38
8
下载
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统,主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费,广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
最近提交(Master分支:9 个月前 )
50a57029 * fix(canal/prometheus): 修复延迟指标的缺陷 -【缺陷描述】 Admin控制台修改canal.properties配置后,Server自动进行重启,在Grafan上观测到同步实例的PUT/GET/ACK延迟居高不下。 -【原因定位】 源库心跳正常触发,MemoryStoreWithBuffer正常推进位点,profiling 正常统计,但是延迟指标依然是越来越来越高。Prometheus调用collect接口采集到的exec time时间始终是固定不变的。经debug排查到StoreCollector采集器内存hold的StoreMetricsHolder与CanalInstance实例中的引用已经不同啦,CanalIntance重启时已经被重建过一份新的实例,但是StoreMetricsHolder却没有保存到内存Hold中。原因是Map.putIfAbsent调用引起。 -【修复效果】 - 修复后,重启Server心跳正常推进,延迟瞬间降下来。梳理其他Collector代码都是调用Map.put,只有这里使用putIfAbsent可能是粗心导致吧。修复效果 * fix(canal/client-adapter): 修复adapter插件重试错误缺陷 -【缺陷描述】 adapter运行期间,针对server或者Instance进行过一次重启或者断网测试,发现adapter通过client拉取数据报错以后持续打印错误日志,永远不得恢复。虽然adapter processor 针对错误有重试逻辑,但是client的tcp连接已经发生中断,且无法恢复。 -【问题定位】 adapter processor 针对错误的处理比较粗,需要细分是从上游get message报错,还是下游sync data报错? 如果是 get message报错,特别是tcp 断开,需要跳出重试,重新尝试重连。如果是下游 sync data 报错,再考虑重试写入。下游插件开发者自行保证连接重连机制、写入幂等。框架只提供重试。 - 说明:顺带更新一下 .gitignore 文件 5 天前
6b7ddbbe -【缺陷描述】 Admin控制台修改canal.properties配置后,Server自动进行重启,在Grafan上观测到同步实例的PUT/GET/ACK延迟居高不下。 -【原因定位】 源库心跳正常触发,MemoryStoreWithBuffer正常推进位点,profiling 正常统计,但是延迟指标依然是越来越来越高。Prometheus调用collect接口采集到的exec time时间始终是固定不变的。经debug排查到StoreCollector采集器内存hold的StoreMetricsHolder与CanalInstance实例中的引用已经不同啦,CanalIntance重启时已经被重建过一份新的实例,但是StoreMetricsHolder却没有保存到内存Hold中。原因是Map.putIfAbsent调用引起。 -【修复效果】 - 修复后,重启Server心跳正常推进,延迟瞬间降下来。梳理其他Collector代码都是调用Map.put,只有这里使用putIfAbsent可能是粗心导致吧。修复效果 5 天前
Logo

旨在为数千万中国开发者提供一个无缝且高效的云端环境,以支持学习、使用和贡献开源项目。

更多推荐