首页 > 其他分享 >Canal

Canal

时间：2023-05-31 11:55:46浏览次数：41  
标签：Canal canal rw mysql 12 2020 MySQL

Canal

第 1 章 Canal 入门

1.1 什么是 Canal

阿里巴巴 B2B 公司，因为业务的特性，卖家主要集中在国内，买家主要集中在国外，所 以衍生出了同步杭州和美国异地机房的需求，从 2010 年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。

Canal 是用 Java 开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。 目前。Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 来处理获得 的相关数据。（数据库同步需要阿里的 Otter 中间件，基于 Canal）。

1.2 MySQL 的 Binlog

1.2.1 什么是 Binlog

MySQL 的二进制日志可以说 MySQL 最重要的日志了，它记录了所有的 DDL 和 DML(除 了数据查询语句)语句，以事件形式记录，还包含语句所执行的消耗的时间，MySQL 的二进 制日志是事务安全型的。

一般来说开启二进制日志大概会有 1%的性能损耗。二进制有两个最重要的使用场景:

  1. 其一：MySQL Replication 在 Master 端开启 Binlog，Master 把它的二进制日志传递给 Slaves 来达到 Master-Slave 数据一致的目的。
  2. 其二：自然就是数据恢复了，通过使用 MySQL Binlog 工具来使恢复数据。 二进制日志包括两类文件：二进制日志索引文件（文件名后缀为.index）用于记录所有 的二进制文件，二进制日志文件（文件名后缀为.00000*）记录数据库所有的 DDL 和 DML(除 了数据查询语句)语句事件。

1.2.2 Binlog 的分类

MySQL Binlog 的格式有三种，分别是 STATEMENT,MIXED,ROW。在配置文件中可以选择配 置 binlog_format= statement|mixed|row。三种格式的区别：

  1. statement：

    语句级，binlog 会记录每次一执行写操作的语句。相对 row 模式节省空 间，但是可能产生不一致性，比如“update tt set create_date=now()”，如果用 binlog 日志 进行恢复，由于执行时间不同可能产生的数据就不同。

    • 优点:节省空间
    • 缺点:有可能造成数据不一致。

  2. row：行级， binlog 会记录每次操作后每行记录的变化。

    • 优点:保持数据的绝对一致性。因为不管 sql 是什么，引用了什么函数，他只记录
    • 缺点:占用较大空间。

  3. mixed：statement 的升级版，一定程度上解决了，因为一些情况而造成的 statement 模式不一致问题，默认还是 statement，在某些情况下譬如：当函数中包含 UUID() 时；包含 AUTO_INCREMENT 字段的表被更新时；执行 INSERT DELAYED 语句时；用 UDF 时；会按照 ROW 的方式进行处理

    • 优点:节省空间，同时兼顾了一定的一致性。
    • 缺点:还有些极个别情况依旧会造成不一致，另外 statement 和 mixed 对于需要对 binlog 的监控的情况都不方便。

综合上面对比，Canal 想做监控分析，选择 row 格式比较合适。

1.3 Canal 的工作原理

1.3.1 MySQL 主从复制过程

  1. Master 主库将改变记录，写到二进制日志(Binary Log)中；
  2. Slave 从库向 MySQL Master 发送 dump 协议，将 Master 主库的 binary log events 拷贝到它的中继日志(relay log)；
  3. Slave 从库读取并重做中继日志中的事件，将改变的数据同步到自己的数据库。

​image​

1.3.2 Canal 的工作原理

很简单，就是把自己伪装成 Slave，假装从 Master 复制数据。

1.4 使用场景

  1. 原始场景： 阿里 Otter 中间件的一部分Otter 是阿里用于进行异地数据库之间的同步框架，Canal 是其中一部分。

    ​image​

  2. 常见场景 1：更新缓存

    ​image​

  3. 常见场景 2：抓取业务表的新增变化数据，用于制作实时统计（我们就是这种场景）

第 2 章 MySQL 的准备

2.1 创建数据库

​image​

2.2 创建数据表

CREATE TABLE user_info(
`id` VARCHAR(255),
`name` VARCHAR(255),
`sex` VARCHAR(255)
);

2.3 修改配置文件开启 Binlog

[atguigu@hadoop102 module]$ sudo vim /etc/my.cnf
server-id=1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=gmall-2021

​注意：binlog-do-db 根据自己的情况进行修改，指定具体要同步的数据库，如果不配置 则表示所有数据库均开启 Binlog​

