簡介

canal，阿里開源工具，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費

應用場景

數據庫實時備份

業務cache刷新

索引構建和實時維護，例：將商品數據推送到es中構建倒排索引

帶業務邏輯的增量數據處理，例：增量數據推送到第三方平臺

官網

原理

MySQL master將數據寫入binlog

canal 向master發送dump協議

master 收到dump請求，推送binlog給canal

canal 解析binlog，可講數據投遞到MQ系統中，目前支持kafka、RocketMQ

安裝

配置mysql

建議的mysql版本是5.7.x

mysql8.0.x見官方說明https://github.com/alibaba/ca...

修改mysql配置文件my.cnf

[mysqld]

log-bin=mysql-bin # 開啟 binlog

binlog-format=ROW # 選擇 ROW 模式

server_id=1

新增用戶并授權，測試的話可以直接使用root用戶

CREATE USER canal IDENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '123456'@'%';

-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

配置canal

查看是否安裝java

? canal-admin java -version

java version "1.8.0_251"

Java(TM) SE Runtime Environment (build 1.8.0_251-b08)

Java HotSpot(TM) 64-Bit Server VM (build 25.251-b08, mixed mode)

若沒有安裝，去oracle官網下載1.8版本(切勿選擇高版本)，選擇適合自己系統的版本安裝即可，mac系統下載的dmg，直接點擊安裝，不再累述image.png

訪問release頁面，下載最新穩定版1.14

wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz

解壓到指定目錄

mkdir /tmp/canal

tar zxvf canal.deployer-1.1.4.tar.gz -C /tmp/canal

進入到canal目錄，查看文件

bin目錄：運行命令

conf目錄：配置文件目錄

logs目錄：記錄運行log

lib目錄：jar包

? ll

total 0

drwxr-xr-x 7 jiao staff 224B 6 3 17:17 bin

drwxr-xr-x 9 jiao staff 288B 6 3 20:12 conf

drwxr-xr-x 83 jiao staff 2.6K 6 1 17:54 lib

drwxr-xr-x 7 jiao staff 224B 6 3 17:05 logs

drwxr-xr-x 7 jiao staff 224B 5 29 17:43 pierced

修改instance配置文件

vi conf/example/instance.properties

我這里修改了數據連接地址，用戶和密碼，以及同步規則：只同步test數據庫的sc_user表

## mysql serverId

canal.instance.mysql.slaveId = 1234

#position info，需要改成自己的數據庫信息

canal.instance.master.address = 127.0.0.1:3306

canal.instance.dbUsername = canal

canal.instance.dbPassword = 123456

#主庫binlog文件路徑

#canal.instance.standby.journal.name =

#主庫binlog偏移量

#canal.instance.standby.position =

canal.instance.filter.regex = test\\.sc_user

修改Server配置文件

vim conf/canal.properties

#tcp bind ip，設置投遞到tcp時需設置

canal.ip = 192.168.101.47

#register ip to zookeeper

canal.register.ip = 192.168.101.47

canal.port = 11111

canal.metrics.pull.port = 11112

#canal instance user/passwd

#canal.user = canal

#canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458

#canal-admin需設置，先不設置

#canal.admin.manager = 127.0.0.1:8089

canal.admin.port = 11110

canal.admin.user = admin

canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441

#zk設置，單機的canal可不設置

canal.zkServers = 127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

#flush data to zk

canal.zookeeper.flush.period = 1000

canal.withoutNetty = false

#tcp, kafka, RocketMQ，投遞到哪

canal.serverMode = RocketMQ

#flush meta cursor/parse position to file

canal.file.data.dir = ${canal.conf.dir}

canal.file.flush.period = 1000

##memory store RingBuffer size, should be Math.pow(2,n)

canal.instance.memory.buffer.size = 16384

##memory store RingBuffer used memory unit size , default 1kb

canal.instance.memory.buffer.memunit = 1024

