目录
一、下载canal
二、解压
三、开启数据库 Binlog 写入功能
四、建立一个canal账号
五、启动canal
六、python客户端
七、教程地址
一、下载canal
Releases · alibaba/canal (github.com)https://github.com/alibaba/canal/releases
二、解压
三、开启数据库 Binlog 写入功能
查看是否开启了binlog功能
show variables like 'log_bin';
开启步骤:
修改mysql配置文件my.cnf或者my.ini
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复
四、建立一个canal账号(用root账号执行一下语句!!!)
用root账号执行一下语句!!!,否则新建canal账号,赋权会失败,导致权限不足。
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
五、配置修改
QuickStart · alibaba/canal Wiki · GitHubhttps://github.com/alibaba/canal/wiki/QuickStart
## 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
七、python客户端
import time
from canal.client import Client
from canal.protocol import EntryProtocol_pb2
from canal.protocol import CanalProtocol_pb2
client: 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'example', filter=b'.*\\..*')
if __name__ == '__main__':
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:
format_data['event_type_name'] = '删除'
for column in row.beforeColumns:
format_data[column.name] = column.value
elif event_type == EntryProtocol_pb2.EventType.INSERT:
format_data['event_type_name'] = '新增'
for column in row.afterColumns:
format_data[column.name] = column.value
else:
format_data['event_type_name'] = '更新'
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()
七、教程地址
GitHub - alibaba/canal: 阿里巴巴 MySQL binlog 增量订阅&消费组件