2.4 重启 MySQL 使配置生效

sudo systemctl restart mysqld

到/var/lib/mysql 目录下查看初始文件大小 154

[atguigu@hadoop102 lib]$ pwd
/var/lib
[atguigu@hadoop102 lib]$ sudo ls -l mysql
总用量 474152
-rw-r-----. 1 mysql mysql 56 8 月 7 2020 auto.cnf
drwxr-x---. 2 mysql mysql 4096 9 月 25 2020 azkaban
-rw-------. 1 mysql mysql 1680 8 月 7 2020 ca-key.pem
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 ca.pem
drwxr-x--- 2 mysql mysql 4096 8 月 18 16:56 cdc_test
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 client-cert.pem
-rw-------. 1 mysql mysql 1676 8 月 7 2020 client-key.pem
drwxr-x---. 2 mysql mysql 4096 9 月 25 2020 gmall_report
-rw-r----- 1 mysql mysql 1085 12 月 1 09:12 ib_buffer_pool
-rw-r-----. 1 mysql mysql 79691776 12 月 13 08:45 ibdata1
-rw-r-----. 1 mysql mysql 50331648 12 月 13 08:45 ib_logfile0
-rw-r-----. 1 mysql mysql 50331648 12 月 13 08:45 ib_logfile1
-rw-r----- 1 mysql mysql 12582912 12 月 13 08:45 ibtmp1
drwxr-x--- 2 mysql mysql 4096 9 月 22 15:30 maxwell
drwxr-x---. 2 mysql mysql 4096 8 月 12 2020 metastore
drwxr-x---. 2 mysql mysql 4096 9 月 22 15:43 mysql
-rw-r-----. 1 mysql mysql 154 12 月 13 08:45 mysql-bin.000001
-rw-r----- 1 mysql mysql 19 12 月 13 08:45 mysql-bin.index
srwxrwxrwx 1 mysql mysql 0 12 月 13 08:45 mysql.sock
-rw------- 1 mysql mysql 5 12 月 13 08:45 mysql.sock.lock
drwxr-x---. 2 mysql mysql 4096 8 月 7 2020 performance_schema
-rw-------. 1 mysql mysql 1680 8 月 7 2020 private_key.pem
-rw-r--r--. 1 mysql mysql 452 8 月 7 2020 public_key.pem
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 server-cert.pem
-rw-------. 1 mysql mysql 1680 8 月 7 2020 server-key.pem
drwxr-x---. 2 mysql mysql 12288 8 月 7 2020 sys
drwxr-x--- 2 mysql mysql 4096 2 月 2 2021 test
[atguigu@hadoop102 lib]$

2.5 测试 Binlog 是否开启

  1. 插入数据

    INSERT INTO user_info VALUES('1001','zhangsan','male');

  2. 再次到/var/lib/mysql 目录下，查看 index 文件的大小

    [atguigu@hadoop102 lib]$ sudo ls -l mysql
总用量 474152
-rw-r-----. 1 mysql mysql 56 8 月 7 2020 auto.cnf
drwxr-x---. 2 mysql mysql 4096 9 月 25 2020 azkaban
-rw-------. 1 mysql mysql 1680 8 月 7 2020 ca-key.pem
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 ca.pem
drwxr-x--- 2 mysql mysql 4096 8 月 18 16:56 cdc_test
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 client-cert.pem
-rw-------. 1 mysql mysql 1676 8 月 7 2020 client-key.pem
drwxr-x---. 2 mysql mysql 4096 9 月 25 2020 gmall_report
-rw-r----- 1 mysql mysql 1085 12 月 1 09:12 ib_buffer_pool
-rw-r-----. 1 mysql mysql 79691776 12 月 13 08:45 ibdata1
-rw-r-----. 1 mysql mysql 50331648 12 月 13 08:45 ib_logfile0
-rw-r-----. 1 mysql mysql 50331648 12 月 13 08:45 ib_logfile1
-rw-r----- 1 mysql mysql 12582912 12 月 13 08:45 ibtmp1
drwxr-x--- 2 mysql mysql 4096 9 月 22 15:30 maxwell
drwxr-x---. 2 mysql mysql 4096 8 月 12 2020 metastore
drwxr-x---. 2 mysql mysql 4096 9 月 22 15:43 mysql
-rw-r-----. 1 mysql mysql 452 12 月 13 08:45 mysql-bin.000001
-rw-r----- 1 mysql mysql 19 12 月 13 08:45 mysql-bin.index
srwxrwxrwx 1 mysql mysql 0 12 月 13 08:45 mysql.sock
-rw------- 1 mysql mysql 5 12 月 13 08:45 mysql.sock.lock
drwxr-x---. 2 mysql mysql 4096 8 月 7 2020 performance_schema
-rw-------. 1 mysql mysql 1680 8 月 7 2020 private_key.pem
-rw-r--r--. 1 mysql mysql 452 8 月 7 2020 public_key.pem
-rw-r--r--. 1 mysql mysql 1112 8 月 7 2020 server-cert.pem
-rw-------. 1 mysql mysql 1680 8 月 7 2020 server-key.pem
drwxr-x---. 2 mysql mysql 12288 8 月 7 2020 sys
drwxr-x--- 2 mysql mysql 4096 2 月 2 2021 test
[atguigu@hadoop102 lib]$

