一、前言

canal 是阿里的一款开源项目，纯 Java 开发。基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了 MySQL(也支持 mariaDB)。

• 1、canal模拟mysql slave与mysql master的交互协议，伪装自己是一个mysql slave，向mysql master发送dump协议。
• 2、mysql master收到mysql slave（canal）发送的dump请求，开始推送binlog增量日志给slave(也就是canal)。
• 3、mysql slave（canal伪装的）收到binlog增量日志后,就可以对这部分日志进行解析，获取主库的结构及数据变更。

Canal总体架构：

二、部署准备

1、canal下载地址

• 下载地址：https://github.com/alibaba/canal/releases

分别下载：canal.admin、canal.deployer、canal.adapter

canal.adapter-1.1.5.tar.gz
canal.admin-1.1.5.tar.gz
canal.deployer-1.1.5.tar.gz

只有1.1.5以上版本才支持es7.x

其他依赖：

• JDK1.8
• MySQL：用于canal-admin存储配置和节点等相关数据
• Zookeeper

三、HA机制

整个 HA 机制的控制主要是依赖了zookeeper的两个特性：watcher、EPHEMERAL节点。canal的 HA 机制实现分为两部分，canal server 和 canal client分别有对应的实现。

canal server实现流程如下：

• 1、canal server 要启动某个 canal instance 时都先向 zookeeper 进行一次尝试启动判断 (实现：创建 EPHEMERAL 节点，谁创建成功就允许谁启动）。
• 2、创建 zookeeper 节点成功后，对应的 canal server 就启动对应的 canal instance，没有创建成功的 canal instance 就会处于 standby 状态。
• 3、一旦 zookeeper 发现 canal server A 创建的节点消失后，立即通知其他的 canal server 再次进行步骤1的操作，重新选出一个 canal server 启动instance。
• 4、canal client 每次进行connect时，会首先向 zookeeper 询问当前是谁启动了canal instance，然后和其建立链接，一旦链接不可用，会重新尝试connect。

为了减少对mysql dump的请求，不同server上的instance要求同一时间只能有一个处于running，其他的处于standby状态。

canal client实现流程

• 1、canal client 的方式和 canal server 方式类似，也是利用 zookeeper 的抢占EPHEMERAL 节点的方式进行控制。
• 2、为了保证有序性，一份 instance 同一时间只能由一个 canal client 进行get/ack/rollback操作，否则客户端接收无法保证有序。

四、集群部署

4.1、MySQL准备

4.1.1、开启binlog

MySQL的 my.cnf 中配置如下

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

注意：如果订阅的是mysql的从库，需求增加配置让从库日志也写到binlog里面

log_slave_updates=1

可以通过在 mysql 终端中执行以下命令判断配置是否生效：

show variables like 'log_bin';
show variables like 'binlog_format';

4.1.2、授权账号权限

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant：

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

4.2、部署canal-admin

4.2.1、作用

• 通过图形化界面管理配置参数。
• 动态启停 Server 和 Instance
• 查看日志信息

4.2.2、执行数据库脚本

执行 conf 目录下载的 canal_manager.sql 脚步，初始化所需的库表。

初始化SQL脚本里会默认创建canal_manager的数据库，建议使用root等有超级权限的账号进行初始化

4.2.3、配置修改

执行 vim conf/application.yml

server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:3306
  database: canal_manager
  username: canal
  password: canal
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin
  adminPasswd: admin

修改 address、database、username、password 四个参数

4.2.4、启停命令

启动

sh bin/startup.sh

停止

sh bin/stop.sh

4.2.5、使用

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

4.2.5.1、创建集群

配置 集群名称 与 ZK地址

配置 主配置，该配置为集群内的所有Server实例共享的

主要修改以下配置：

• canal.zkServers 配置zookeeper集群地址
• canal.instance.global.spring.xml 改为classpath:spring/default-instance.xml

4.2.5.2. 创建Server

配置项：

• 所属集群，可以选择为单机 或者 集群。一般单机Server的模式主要用于一次性的任务或者测试任务。
• Server名称，唯一即可，方便自己记忆。
• Server Ip，机器ip。
• admin端口，canal 1.1.4版本新增的能力，会在canal-server上提供远程管理操作，默认值11110。
• tcp端口，canal提供netty数据订阅服务的端口。
• metric端口， promethues的exporter监控数据端口 (未来会对接监控)

多台Server关联同一个集群即可形成主备HA架构

4.2.5.3. 创建Instance

每个 Instance 关联一个同步的数据源，如果有多个数据源需要同步则需要创建多个 实例

• 1、先填写实例名
• 2、选择刚刚创建的集群
• 3、载入模板配置

主要修改以下配置：

• canal.instance.master.address：配置要同步的数据库地址
• canal.instance.dbUsername：数据库用户名（需同步权限）
• canal.instance.dbPassword：数据库密码
• canal.instance.filter.regex：mysql 数据解析关注的表，Perl正则表达式.多个正则之间以逗号(,)分隔，转义符需要双斜杠()

canal.instance.filter.regex常见例子：

• 1、所有表：. or ...*
• 2、canal schema下所有表： canal..*
• 3、canal下的以canal打头的表：canal.canal.*
• 4、canal schema下的一张表：canal.test1
• 5、多个规则组合使用：canal..*,mysql.test1,mysql.test2 (逗号分隔)
  注意：此过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql，所以无法准确提取tableName进行过滤)

