MySQL數據庫增量日誌解析工具 Canal 實戰
簡介
canal，阿里開源工具，主要用途是基於 MySQL 數據庫增量日誌解析，提供增量數據訂閱和消費
應用場景
- 數據庫實時備份
- 業務cache刷新
- 索引構建和實時維護，例：將商品數據推送到es中構建倒排索引
- 帶業務邏輯的增量數據處理，例：增量數據推送到第三方平臺
官網
https://github.com/alibaba/canalphp
原理
- 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...html
修改mysql配置文件my.cnfjava
[mysqld] log-bin=mysql-bin # 開啓 binlog binlog-format=ROW # 選擇 ROW 模式 server_id=1
新增用戶並受權，測試的話能夠直接使用root用戶mysql
CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '123456'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES;
配置canal
查看是否安裝javagit
➜ 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.pnggithub
訪問release頁面，下載最新穩定版1.14wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gzspring
解壓到指定目錄sql
mkdir /tmp/canal tar zxvf canal.deployer-1.1.4.tar.gz -C /tmp/canal
進入到canal目錄，查看文件docker
- 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
配置參數詳解：https://github.com/alibaba/ca...
我這裏修改了數據鏈接地址，用戶和密碼，以及同步規則：只同步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
下載canal php客戶端: https://github.com/xingwenge/canal-php
修改src/sample/client.php文件
<?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;
}
運行canalsh 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.安裝serverdocker 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 Webdocker 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
訪問Rocket管理頁面
http://localhost:8080/
修改配置文件
修改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.logtail -n 50 -f logs/example/example.log
順利的話，在RocketMQ中能夠查看到，新增的Topic和Message，RocketMQ能夠在manage頁面直接查看messge內容
後記
在安裝和使用canal的時候仍是遇到了一些坑，好比java高版本報錯，投遞到消息隊列失敗等，須要耐心排查log日誌，分析緣由，canal也用到了不少技術棧[zookeeper、kafka、RocketMQ]，後續將會進一步去深刻研究。
- 1. canal 基於Mysql數據庫增量日誌解析
- 2. mysql日誌解析工具canal介紹
- 3. mysql日誌解析工具canal
- 4. canal實戰（二）：使用canal-kafka實現數據庫增量實時更新
- 5. canal基於數據庫的日誌解析
- 6. MySql的Binlog日誌工具分析：Canal、Maxwell、Databus、DTS
- 7. 監聽mysql的binlog日誌工具分析：canal、Maxwell、Databus、DTS
- 8. Mysql數據庫優化--慢查日誌分析工具
- 9. Canal：同步mysql增量數據工具，一篇詳解核心知識點
- 10. 用 logparser日誌分析工具把nginx日誌寫入mysql數據庫
- 更多相關文章...
- • jQuery Mobile 工具欄 - jQuery Mobile 教程
- • AJAX 數據庫實例 - ASP 教程
- • PHP開發工具
- • Flink 數據傳輸及反壓詳解
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. No provider available from registry 127.0.0.1:2181 for service com.ddbuy.ser 解決方法
- 2. Qt5.7以上調用虛擬鍵盤(支持中文)，以及源碼修改(可拖動，水平縮放)
- 3. 軟件測試面試- 購物車功能測試用例設計
- 4. ElasticSearch（概念篇）：你知道的, 爲了搜索…
- 5. redux理解
- 6. gitee創建第一個項目
- 7. 支持向量機之硬間隔（一步步推導，通俗易懂）
- 8. Mysql 異步複製延遲的原因及解決方案
- 9. 如何在運行SEPM配置嚮導時將不可認的複雜數據庫密碼改爲簡單密碼
- 10. windows系統下tftp服務器使用
- 1. canal 基於Mysql數據庫增量日誌解析
- 2. mysql日誌解析工具canal介紹
- 3. mysql日誌解析工具canal
- 4. canal實戰（二）：使用canal-kafka實現數據庫增量實時更新
- 5. canal基於數據庫的日誌解析
- 6. MySql的Binlog日誌工具分析：Canal、Maxwell、Databus、DTS
- 7. 監聽mysql的binlog日誌工具分析：canal、Maxwell、Databus、DTS
- 8. Mysql數據庫優化--慢查日誌分析工具
- 9. Canal：同步mysql增量數據工具，一篇詳解核心知識點
- 10. 用 logparser日誌分析工具把nginx日誌寫入mysql數據庫