canal快速监听数据库变化-window版

目录

一、下载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 增量订阅&消费组件