canal同步binlog实战

canal原理介绍

canal简介

picture.image canal [kǝ'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

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

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

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

canal原理

picture.image 主从复制原理：

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

所以canal模拟mysql slave也实现了自己的binlog同步，原理如下：

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

首先去下载安装相关canal组件

这里可以参考官网：https://github.com/alibaba/canal

安装完canal服务，然后按照文档介绍去配置相应的参数，就可以启动canal的服务了，由于官网介绍的比较细，大家自行去安装和配置，这里着重介绍canal同步binlog代码实践。

canal-go订阅和消费canal binlog

下载canal-go客户端，运行里面samples下的main.go，当然可能报错，将路径替换正确就OK了。(有问题欢迎私信) 部分订阅消费代码如下：

  
func main() {  
  
 // 192.168.199.17 替换成你的canal server的地址  
 // example 替换成-e canal.destinations=example 你自己定义的名字  
 connector := client.NewSimpleCanalConnector("127.0.0.1", 11113, "", "", "example", 60000, 60*60*1000)  
 err := connector.Connect()  
 if err != nil {  
  log.Println(err)  
  os.Exit(1)  
 }  
  
 err = connector.Subscribe(".*\\..*")  
 if err != nil {  
  log.Println(err)  
  os.Exit(1)  
 }  
  
 for {  
  message, err := connector.Get(100, nil, nil)  
  if err != nil {  
   log.Println(err)  
   os.Exit(1)  
  }  
  batchId := message.Id  
  if batchId == -1 || len(message.Entries) <= 0 {  
   time.Sleep(300 * time.Millisecond)  
   fmt.Println("===没有数据了===")  
   continue  
  }  
  printEntry(message.Entries)  
 }  
}

我们可以对canal配置文件中已经配置的数据库表进行增删改，看看我们程序是否能显示相关binlog日志

增加

  
mysql> insert into canal(`id`, `name`) values(1,"hah");  
Query OK, 1 row affected (0.04 sec)  
  
mysql>

打印日志：

  
===没有数据了===  
================> binlog[mysql-bin.000002 : 2069],name[test,canal], eventType: INSERT  
id : 1  update= true  
name : hah  update= true  
sex :   update= true  
age :   update= true  
amount :   update= true  
email :   update= true  
occur_time : 2021-09-15 09:47:09  update= true

更新

  
mysql> update canal set name="hello" where id=1;  
Query OK, 1 row affected (0.04 sec)  
Rows matched: 1  Changed: 1  Warnings: 0  
  
mysql>

打印日志如下：

  
===没有数据了===  
================> binlog[mysql-bin.000002 : 2348],name[test,canal], eventType: UPDATE  
-------> before  
id : 1  update= false  
name : hah  update= false  
sex :   update= false  
age :   update= false  
amount :   update= false  
email :   update= false  
occur_time : 2021-09-15 09:47:09  update= false  
-------> after  
id : 1  update= false  
name : hello  update= true  
sex :   update= false  
age :   update= false  
amount :   update= false  
email :   update= false  
occur_time : 2021-09-15 09:47:09  update= false  
===没有数据了===

删除

  
mysql> delete from canal where id=1;  
Query OK, 1 row affected (0.03 sec)  
  
mysql>

打印日志如下：

  
================> binlog[mysql-bin.000002 : 2643],name[test,canal], eventType: DELETE  
id : 1  update= false  
name : hello  update= false  
sex :   update= false  
age :   update= false  
amount :   update= false  
email :   update= false  
occur_time : 2021-09-15 09:47:09  update= false  
===没有数据了===

总结

通过canal-go客户端可以订阅和消费canal server，那么同样我们可以配置kafka，mysql，es，mq等也可以去订阅和消费数据，项目中比较常见的就是将mysql数据同步到es中提供搜索查询能力，而这最好的方式就是通过canal中间件来做，减少人工的维护成本，提供工作效率。

关注：堆栈future