canal

canal使用的是1.1.5版本

# 下載鏡像
docker pull canal/canal-server:v1.1.5
# 創(chuàng)建掛載的目錄
mkdir /root/canal/conf
# 創(chuàng)建掛載的配置文件
touch /root/canal/conf/instance.properties
復(fù)制代碼

instance.properties配置

## mysql serverId
# 目前最新的服務(wù)端版本已不需要配置serverId參數(shù)
canal.instance.mysql.slaveId = 1234  
#position info，需要改成自己的數(shù)據(jù)庫信息
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召娜，需要改成自己的數(shù)據(jù)庫信息
canal.instance.dbUsername = canal  
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
# table regex
# binlog解析的過濾規(guī)則尖奔，采用正則表達式
canal.instance.filter.regex = .*\\..*
復(fù)制代碼

Note

• canal.instance.master.address = 172.16.204.13:3306,這個地址一定是myql對應(yīng)的地址沉衣。
• 雖然兩個鏡像是在同一個宿主機上部署逆航，但是不能通過127.0.0.1:3306進行訪問捣卤。這里把服務(wù)映射到宿主機的ip上了仇让，互相訪問是需要通過宿主機蜈膨。
• 如果不是使用docker悼沿，而是直接下載canal和mysql在本地安裝等舔，可以設(shè)置為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
復(fù)制代碼

mysql ip配置不正確： [圖片上傳失敗...(image-90c39f-1654766495553)]

mysql ip配置正確： [圖片上傳失敗...(image-62cfdf-1654766495553)]

python操作

python 3.6 安裝包

pip install canal-python
復(fù)制代碼

需要在mysql里面創(chuàng)建庫和表，我直接使用navicat創(chuàng)建的 這里建了一個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服務(wù)端的連接
client = Client()
client.connect(host='172.16.204.1', port=11111)   # canal服務(wù)端部署的主機IP與端口
client.check_valid(username=b'canal', password=b'canal')  # 自行填寫配置的數(shù)據(jù)庫賬戶密碼
# destination是canal服務(wù)端的服務(wù)名稱慌植， filter即獲取數(shù)據(jù)的過濾規(guī)則，采用正則表達式
client.subscribe(client_id=b'1001', destination=b'example', filter=b'.*\..*')

while True:
    message = client.get(100)
    # entries是每個循環(huán)周期內(nèi)獲取到數(shù)據(jù)集
    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
        # 數(shù)據(jù)庫名
        database = header.schemaName
        # 表名
        table = header.tableName
        event_type = header.eventType
        # row是binlog解析出來的行變化記錄义郑，一般有三種格式蝶柿，對應(yīng)增刪改
        for row in row_change.rowDatas:
            format_data = dict()
            # 根據(jù)增刪改的其中一種情況進行數(shù)據(jù)處理
            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()  采用下面的寫法應(yīng)該更好
                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即最后獲取的數(shù)據(jù)劫笙，包含庫名芙扎，表明，事務(wù)類型填大，改動數(shù)據(jù)
            data = dict(
                db=database,
                table=table,
                event_type=event_type,
                data=format_data,
            )
            print(data)
    time.sleep(1)

client.disconnect()
復(fù)制代碼

啟動python文件 python canal_demo.py

connected to 172.16.204.13:11111
Auth succed
Subscribe succed
復(fù)制代碼

在canal_test表修改數(shù)據(jù)

[圖片上傳失敗...(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': '傳參'}}}
復(fù)制代碼

Note

• 由于canal是讀取binlog戒洼，只對日志中增刪改進行處理，所有對于創(chuàng)建表和刪除表是不會有記錄的
• canal返回數(shù)據(jù)匯總event_type：1表示新增允华，2表示更新圈浇，3表示刪除

作者：無糖朋克
鏈接：https://juejin.cn/post/7051863581742071821
來源：稀土掘金
著作權(quán)歸作者所有寥掐。商業(yè)轉(zhuǎn)載請聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請注明出處磷蜀。