canal官方github

超详细的Canal入门，看这篇就够了！

阿里开源Canal–①简介

centos7用tar包安装mysql

WINDOWS下MYSQL开启BINLOG

一.简介

canal 主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

基于日志增量订阅和消费包括

数据库镜像
数据库实时备份
索引构建和实时维护
业务 cache 刷新
带业务逻辑的增量数据处理

二.工作原理

MySQL 主备复制原理

MySQL master 将数据变更写入二进制日志（binary log，其中记录叫做二进制日志事件 binary log events，可以通过 show binlog events 进行查看）
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志（relay log）
MySQL slave 重放 relay log 中的事件，将数据变更反映它自己的数据

canal 工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave，向 MySQL master 发送 dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave （即 canal）
canal 解析 binary log 对象（原始为 byte 流）

三.MySQL开启log_bin

–> windows

打开 my.ini 文件，在 mysqld 下面添加

server-id=1
log_bin=D:\mysql\mysql-5.7.29-winx64\lib\mysql-bin
binlog-format =ROW

保存文件，重启 mysql 服务

net stop mysql
net start mysql

查看 binlog 状态：show variables like 'log_bin'

四.配置canal

4.1 基于Linux

（仍在探索，敬请期待！）

执行命令 vi conf/example/instance.properties

#################################################
## mysql serverId , v1.0.26+ will autoGen
canal.instance.mysql.slaveId=8# enable gtid use true/false
canal.instance.gtidon=false# position info 改成自己的数据库信息
canal.instance.master.address=10.187.129.134:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password 改成自己的数据库信息
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

执行命令

sh bin/startup.sh

查看日志

cat /home/tools/canal/canal/logs/canal/canal.log

2020-12-15 03:27:27.708 [Thread-6] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## stop the canal server
2020-12-15 03:27:27.760 [Thread-6] INFO  com.alibaba.otter.canal.deployer.CanalController - ## stop the canal server[192.168.133.112(192.168.133.112):11111]
2020-12-15 03:27:27.768 [Thread-6] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## canal server is down.
2020-12-15 03:27:34.256 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2020-12-15 03:27:34.317 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2020-12-15 03:27:34.335 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
2020-12-15 03:27:34.428 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.133.112(192.168.133.112):11111]
2020-12-15 03:27:36.135 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......

到目前为止 canal的服务端我们已经搭建好了但是到目前我们只是把数据库的binlog 拉到canal中，我们还得把数据用otter去消费

启动了canal的server之后,便是基于java的客户端搭建了。

首先在canalconf目录底下创建一个独立的文件夹（文件命名 idea_user_data），用于做额外的数据源配置。

然后再 idea_user_data 中创建 properties 文件，这个文件可以直接从 example 文件夹中复制，其余几份文件会自动生成

4.2 基于 Windows

从 canal 启动打印出的日志可以看出读取配置文件 canal.properties , 该配置文件里配置了 destination=example。因此会去 example 文件夹中读取配置文件

打开配置文件 conf/example/instance.properties，，配置信息如下：

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0# enable gtid use true/false
canal.instance.gtidon=false# position info
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

在 windows 下启动可能会报错：

要修改下启动的脚本 startup.bat：

再启动脚本。

这样就启动成功了

五.Java客户端操作

首先引入maven依赖

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.4</version>
</dependency>

在CannalClient类使用Spring Bean的生命周期函数afterPropertiesSet()：

