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

待我称王封你为后i 2024-03-26 22:06 81阅读 0赞

**目录**

一、下载canal

二、解压

三、开启数据库 Binlog 写入功能

四、建立一个canal账号

五、启动canal

六、python客户端

七、教程地址

--------------------

#### 一、下载canal ####

[Releases · alibaba/canal (github.com)![icon-default.png_t_N2N8][]https://github.com/alibaba/canal/releases][Releases _ alibaba_canal _github.com_icon-default.png_t_N2N8_https_github.com_alibaba_canal_releases]![02c2f891bafe4230b9eb05b9ddd132c1.png][]

#### 二、解压 ####

![9f94a73ee8dd475cb8d591de41204198.png][]

#### 三、开启数据库 Binlog 写入功能 ####

查看是否开启了binlog功能

show variables like 'log_bin';

![dfe449ebbb4c4c75918a610a8dd8a75b.png][]

开启步骤：

修改mysql配置文件my.cnf或者my.ini

![c6e300d4af8d411bb3a26e5b4dc18fa1.png][]

[mysqld]
    log-bin=mysql-bin # 开启 binlog
    binlog-format=ROW # 选择 ROW 模式
    server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

#### 四、建立一个canal账号（**用root账号执行一下语句！！！**） ####

**用root账号执行一下语句！！！，否则新建canal账号，赋权会失败，导致权限不足。**

![5c3faf3d0d8e4a129da993bacdcb16f3.png][]

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 · GitHub![icon-default.png_t_N2N8][]https://github.com/alibaba/canal/wiki/QuickStart][QuickStart _ alibaba_canal Wiki _ GitHub_icon-default.png_t_N2N8_https_github.com_alibaba_canal_wiki_QuickStart]

![20a3db66e29d48faa93daf2b7b3146b1.png][]

## 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 ####

![c03b83eb3c72486baa041450eb8557d8.png][]

#### 七、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 增量订阅&消费组件][GitHub - alibaba_canal_ _ MySQL binlog]

[icon-default.png_t_N2N8]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/fc00813b64eb4b07bffaadd27bba0c82.png
[Releases _ alibaba_canal _github.com_icon-default.png_t_N2N8_https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases
[02c2f891bafe4230b9eb05b9ddd132c1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/d65ef2a9ac634184ac44fe943b495856.png
[9f94a73ee8dd475cb8d591de41204198.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/02c0a89f8c2242f48236f218aa848e6e.png
[dfe449ebbb4c4c75918a610a8dd8a75b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/a63f1a825bde4070b0297fbd83e4617a.png
[c6e300d4af8d411bb3a26e5b4dc18fa1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1beda78f7e7a4437a0da5b048bb0cc04.png
[5c3faf3d0d8e4a129da993bacdcb16f3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/976e07f7e0b241a7b4ef1d0c824611ce.png
[QuickStart _ alibaba_canal Wiki _ GitHub_icon-default.png_t_N2N8_https_github.com_alibaba_canal_wiki_QuickStart]: https://github.com/alibaba/canal/wiki/QuickStart
[20a3db66e29d48faa93daf2b7b3146b1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/d73e44422b4548f9b83c93c59e881e5b.png
[c03b83eb3c72486baa041450eb8557d8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/cee0fdb9738341c7958bac72c5c24de9.png
[GitHub - alibaba_canal_ _ MySQL binlog]: https://github.com/alibaba/canal