##meory store gets mode used MEMSIZE or ITEMSIZE

canal.instance.memory.batch.mode = MEMSIZE

canal.instance.memory.rawEntry = true

##detecing config

canal.instance.detecting.enable = false

#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()

canal.instance.detecting.sql = select 1

canal.instance.detecting.interval.time = 3

canal.instance.detecting.retry.threshold = 3

canal.instance.detecting.heartbeatHaEnable = false

#support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery

canal.instance.transaction.size = 1024

#mysql fallback connected to new master should fallback times

canal.instance.fallbackIntervalInSeconds = 60

#network config

canal.instance.network.receiveBufferSize = 16384

canal.instance.network.sendBufferSize = 16384

canal.instance.network.soTimeout = 30

#binlog filter config

canal.instance.filter.druid.ddl = true

canal.instance.filter.query.dcl = false

canal.instance.filter.query.dml = false

canal.instance.filter.query.ddl = false

canal.instance.filter.table.error = false

canal.instance.filter.rows = false

canal.instance.filter.transaction.entry = false

#binlog format/image check

canal.instance.binlog.format = ROW,STATEMENT,MIXED

canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

#binlog ddl isolation

canal.instance.get.ddl.isolation = false

#parallel parser config

canal.instance.parser.parallel = true

##concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()

#canal.instance.parser.parallelThreadSize = 16

##disruptor ringbuffer size, must be power of 2

canal.instance.parser.parallelBufferSize = 256

#table meta tsdb info

canal.instance.tsdb.enable = true

canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}

canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;

canal.instance.tsdb.dbUsername = canal

canal.instance.tsdb.dbPassword = canal

#dump snapshot interval, default 24 hour

canal.instance.tsdb.snapshot.interval = 24

#purge snapshot expire , default 360 hour(15 days)

canal.instance.tsdb.snapshot.expire = 360

#aliyun ak/sk , support rds/mq

canal.aliyun.accessKey =

canal.aliyun.secretKey =

#################################################

######### destinations #############

#################################################

#使用的instance，可設置上面定義的example

canal.destinations = example

#conf root dir

canal.conf.dir = ../conf

#auto scan instance dir add/remove and start/stop instance

canal.auto.scan = true

canal.auto.scan.interval = 5

canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml

#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = manager

canal.instance.global.lazy = false

canal.instance.global.manager.address = ${canal.admin.manager}

#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml

canal.instance.global.spring.xml = classpath:spring/file-instance.xml

#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

##################################################

######### MQ #############

##################################################

#rocketmq地址設置

canal.mq.servers = 192.168.101.47:9876

canal.mq.retries = 0

canal.mq.batchSize = 16384

canal.mq.maxRequestSize = 1048576

canal.mq.lingerMs = 100

canal.mq.bufferMemory = 33554432

canal.mq.canalBatchSize = 50

canal.mq.canalGetTimeout = 100

canal.mq.flatMessage = true

canal.mq.compressionType = none

canal.mq.acks = all

#canal.mq.properties. =

canal.mq.producerGroup = test

#Set this value to "cloud", if you want open message trace feature in aliyun.

canal.mq.accessChannel = local

#aliyun mq namespace

#canal.mq.namespace =

##################################################

######### Kafka Kerberos Info #############

##################################################

canal.mq.kafka.kerberos.enable = false

canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"

canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"

測試tcp

修改src/sample/client.php文件

namespace xingwenge\canal_php\sample;

use xingwenge\canal_php\CanalClient;

use xingwenge\canal_php\CanalConnectorFactory;

use xingwenge\canal_php\Fmt;

require_once __DIR__ . '/../../vendor/autoload.php';

ini_set('display_errors', 'On');

error_reporting(E_ALL);

try {

$host = '192.168.101.47';//canal server地址

$client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET);

# $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SWOOLE);

$client->connect($host, 11111);

