canal

canal使用的是1.1.5版本

# 下载镜像
docker pull canal/canal-server:v1.1.5
# 创建挂载的目录
mkdir /root/canal/conf
# 创建挂载的配置文件
touch /root/canal/conf/instance.properties
复制代码

instance.properties配置

## mysql serverId
# 目前最新的服务端版本已不需要配置serverId参数
canal.instance.mysql.slaveId = 1234  
#position info，需要改成自己的数据库信息
canal.instance.master.address = 172.16.204.13: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
# binlog解析的过滤规则，采用正则表达式
canal.instance.filter.regex = .*\\..*
复制代码

Note

• canal.instance.master.address = 172.16.204.13:3306,这个地址一定是myql对应的地址。
• 虽然两个镜像是在同一个宿主机上部署，但是不能通过127.0.0.1:3306进行访问。这里把服务映射到宿主机的ip上了，互相访问是需要通过宿主机。
• 如果不是使用docker，而是直接下载canal和mysql在本地安装，可以设置为127.0.0.1:3306

启动宿主机

docker run --name=canal -p 11111:11111 -d -v /root/CC/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties canal/canal-server
# 进入宿主机
docker exec -it canal /bin/bash
# 查看日志
cat /home/admin/canal-server/logs/example/example.log
复制代码

mysql ip配置不正确： [图片上传失败...(image-90c39f-1654766495553)]

mysql ip配置正确： [图片上传失败...(image-62cfdf-1654766495553)]

python操作

python 3.6 安装包

pip install canal-python
复制代码

需要在mysql里面创建库和表，我直接使用navicat创建的 这里建了一个demo的库，canal_test的表 [图片上传失败...(image-d429b2-1654766495553)]

python演示代码

# canal_demo.py
import time

from canal.client import Client
from canal.protocol import EntryProtocol_pb2
from canal.protocol import CanalProtocol_pb2

# 建立与canal服务端的连接
client = Client()
client.connect(host='172.16.204.1', port=11111)   # canal服务端部署的主机IP与端口
client.check_valid(username=b'canal', password=b'canal')  # 自行填写配置的数据库账户密码
# destination是canal服务端的服务名称， filter即获取数据的过滤规则，采用正则表达式
client.subscribe(client_id=b'1001', destination=b'example', filter=b'.*\..*')

while True:
    message = client.get(100)
    # entries是每个循环周期内获取到数据集
    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
        # row是binlog解析出来的行变化记录，一般有三种格式，对应增删改
        for row in row_change.rowDatas:
            format_data = dict()
            # 根据增删改的其中一种情况进行数据处理
            if event_type == EntryProtocol_pb2.EventType.DELETE:
                format_data['before'] = dict()
                for column in row.beforeColumns:
                    #format_data = {
                    #    column.name: column.value
                    #}
                    #此处注释为原demo，有误，下面是正确写法
                    format_data['before'][column.name] = column.value
            elif event_type == EntryProtocol_pb2.EventType.INSERT:
                format_data['after'] = dict()
                for column in row.afterColumns:
                    #format_data = {
                    #    column.name: column.value
                    #}
                    #此处注释为原demo，有误，下面是正确写法
                    format_data['after'][column.name] = column.value
            else:
                # format_data['before'] = format_data['after'] = dict()  采用下面的写法应该更好
                format_data['before'] = dict()
                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即最后获取的数据，包含库名，表明，事务类型，改动数据
            data = dict(
                db=database,
                table=table,
                event_type=event_type,
                data=format_data,
            )
            print(data)
    time.sleep(1)

client.disconnect()
复制代码

启动python文件 python canal_demo.py

connected to 172.16.204.13:11111
Auth succed
Subscribe succed
复制代码

在canal_test表修改数据

[图片上传失败...(image-9859f0-1654766495553)]

{'db': 'demo', 'table': 'canal_test', 'event_type': 1, 'data': {'after': {'id': '3', 'name': '33'}}}
{'db': 'demo', 'table': 'canal_test', 'event_type': 2, 'data': {'before': {'id': '3', 'name': '33'}, 'after': {'id': '3', 'name': '44'}}}
{'db': 'demo', 'table': 'canal_test', 'event_type': 2, 'data': {'before': {'id': '2', 'name': '22'}, 'after': {'id': '2', 'name': '222'}}}
{'db': 'demo', 'table': 'canal_test', 'event_type': 2, 'data': {'before': {'id': '1', 'name': 'cc'}, 'after': {'id': '1', 'name': '111'}}}
{'db': 'demo', 'table': 'canal_test2', 'event_type': 1, 'data': {'after': {'id': '1', 'level': '1'}}}
{'db': 'demo', 'table': 'canal_test', 'event_type': 1, 'data': {'after': {'id': '4', 'name': '传参'}}}
{'db': 'demo', 'table': 'canal_test', 'event_type': 3, 'data': {'before': {'id': '4', 'name': '传参'}}}
复制代码

Note

• 由于canal是读取binlog，只对日志中增删改进行处理，所有对于创建表和删除表是不会有记录的
• canal返回数据汇总event_type：1表示新增，2表示更新，3表示删除

作者：无糖朋克
链接：https://juejin.cn/post/7051863581742071821
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。