Canal

canal介绍

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

• 数据库镜像
• 数据库实时备份
• 索引构建和实时维护(拆分异构索引、倒排索引等)
• 业务 cache 刷新
• 带业务逻辑的增量数据处理

当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

主从同步

如上是mysql的主从同步

1.Master 主库将改变记录，写到binary log中

2.Slave 从库向 MySQL Master 发送 dump 协议，将 Master 主库的 binary log events 拷到它的relay log

3.Slave 从库读取并重做relay log中的事件，将改变的数据同步到自己的数据库

其他优点

在线上环境，对于读远大于写的场景，一般都是一主多从，主库进行写操作，从库进行读操作，然后主库内容更新再同步给从库

binary log主要作⽤是记录数据库中表的更改，它只记录改变数据的sql，不改变数据的sql不会写⼊，⽐如select语句⼀般不会被记录，因为他们不会对数据产⽣任何改动

relay log的结构和binlog非常相似，只不过他多了一个master.info和relay-log.info的文件，master.info记录了上一次读取到master同步过来的binlog的位置，以及连接master和启动复制必须的所有信息。relay-log.info记录了文件复制的进度，下一个事件从什么位置开始，由sql线程负责更新

canal原理

canal其实本质就是将自己伪装成一个slave，同步主库的binary log

1.canal 模拟 mysql slave 的交互协议，伪装自己为 mysql slave，向 mysql master 发送 dump 协议

2.ysql master 收到 dump 请求，开始推送 binary log 给 slave (也就是 canal)

3.canal 解析 binary log 对象

binlog日志

canal部署

web页面部署

1.前期准备

如果为OFF，则需要配置开启。

window版5.7.43，在安装目录找到my.ini

[mysqld]

log-bin= H:\\binlog\mysql-bin
server-id=100
binlog_format = ROW

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

下载发布和admin

配置UI界面

1.执行脚本

SQL脚本位置：conf/canal_manager.sql

脚本中是单独创建了一个数据库canal_manager，我们也可以不单独创建数据库，与其他业务表放在一起

修改配置文件

conf/application.yml

修改内容：

数据库账户密码

点击bin/startup.bat

2.集群设置

在集群管理中，点击新建集群，新建一个集群配置。这里的zk地址就是服务端集群的zk地址。

Canal的集群原理是指如何将多个Canal节点组成一个集群,以提高系统的可用性和扩展性。

服务端部署

1.配置更改

mysql connector

canal-admin中默认提供的驱动器是mysql5.0的，因此要替换一下（canal-deployer解压目录的lib文件夹）自行查看是否替换

在conf目录下，备份canal.properties，将canal_local.properties重命名为canal.properties。

2.启动

在主配置中统一修改集群下的服务端deployer的配置文件，也可以统一的查看集群下的服务端实例。

在【Instance管理】中新增一个实例，也就是我们之前在服务端的conf文件夹下配置的，每一个子文件夹就代表了一个实例，每个实例都有自己的instance.properties配置文件，这里新增的实例就是这个配置文件：

常用配置参数

29行: canal.serverMode = rocketMQ (这里设置了rocketMQ所以需要配置162行有关的内容，不同选择修改不同地方的配置)

86行: canal.instance.tsdb.enable = false

127行: canal.mq.flatMessage = true（flatMessage 为true 生产到mq的消息就是json的, 否则就是protobuf二进制的）

集成Spirngboot

<!-- 这里要考虑版本兼容问题，这是与jdk8匹配的jar版本 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    <version>2021.0.4.0</version>
</dependency>
 
<!-- Canal 相关 -->
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.protocol</artifactId>
    <version>1.1.7</version>
    <exclusions>
        <exclusion>
            <groupId>com.alibaba.fastjson2</groupId>
            <artifactId>fastjson2</artifactId>
        </exclusion>
        <exclusion>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
        </exclusion>
    </exclusions>
</dependency>

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

# rocket配置 
spring:
  cloud:
    stream:
      function:
        definition: twoConsumer # 方法定义
      bindings:
        # 发送必须配置此处
        producerInfo-out-0:
          destination: test_topic # topic消息主题
        # 配置channel消息通道 （接收必须配置此处）
        twoConsumer-in-0:
          destination: test-top # topic消息主题
          group: consumer-group # 消费者组
      rocketmq:
        binder:
          name-server: 192.168.152.130:9876 # rocketmq服务地址
        # 配置消息通道独特属性（仅适用于rocketmq）
        bindings:
          # 发送必须配置此处
          # 配置channel消息通道（生产者：[functionName]-out-[index]，消费者：[functionName]-in-[index]）
          producerInfo-out-0:
            producer:
              group: consumer-group #生产者的组名称（一般与消费者组相同），用于负载均衡
              sync: true # 是否开启同步发送

再配置监听者-具体到MQ中配置