$client->checkValid('canal','E3619321C1A937C46A0D8BD1DAC39F93B27D4458');//認證信息，對于canal.properties配置的canal.user 和canal.passwd

// $client->subscribe("1001", "example", ".*\\..*");

$client->subscribe("1001", "example", "test.sc_user"); # 設置過濾規則

while (true) {

$message = $client->get(100);

if ($entries = $message->getEntries()) {

foreach ($entries as $entry) {

Fmt::println($entry);

}

}

sleep(1);

}

$client->disConnect();

} catch (\Exception $e) {

echo $e->getMessage(), PHP_EOL;

}

運行canal

sh bin/startup.sh

查看log是否有錯誤信息

tail -n 50 -f logs/example/example.log

輸出一下內容說明canal已正常啟動，canal已成功連接mysql準備發送dump指令同步數據

2020-06-03 19:30:47.625 [destination = example , address = /127.0.0.1:3307 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=mysql-bin.000001,position=303179,serverId=1,gtid=,timestamp=1591183839000] cost : 5ms , the next step is binlog dump

運行php client.php，并在數據庫sc_user表中修改一條數據，可以看到client輸出了數據庫修改的內容

`================> binlog[mysql-bin.000001 : 208118],name[wcc-scm-service,sc_status_history], eventType: 2

-------> before

status_history_id : 2543 update= false

record_id : 662 update= false

status_type : 151 update= false

old_status : 158 update= false

new_status : 153 update= false

opt_type : SELLER_REVOKE update= false

user_id : 282 update= false

user_name : 加工廠1 update= false

remark : update= false

created : 2020-05-28 18:40:02 update= false

-------> after

status_history_id : 2543 update= false

record_id : 662 update= false

status_type : 151 update= false

old_status : 158 update= false

new_status : 153 update= false

opt_type : SELLER_REVOKE update= false

user_id : 282 update= false

user_name : 加工廠12 update= true

remark : update= false

created : 2020-05-28 18:40:02 update= false

TSocket: Could not read 4 bytes from 192.168.101.47:11111`

測試RocketMQ

安裝RocketMQ

這里直接在docker里安裝

1.安裝server

docker run -d -p 9876:9876 --name rocketmq-server -e "MAX\_POSSIBLE\_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv

2.新增配置文件

echo "brokerIP1=192.168.101.47" > broker.properties

3.安裝broker將/path/broker.properties替換成第2步文件路徑

docker run -d -p 10911:10911 -p 10909:10909 -v /path/broker.properties:/opt/rocketmq-4.4.0/bin/broker.properties --name rocketmq-broker --link rocketmq-server -e "NAMESRV\_ADDR=rocketmq-server:9876" -e "MAX\_POSSIBLE\_HEAP=200000000" rocketmqinc/rocketmq sh mqbroker -c broker.properties

4.安裝RocketMQ Web

docker run -e "JAVA\_OPTS=-Drocketmq.namesrv.addr=192.168.101.47:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -p 8080:8080 -t styletang/rocketmq-console-ng -d

修改配置文件

修改canal.properties

投遞到tcp修改成投遞到RocketMQ

`canal.serverMode = RocketMQ

canal.mq.servers = 192.168.101.47:9876`

運行canal

sh bin/startup.sh

查看log文件，排查是否有錯誤

tail -n 50 -f logs/canal/canal.log

tail -n 50 -f logs/example/example.log

順利的話，在RocketMQ中可以查看到，新增的Topic和Message，RocketMQ可以在manage頁面直接查看messge內容

后記

在安裝和使用canal的時候還是遇到了一些坑，比如java高版本報錯，投遞到消息隊列失敗等，需要耐心排查log日志，分析原因，canal也用到了很多技術棧[zookeeper、kafka、RocketMQ]，后續將會進一步去深入研究。

總結

以上是生活随笔為你收集整理的mysql batch mode_MySQL数据库增量日志解析工具 Canal 实战的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。