Canal数据同步

一、Canal简介

早期阿里巴巴在杭州和美国部署机房，存在跨机房同步的业务需求，最初的实现是基于业务触发的方式 ，不是是很方便， 2010 年，逐步使用数据库日志解析方式取代，这由此衍生出了大量的数据库增量订阅和消费操作。在这种背景下Canal就出来了。

2014年左右，天猫双十一首次引入，用于解决大型促销活动MySQL 数据库的高并发读写问题。后来，在阿里内部得到了广泛应用和推广，并于2017年正式开源。

Github：https://github.com/alibaba/canal

Canal 组件是一个基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费，支持将增量数据投递到下游消费者（如 Kafka、RocketMQ 等）或者存储（如 Elasticsearch、HBase 等）的组件。

简单来说： Canal 感知到MySQL数据变动，然后解析变动数据，将变动数据发送到MQ或者同步到其他数据库，等待进一步业务逻辑处理。

1、Canal的工作原理

a、 MySQL主从复制原理

• MySQL master 将数据变更写入二进制日志binary log，简称Binlog。
• MySQL slave 将 master 的 binary log 拷贝到它的中继日志(relay log)
• MySQL slave 重放 relay log 操作，将变更数据同步到最新。

b、MySQL Binlog日志

MySQL 的Binlog可以说 MySQL 最重要的日志，它记录了所有的 DDL 和 DML语句，以事件形式记录。

MySQL默认情况下是不开启Binlog，因为记录Binlog日志需要消耗时间，官方给出的数据是有1%的性能损耗。

具体开不开启，开发中需要根据实际情况做取舍。一般来说，在下面两场景下会开启Binlog日志:

• MySQL 主从集群部署时，需要将在 Master 端开启 Binlog，方便将数据同步到Slaves中。

• 数据恢复了，通过使用 MySQL Binlog 工具来使恢复数据。

MySQL Binlog 的格式有三种，分别是 STATEMENT,MIXED,ROW。在配置文件中可以选择配

置 binlog_format= statement|mixed|row。

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

c、Canal 工作原理

• Canal 将自己伪装为 MySQL slave(从库) ，向 MySQL master (主库)发送dump 协议
• MySQL master(主库) 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
• Canal 接收并解析 Binlog 日志，得到变更的数据，执行后续逻辑

2、Canal运用场景

• 数据同步

  Canal 可以帮助用户进行多种数据同步操作，如实时同步 MySQL 数据到 Elasticsearch、Redis 等数据存储介质中。

• 数据库实时监控

  Canal 可以实时监控 MySQL 的更新操作，对于敏感数据的修改可以及时通知相关人员。

• 数据分析和挖掘

  Canal 可以将 MySQL 增量数据投递到 Kafka 等消息队列中，为数据分析和挖掘提供数据来源。

• 数据库备份

  Canal 可以将 MySQL 主库上的数据增量日志复制到备库上，实现数据库备份。

• 数据集成

  Canal 可以将多个 MySQL 数据库中的数据进行集成，为数据处理提供更加高效可靠的解决方案。

• 数据库迁移

  Canal 可以协助完成 MySQL 数据库的版本升级及数据迁移任务。

二、MySQL准备

canal需要开启数据库的Binlog日志支持。

