mysql batch mode_MySQL数据库增量日志解析工具 Canal 实战
简介
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 实战相关推荐
- canal 历史数据如何处理_MySQL日志解析工具Canal的使用
canal伪装为slave存在 接收master的同步日志 什么是Canal 简单来说就是MySQL数据同步工具,通过canal可以实现对MySQL数据变更的监听,获取数据变更情况,并应用到业务中,举 ...
- canal应用一:基于mysql binlog的日志解析工具
文章目录 前言 mysql 配置 canal 下载安装 canal client canal admin 常见问题 前言 canal 是阿里巴巴开发的MySQL数据实时同步工具,基于binlog增量日 ...
- postgresql 安装 配置远程访问 PostgreSQL WAL日志解析工具: wal2json cdc数据同步方案
文章目录 一.安装postgres 二.WAL日志解析工具 三.cdc增量数据同步方案 一.安装postgres (1)新增postgres用户组: groupadd postgres 新增postg ...
- 日志记录到字段变更_Wal日志解析工具开源: Walminer
作者简介 李传成: 瀚高软件内核研发工程师,主要研究方向为数据库的备份和恢复,对wal日志的原理和应用有较深的理解.自研了wal日志解析工具walminer.pg块恢复工具pg_lightool. 一 ...
- 日志分析工具Awstats实战之Apache篇-多站点日志分析
前面两篇都在讲述如何去部署nginx下的awstats日志分析工具,现在终于轮到apache.作为老牌的网页服务器,awstats对apache的支持非常完美,所以整个配置过程也是十分简单.因此,在这 ...
- mysql binlog 二进制_二进制日志-mysqlbinlog工具的使用
mysqlbinlog是一个客户端程序,可以审查binlog文件及中继日志文件的内容.既可以读取本地也可以读取远程的binlog文件,输出二进制日志的内容. mysqlbinlog常用的选项: -da ...
- 达梦数据库SQL日志分析工具Dmlog的使用
1.使用条件 运行环境预先安装Java环境:支持liunx和windows系统运行. 说明:推荐使用java1.8版本,linux最小化安装最少要安装打印服务组件,windows下不支持java1.6 ...
- MySQL连接远程数据库(使用Navicat工具)
一.首先需要进行数据库,利用Navicat进行本地连接数据库. 二.进行连接测试,登录本地数据库,让本不能远程共享的数据库,能够远程连接,远程连接顾名思义就是所有IP都能连接数据库,而非单单只限制为l ...
- 数据库增量同步/ETL工具
数据同步与订阅: 比较推荐的是阿里的Canal.Zendesk的Maxwell.以及来自社区的Python-Mysql-Replication ETL工具 kettle Kettle 是一款国外开源的 ...
最新文章
- 批改网作文提交时分析不出来_小学生作文写作从哪些方面提高呢?
- Java黑皮书课后题第10章:**10.25(新的字符串split方法)String类中的split方法会返回一个字符串数组,该数组是由分隔符分隔开的子串构成的
- 关于PHP程序员解决问题的能力
- Python基础——字符串、列表、元组
- python版本管理工具_Python版本管理工具与虚拟环境的介绍
- Determining if ip address is already in use for device eth0
- vs 2005應用2003框架
- linux的abrt目录满了,linux:abrt-cli list
- 【bug:鳄梨】【上线前修改其他bug急着提交造成的bug】
- VASP学习记录与总结
- 海康web3.0使用记录
- 唯一分解之Pollard-Rho算法
- nomad 服务编排_Nomad微服务的容器模式
- 在origin中画一条Y=x的自定义函数曲线
- Python字符串算法
- 【概率】COGS1487 麻球繁衍
- java开发wap_C#开发WAP程序
- 论PMI-ACP敏捷项目管理认证考前培训必要性
- 记录ab蓝学网新改版完成
- BigeMap 下载器使用教程
热门文章
- SAP Spartacus 服务器端渲染文件 server.js 和构建后的 main.js 对比
- 安装 SAP Cloud Application Programming SDK @sap/cds-dk 时报错以及解决方案
- SAP UI5里一个包含众多操作的OData请求url实例
- shadow dom一个最简单的例子
- Angular应用一个创建场景的问题分析
- Chrome扩展应用Angular state inspector的使用方法
- SAP Authenticator apk下载
- ABAP BSP的工具类CL_BSP_UTILITY
- when is completed field filled in the backend
- SAP UI5 sap.ui.core.OpenState.CLOSED Dialog open and close