2.6 赋权限

在 MySQL 中执行

mysql> set global validate_password_length=4;
mysql> set global validate_password_policy=0;
mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 
'canal'@'%' IDENTIFIED BY 'canal' ;

开启MySQL主从(WindowsAndDocKer)

Windows开启

  1. 进入到​mysql​的配置文件my.ini​(如果是默认安装路径的话就在

    ​C:\ProgramData\MySQL\MySQL Server 8.0​)

  2. 打开my.ini，增加配置

    # 主从复制
server-id=1 #设置server-id
log-bin=mysql-bin #开启二进制文件 
#同步的数据库，除此之外别的不同步（和下面binlog-ignore-db二选一）
#如果指定所有数据库的话，就不用写下面那条语句，默认所有库
binlog_do_db=lkj 
#不同步数据库，除此之外都同步
#binlog-ignore-db = information_schema 
#binlog-ignore-db = mysql

  3. 添加一个仅用于数据同步的账户

    create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;

  4. 重启​mysql

    #1.进入到MySQL服务：
mysql -uroot -p[passwor]
#2.停止 MySQL 服务。执行以下命令：
net stop mysql
#3.启动 MySQL 服务。执行以下命令：
net start mysql
#4.检查 MySQL 服务的状态是否正常。可以使用以下命令检查 MySQL 服务的状态：
sc query mysql

  5. 查看主从状态

    SHOW MASTER STATUS

    ​image​

Docker运行的mysql

  1. 打开mysql容器挂载的日志文件，我的在​/tmp/mysql/conf​目录:

    ​image​

  2. 修改文件：

    vi /tmp/mysql/conf/my.cnf

  3. 添加内容：

    log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima

    1. 配置解读：

      ​log-bin=/var/lib/mysql/mysql-bin​：设置binary log文件的存放地址和文件名，叫做mysql-bin

      ​binlog-do-db=heima​：指定对哪个database记录binary log events，这里记录heima这个库

    2. 最终效果：

      [mysqld]
skip-name-resolve
character_set_server=utf8
datadir=/var/lib/mysql
server-id=1000
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima

  4. 添加一个仅用于数据同步的账户

    create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;

  5. 重启mysql容器

    docker restart mysql

  6. 查看主从状态

    show master status;

第 3 章 Canal 的下载和安装

3.1 下载并解压 Jar 包

https://github.com/alibaba/canal/releases

我们直接/2.资料下的 canal.deployer-1.1.2.tar.gz 拷贝到/opt/sortware 目录下，然后解压 到/opt/module/canal 包下

​注意：canal 解压后是分散的，我们在指定解压目录的时候需要将 canal 指定上​

[atguigu@hadoop102 software]$ mkdir /opt/module/canal
[atguigu@hadoop102 software]$ tar -zxvf canal.deployer-1.1.2.tar.gz -C 
/opt/module/canal

3.2 修改 canal.properties 的配置

[atguigu@hadoop102 conf]$ pwd
/opt/module/canal/conf
[atguigu@hadoop102 conf]$ vim canal.properties
#################################################
######### common argument ############# 
#################################################
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file

​说明：​

这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111，修改 canal 的 输出 model，默认 tcp，改为输出到 kafka