• 创建项目所需要的库

  示例：创建woniu_db库并创建user表

  CREATE TABLE `user` (
    `id` bigint NOT NULL AUTO_INCREMENT,
    `name` varchar(255) DEFAULT NULL,
    `age` int DEFAULT NULL,
    PRIMARY KEY (`id`)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
  1
  2
  3
  4
  5
  6

• 修改mysql配置文件开启Binlog支持

  windows版
  [mysqld]
  server-id=1
  log-bin=./binlogs/mysql-bin.log
  binlog_format=row
  binlog-do-db=woniu_db  #要同步的数据库
  
  docker版
  [mysqld]
  server-id=1
  log-bin=/var/lib/mysql/mysql-bin.log  #挂载的日志路径，注意这个地址是docker容器里面的地址，不是宿主机里面的地址
  binlog-format=row
  binlog-do-db=woniu_db #要同步的数据库
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

  server-id：mysql 实例id，集群时用于区分实例

  lob-bin：binlog日志文件名称

  binlog_format：binlog日志数据保存格式

  binlog-do-db：指定开启binlog日志数据库。

  注意：一般根据情况进行指定需要同步的数据库，如果不配置则表示所有数据库均开启 Binlog。

• 校验Binlog生效

  重启MySQL服务，查看Binlog日志

  • 方式一

     use mysql;
     show  VARIABLES like 'log_bin'
    1
    2

    ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/3df630bdead94e239a59017d05d16a73.png#pic_center

  • 方式二：

    进入指定目录查看

    ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/f4ba9b6fbd274ad59b3191b1d1931698.png#pic_center

    insert into user(name, age) values('woniu', 18);
    insert into user(name, age) values('woniu', 18);
    insert into user(name, age) values('woniu', 18);
    1
    2
    3

• 创建mysql与canal专用连接帐号

  -- 创建帐号
  CREATE USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY '123456';
  -- 授权
  grant SELECT,REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%';
  -- 刷新配置
  FLUSH PRIVILEGES;
  1
  2
  3
  4
  5
  6

三、Canal安装与配置

• 拉取镜像

  docker pull canal/canal-server
  1

• 创建挂载目录

  mkdir -p /usr/local/canal/conf
  1

• 修改canal配置文件并上传到/usr/local/canal/conf中

  找到资料中的instance.properties

  canal.instance.mysql.slaveId=20
  
  # position info
  canal.instance.master.address=127.0.0.1:3306
  
  # username/password
  canal.instance.dbUsername=canal
  canal.instance.dbPassword=123456
  1
  2
  3
  4
  5
  6
  7
  8

  属性说明:

  • canal.instance.mysql.slaveId：使用canal 从阶段id

  • canal.instance.master.address：数据库ip端口

  • canal.instance.dbUsername：连接mysql账号

  • canal.instance.dbPassword：连接mysql密码

• 创建容器

  docker run -id -p 11111:11111 \
  --name canal \
  -v /usr/local/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
  -v /usr/local/canal/conf/canal.properties:/home/admin/canal-server/conf/canal.properties \
  canal/canal-server:latest
  1
  2
  3
  4
  5

四、springboot整合Canal

• 创建canal客户端项目

• 添加依赖

   <dependencies>
          <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-web</artifactId>
          </dependency>
       
       <!-- canal与springboot整合 -->
          <dependency>
              <groupId>top.javatool</groupId>
              <artifactId>canal-spring-boot-starter</artifactId>
              <version>1.2.1-RELEASE</version>
          </dependency>
  
          <!--实体类工具-->
          <dependency>
              <groupId>org.projectlombok</groupId>
              <artifactId>lombok</artifactId>
              <optional>true</optional>
          </dependency>
  
          <!--mysql驱动-->
          <dependency>
              <groupId>mysql</groupId>
              <artifactId>mysql-connector-java</artifactId>
          </dependency>
  
          <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-test</artifactId>
              <scope>test</scope>
          </dependency>
      </dependencies>
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32

• 修改yml

  server:
      port: 8080
  
  spring:
      application:  #设置当前项目名称,在四阶段很重要
          name: springboot-canal
      datasource: #数据源配置
          url: jdbc:mysql://192.172.0.14:3306/woniu_db?serverTimezone=Asia/Shanghai&characterEncoding=UTF-8&allowMultiQueries=true
          username: root
          password: 123456
  
  canal:
      server: 192.172.0.14:11111 #canal服务器地址
      destination: example #读取哪个canal中的哪个数据源配置
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• 创建监控处理类

  当对目标数据库进行insert ,update ,delete操作会自动触发该类

  @Component
  @CanalTable(value = "user")  //user为要监控的表名
  public class UserHandler implements EntryHandler<User> {
  
      @Override
      public void insert(User user) {
          System.err.println("添加：" + user);
      }
  
      @Override
      public void update(User before, User after) {
          System.err.println("修改改前：" + before);
          System.err.println("修改改后：" + after);
      }
  
      @Override
      public void delete(User user) {
          System.err.println("删除：" + user);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20

• 启动项目

  步骤：

  - 先启动canal服务器
  - 再启动项目
  - 修改user表
  - 观察结果
  1
  2
  3
  4

  启动项目后可以看到控制台在不停的获取数据库信息。当表中的数据发生改变时会自动获取到改变的数据

  

五、同类型技术

类型1：基于日志解析的数据同步组件

这类组件主要是通过解析数据库的Binlog（MySQL）或者Redo Log（Oracle）等日志文件，获取到数据库的增删改操作，并将这些操作记录下来。然后，这些操作记录可以被传输到另一个数据库中，以达到数据同步的目的。此类组件的代表产品包括阿里开源的Canal、腾讯云的DBSync等。

类型2：基于ETL的数据同步组件

ETL即Extract-Transform-Load，指的是从源系统抽取数据、对数据进行转换处理、最终加载到目标系统中。这类组件通常需要编写复杂的数据转换规则和数据映射关系，适用于数据结构变动频繁、数据量较大、数据来源较多的场景。代表产品包括阿里云的DataWorks、Informatica PowerCenter等。

类型3：基于CDC的数据同步组件

CDC（Change Data Capture）即变更数据捕获，它是一种数据同步技术，能够在实时或准实时地捕获数据库中的数据变更，并将其传输到另一个数据库中。CDC技术基于数据库的事务日志或重做日志实现，能够实现低延迟、高性能的数据同步。CDC组件的代表产品包括Oracle GoldenGate、IBM Infosphere Data Replication等。

类型4：基于消息队列的数据同步组件

这类组件通常将数据库中发生的变更操作抽象成一种数据结构，并通过消息队列将其发布到其他系统中进行处理，实现数据的异步传输和解耦。代表产品包括Apache Kafka、RabbitMQ等。

六、Canal常见面试题

问：Canal是什么？有哪些特性？

答：Canal是阿里巴巴开源的一款基于Netty实现的分布式、高性能、可靠的消息队列，在实时数据同步和数据分发场景下有着广泛的应用。Canal具有以下特性：支持MySQL、Oracle等数据库的日志解析和订阅；支持多种数据输出方式，如Kafka、RocketMQ、ActiveMQ等；支持支持数据过滤和格式转换；拥有低延迟和高可靠性等优秀的性能指标。

问：Canal的工作原理是什么？

答：Canal主要通过解析数据库的binlog日志来获取到数据库的增、删、改操作，然后将这些变更事件发送给下游的消费者。Canal核心组件包括Client和Server两部分，Client负责连接数据库，并启动日志解析工作，将解析出来的数据发送给Server；Server则负责接收Client发送的数据，并进行数据过滤和分发。Canal还支持多种数据输出器，如Kafka、RocketMQ、ActiveMQ等，可以将解析出来的数据发送到不同的消息队列中，以便进行进一步的处理和分析。

问：Canal的优缺点是什么？

答：Canal的优点主要包括：高性能、分布式、可靠性好、支持数据过滤和转换、跨数据库类型（如MySQL、Oracle等）等。缺点包括：使用难度较大、对数据库的日志产生一定的影响、不支持数据的回溯（即无法获取历史数据）等。

问：Canal在业务中有哪些应用场景？

答：Canal主要用于实时数据同步和数据分发场景，常见的应用场景包括：数据备份与灾备、增量数据抽取和同步、数据实时分析、在线数据迁移等。特别是在互联网大数据场景下，Canal已经成为了各种数据处理任务的重要工具之一。