package com.developer.canalspringdemo.deploy;import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;import java.net.InetSocketAddress;
import java.util.List;@Component
public class CanalClient implements InitializingBean {private final static int BATCH_SIZE = 1000;@Overridepublic void afterPropertiesSet() throws Exception {//创建连接CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");try {//打开链接connector.connect();//订阅数据库表，全部表connector.subscribe(".*\\..*");//回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿connector.rollback();while (true) {//获取指定数量的数据Message message = connector.getWithoutAck(BATCH_SIZE);//获取批量IDlong batchId = message.getId();//获取批量的数量int size = message.getEntries().size();//如果没有数据if (batchId == -1 || size == 0) {try {//线程休眠2秒Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}} else {//如果有数据，处理数据printEntry(message.getEntries());}//进行batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认connector.ack(batchId);}} catch (Exception e) {e.printStackTrace();} finally {connector.disconnect();}}/*** 打印canal server解析binlog获得的实体类信息*/private static void printEntry(List<Entry> entrys) {for (Entry entry : entrys) {if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {//开启/关闭事务的实体类型，跳过continue;}//RowChange对象，包含了一行数据变化的所有特征//比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等RowChange rowChage;try {rowChage = RowChange.parseFrom(entry.getStoreValue());} catch (Exception e) {throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);}//获取操作类型：insert/update/delete类型EventType eventType = rowChage.getEventType();//打印Header信息System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),eventType));//判断是否是DDL语句if (rowChage.getIsDdl()) {System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());}//获取RowChange对象里的每一行数据，打印出来for (RowData rowData : rowChage.getRowDatasList()) {//如果是删除语句if (eventType == EventType.DELETE) {printColumn(rowData.getBeforeColumnsList());//如果是新增语句} else if (eventType == EventType.INSERT) {printColumn(rowData.getAfterColumnsList());//如果是更新的语句} else {//变更前的数据System.out.println("------->; before");printColumn(rowData.getBeforeColumnsList());//变更后的数据System.out.println("------->; after");printColumn(rowData.getAfterColumnsList());}}}}private static void printColumn(List<Column> columns) {for (Column column : columns) {System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());}}
}

创建表：

CREATE TABLE `tb_commodity_info` (`id` varchar(32) NOT NULL,`commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名称',`commodity_price` varchar(36) DEFAULT '0' COMMENT '商品价格',`number` int(10) DEFAULT '0' COMMENT '商品数量',`description` varchar(2048) DEFAULT '' COMMENT '商品描述',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';

在控制台就可以看到如下信息：

================》; binlog[mysql-bin.000001:29066] , name[demo,tb_commodity_info] , eventType : CREATE
================》;isDdl: true,sql:CREATE TABLE `tb_commodity_info` (`id` varchar(32) NOT NULL,`commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名称',`commodity_price` varchar(36) DEFAULT '0' COMMENT '商品价格',`number` int(10) DEFAULT '0' COMMENT '商品数量',`description` varchar(2048) DEFAULT '' COMMENT '商品描述',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表'

新增一条数据到表中：

INSERT INTO tb_commodity_info VALUES('3e71a81fd80711eaaed600163e046cc3','叉烧包','3.99',3,'又大又香的叉烧包，老人小孩都喜欢');

结果

================》; binlog[mysql-bin.000001:29763] , name[demo,tb_commodity_info] , eventType : INSERT
id : 3e71a81fd80711eaaed600163e046cc3    update=true
commodity_name : 叉烧包    update=true
commodity_price : 3.99    update=true
number : 3    update=true
description : 又大又香的叉烧包，老人小孩都喜欢    update=true

六.Canal Web UI

官网

注意：

执行 conf/canal_manager.sql 文件
赋予用户 canal 最高权限

主界面

访问 127.0.0.1:8089，默认密码：admin/123456

Canal使用入门学习相关推荐

Canal Java 入门与使用
Canal Java 入门与使用 0.前言在如今时代数据是最关键的,大的数据量势必会影响用户体验,在我的日常开发中遇到了一个类似的问题,由于数据库中数据量过于庞大造成查询效率低,决定采用redi ...
python速成要多久2019-8-28_2019最全Python入门学习路线，不是我吹，绝对是最全
近几年Python的受欢迎程度可谓是扶摇直上,当然了学习的人也是愈来愈多.一些学习Python的小白在学习初期,总希望能够得到一份Python学习路线图,小编经过多方汇总为大家汇总了一份Python学 ...
MAYA 2022基础入门学习教程
流派:电子学习| MP4 |视频:h264,1280×720 |音频:AAC,48.0 KHz 语言:英语+中英文字幕(根据原英文字幕机译更准确)|大小解压后:3.41 GB |时长:4.5小时包含 ...
3dmax Vray建筑可视化入门学习教程
面向初学者的3Ds Max Vray最佳Archviz可视化课程从安装到最终图像的一切都将从头开始教授,不需要任何经验大小解压后:3.25G 时长4h 6m 1280X720 MP4 语言:英语+ ...
Blender 3.0基础入门学习教程 Introduction to Blender 3.0
成为Blender通才,通过这个基于项目的循序渐进课程学习所有主题的基础知识. 你会学到什么教程获取:Blender 3.0基础入门学习教程 Introduction to Blender 3.0- ...
Maya游戏角色绑定入门学习教程 Game Character Rigging for Beginners in Maya
准备好开始为游戏制作自己的角色动画了吗? 你会学到什么了解Maya的界面优化并准备好你的模型,为游戏做准备了解关节以及如何使用它们来构建健壮的角色骨骼,以便在任何游戏引擎中制作动画了解IK和F ...
三维地形制作软件 World Machine 基础入门学习教程
<World Machine课程>涵盖了你需要的一切,让你有一个坚实的基础来构建自己的高质量的电影或视频游戏地形. 你会学到什么为渲染或游戏开发创建高分辨率.高细节的地形. 基于Worl ...
Blender3.0动画制作入门学习教程 Learn Animation with Blender (2021)
要求下载并安装Blender.免费下载和免费用于任何目的. 描述加入我的动画课程. 在本课程中,我将从头开始讲述在Blender中创建动画场景的过程. 从第一步到最终渲染.在这个课程中,我们将使用 ...
UE5真实环境设计入门学习教程
大小解压后:4.69G 时长4h 30m 1280X720 MP4 语言:英语+中英文字幕(根据原英文字幕机译更准确) 虚幻引擎5–面向初学者的真实环境设计 Unreal Engine 5 – Rea ...

Canal使用入门学习