4.3. 部署canal-deployer

4.3.1. 作用

伪装成 MySQL 的从库，同步主库的binlog日志。
解析并结构化 binary log 对象。

4.3.2. 修改配置

执行 vim conf/canal_local.properties 修改配置项 canal.admin.manager 为canal-admin的地址

4.3.3. 启停命令

使用 local 配置启动

bin/startup.sh local

停止

bin/stop.sh

4.4. 部署canal-adapter

4.4.1. 作用

• 1、对接上游消息，包括kafka、rocketmq、canal-server
• 2、实现mysql数据的增量同步
• 3、实现mysql数据的全量同步
• 4、下游写入支持mysql、es、hbase等

4.4.2. 修改配置

注意：目前 adapter 是支持动态配置的，也就是说修改配置文件后无需重启，任务会自动刷新配置！

(1) 修改application.yml

执行 vim conf/application.yml 修改consumerProperties、srcDataSources、canalAdapters的配置

canal.conf:
  mode: tcp # kafka rocketMQ                # canal client的模式: tcp kafka rocketMQ
  flatMessage: true                         # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
  syncBatchSize: 1000                       # 每次同步的批数量
  retries: 0                                # 重试次数, -1为无限重试
  timeout:                                  # 同步超时时间, 单位毫秒
  consumerProperties:
    canal.tcp.server.host:                  # 对应单机模式下的canal
    canal.tcp.zookeeper.hosts: 127.0.0.1:2181 # 对应集群模式下的zk地址, 如果配置了canal.tcp.server.host, 则以canal.tcp.server.host为准
    canal.tcp.batch.size: 500               # tcp每次拉取消息的数量
  srcDataSources:                           # 源数据库
    defaultDS:                              # 自定义名称
      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true   # jdbc url 
      username: root                                            # jdbc 账号
      password: 121212                                          # jdbc 密码
  canalAdapters:                            # 适配器列表
  - instance: example                       # canal 实例名或者 MQ topic 名
    groups:                                 # 分组列表
    - groupId: g1                           # 分组id, 如果是MQ模式将用到该值
      outerAdapters:                        # 分组内适配器列表
      - name: es7                           # es7适配器
        mode: rest                          # transport or rest
        hosts: 127.0.0.1:9200               # es地址
        security.auth: test:123456          # 访问es的认证信息，如没有则不需要填
        cluster.name: my-es                 # 集群名称，transport模式必需配置
......

• 1、一份数据可以被多个group同时消费，多个group之间会是一个并行执行，一个group内部是一个串行执行多个outerAdapters，比如例子中logger和hbase。
• 2、目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息。

(2) conf/es7目录下新增映射配置文件

adapter将会自动加载 conf/es7 下的所有 .yml 结尾的配置文件
新增表映射的配置文件，如 sys_user.yml 内容如下：

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: sys_user
  _id: id
  upsert: true
  sql: "select id, username, 
        , case when sex = 0 then '男' else '女' end sex
        , case when is_del = 0 then '否' else '是' end isdel
      from sys_user"
  etlCondition: "where update_time>={}"
  commitBatch: 3000

• dataSourceKey：配置 application.yml 里 srcDataSources 的值
• destination：配置 canal.deployer 的 Instance 名
• groupId：配置 application.yml 里 canalAdapters.groups 的值
• _index：配置索引名
• _id：配置主键对应的字段
• upsert：是否更新
• sql：映射sql
• etlCondition：etl 的条件参数，全量同步时可以使用
• commitBatch：提交批大小

sql映射支持多表关联自由组合，但是有一定的限制:

• 1、主表不能为子查询语句
• 2、只能使用left outer join即最左表一定要是主表
• 3、关联从表如果是子查询不能有多张表
• 4、主sql中不能有where查询条件(从表子查询中可以有where条件但是不推荐，可能会造成数据同步的不一致，比如修改了where条件中的字段内容)
• 5、关联条件只允许主外键的'='操作不能出现其他常量判断比如: on a.role_id=b.id and b.statues=1
• 6、关联条件必须要有一个字段出现在主查询语句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主select语句中

Elastic Search的mapping 属性与sql的查询值将一一对应(不支持 select *), 比如: select a.id as _id, a.name, a.email as _email from user, 其中name将映射到es mapping的name field, _email将 映射到mapping的_email field, 这里以别名(如果有别名)作为最终的映射字段. 这里的_id可以填写到配置文件的 _id: _id映射

4.4.3. 启停命令

启动

bin/startup.sh

关闭

bin/stop.sh

五、监控

canal 默认已通过 11112 端口暴露同步相关的 metrics 信息，只需通过集成 prometheus 与 grafana 即可实现实时监控同步情况，效果图如下：

参考：
https://segmentfault.com/a/1190000039683805