多实例配置如果创建多个实例，通过前面 canal 架构，我们可以知道，一个 canal 服务 中可以有多个 instance，conf/下的每一个 example 即是一个实例，每个实例下面都有独立的 配置文件。默认只有一个实例 example，如果需要多个实例处理不同的 MySQL 数据的话，直 接拷贝出多个 example，并对其重新命名，命名和配置文件中指定的名称一致，然后修改 canal.properties 中的 canal.destinations=实例 1，实例 2，实例 3。

#################################################
######### destinations ############# 
#################################################
canal.destinations = example

3.3 修改 instance.properties

我们这里只读取一个 MySQL 数据，所以只有一个实例，这个实例的配置文件在 conf/example 目录下

[atguigu@hadoop102 example]$ pwd
/opt/module/canal/conf/example
[atguigu@hadoop102 example]$ vim instance.properties

  1. 配置 MySQL 服务器地址

    #################################################
## mysql serverId , v1.0.26+ will autoGen 
canal.instance.mysql.slaveId=20
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=hadoop102:3306

  2. 配置连接 MySQL 的用户名和密码，默认就是我们前面授权的 canal

    # username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName =test
# enable druid Decrypt database password
canal.instance.enableDruid=false

第 4 章 实时监控测试

4.1 TCP 模式测试

4.1.1 创建 gmall-canal 项目

​image​

​image​

4.1.2 在 gmall-canal 模块中配置 pom.xml

<dependencies>
 	<dependency>
		 <groupId>com.alibaba.otter</groupId>
		 <artifactId>canal.client</artifactId>
		 <version>1.1.2</version>
	</dependency>
	 <dependency>
		 <groupId>org.apache.kafka</groupId>
		 <artifactId>kafka-clients</artifactId>
		 <version>2.4.1</version>
 	</dependency>
</dependencies>

4.1.3 通用监视类 –CanalClient

  1. Canal 封装的数据结构

    • Canal数据结构

      ​image​

  2. 在 gmall-canal 模块下创建 com.atguigu.app 包，并在包下创建 CanalClient（java 代码） 代码如下：

    import com.alibaba.fastjson.JSONObject;
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 com.atguigu.constants.GmallConstants;
import com.atguigu.utils.MyKafkaSender;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;
public class CanalClient {
 public static void main(String[] args) throws 
InvalidProtocolBufferException {
 //1.获取 canal 连接对象
 CanalConnector canalConnector = 
CanalConnectors.newSingleConnector(new 
InetSocketAddress("hadoop102", 11111), "example", "", "");
 while (true) {
 //2.获取连接
 canalConnector.connect();
 //3.指定要监控的数据库
 canalConnector.subscribe("gmall.*");
 //4.获取 Message
 Message message = canalConnector.get(100);
 List<CanalEntry.Entry> entries = message.getEntries();
 if (entries.size() <= 0) {
 System.out.println("没有数据，休息一会");
 try {
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 } else {
 for (CanalEntry.Entry entry : entries) {
 //TODO 获取表名
 String tableName = 
entry.getHeader().getTableName();
 // TODO Entry 类型
 CanalEntry.EntryType entryType = 
entry.getEntryType();
 // TODO 判断 entryType 是否为 ROWDATA
 if 
(CanalEntry.EntryType.ROWDATA.equals(entryType)) {
 // TODO 序列化数据
 ByteString storeValue = entry.getStoreValue();
 // TODO 反序列化
 CanalEntry.RowChange rowChange = 
CanalEntry.RowChange.parseFrom(storeValue);
 //TODO 获取事件类型
CanalEntry.EventType eventType = 
rowChange.getEventType();
 //TODO 获取具体的数据
List<CanalEntry.RowData> rowDatasList = 
rowChange.getRowDatasList();
 //TODO 遍历并打印数据
for (CanalEntry.RowData rowData : rowDatasList) 
{
 List<CanalEntry.Column> beforeColumnsList = 
rowData.getBeforeColumnsList();
 JSONObject beforeData = new JSONObject();
for (CanalEntry.Column column : 
beforeColumnsList) {
 beforeData.put(column.getName(), 
column.getValue());
 }
 JSONObject afterData = new JSONObject();
List<CanalEntry.Column> afterColumnsList = 
rowData.getAfterColumnsList();
 for (CanalEntry.Column column : 
afterColumnsList) {
 afterData.put(column.getName(), 
column.getValue());
 }
System.out.println("TableName:" + tableName 
+
 ",EventType:" + eventType +
 ",After:" + beforeData +
 ",After:" + afterData);
 }
 }
 }
 }
 }
 }
}

  3. ‍

标签：Canal,canal,rw,mysql,12,2020,MySQL
From： https://www.cnblogs.com/lkj1223456/p/17439298.html

相关文章

  • canal+rabbitmq: Could not convert incoming message with content-type [null]
    SpringBoot整合Canal+RabbitMQ实现监听MySQL数据库同步更新Redis缓存，编写RabbitMQ消费端监听同步缓存。接收消息是字符串返回的是字节数据，eg：-30,-128,-100,-25,-126,-71,-27,-81,-71,-25,-126,-71,-30,-128,-99使用了jackson的converter后，报了如下的异常：Causedby:......
  • Java：SpringBoot整合Canal+RabbitMQ组合实现MySQL数据监听
    canal[kə’næl]，译意为水道/管道/沟渠，主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费目录一、MySQL设置二、启动Canal服务端三、通过Canal客户端消费数据四、通过RabbitMQ消费数据1、启动RabbitMQ2、修改canal配置3、消费RabbitMQ中的数据文档资料github：https......
  • Canal 安装 && 使用，监听Mysql数据库
    1.下载canal.deployer-1.1.5.tar.gz2.查看数据库是否开启binlog模式(必须)showvariableslike'log_%';log_bin属性值为ON,则binlog模式开启；为OFF则binlog模式关闭。若binlog模式关闭，则在MySQL配置文件my.cnf(my.ini)设置如下信息:[mysqld]log-bin=mysql-bin#开启binlogbi......
  • 使用canal同步mysql数据到elasticsearch
    官方去下载canal包https://github.com/alibaba/canal/releases/tag/canal-1.1.6分为deployer、admin、adapter三个模块。deployer是数据库数据同步服务端。adapter是适配同步到不同终端,可以是es，hbase，redis其它数据库等。admin是一个配置管理中心，但是吧又没有配置adapter的界面，a......
  • .Net 5 CanalSharp Mysql CDC (增量同步，捕获变更数据)Docker 系列之 Canal (CDC 增量
    CanalSharp阿里云的解决方案，需要两部分Canal 服务端要和Mysql连在一起（目前我是用docker部署的服务）另外一部分就是CanalSharp单独的客户端服务(.Net5服务)CanalSharp文档可以参考：https://canalsharp.azurewebsites.net/zh/安装服务，可以点击下边的连接。Docker系列之Canal......
  • .NET 6 Mysql Canal (CDC 增量同步,捕获变更数据) 案例版
    之前业务需要捕捉到业务数据增量部分，并对其进行宽表处理，这也是其中的一个技术方案，方案主要是用了CDC的技术。CDC全称是ChangeDataCapture，捕获变更数据，是一个比较广泛的概念，只要是能够捕获所有数据的变化，比如数据库捕获完整的变更日志记录增、删、改等，都可以称为CDC。该功能被......
  • 最简单的canal 1.1.6服务搭建方法
    前言：    因为在项目中集成了ElasticSearch，用于某些业务场景的搜素或筛选。这里关于ElasticSearch就不做介绍了，虽然解决了全文搜索的性能问题。但是当出现一些频繁更新的数据放置在ES就有点麻烦了。    这时候，一款能把MySQL数据即时同步到ElasticSearch的工具就显得......
  • canal数据库数据备份时，不同步删除分区动作。
        场景：我们有一大批业务数据，为减少数据库压力和提高使用效率，需要周期性的清理历史分区数据。但是这些数据并不是代表不要了。需要在另一个地方保存 传统数据库的主从复制功能，主要强调的是可以互为备份，强调高可用和可读写分离。但是在我们这个场景下并不适用......
  • docker 部署 canal
    1、docker部署canal/canal-server:v1.1.61.1、下载镜像dockerpullcanal/canal-server:v1.1.61.2、挂载properties配置文件1.2.1、先进行第一次运行，拷贝properties配置文件dockerrun-p11111:11111--namecanal-dcanal/canal-server:v1.1.61.2.2、拷贝运行后的容器......
  • CloudCanal x OceanBase 数据迁移同步优化
    简述CloudCanal去年支持OceanBase数据迁移同步能力后，随着使用用户增多以及问题反馈，近期对该能力进行了一轮较大规模的优化。本篇文章简要介绍这些优化点，以及未来该能力的演进方向。优化点大幅提升同步性能CloudCanal目前使用OceanBaseLogProxy做增量数据订阅，使用方式......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99