canal使用记录

喜欢ヅ旅行 2021-06-10 20:38 926阅读 0赞

canal是阿里巴巴的来源项目。我们可以通过配置binlog实现数据库监控，得到数据库表或者数据的更新信息。  
参考我的文档前先去官网看下，可能已经支持更高版本的MySQL了

## 1. 查看官方开源项目 ##

[https://github.com/alibaba/canal][https_github.com_alibaba_canal]

## 2. 下载最新的canal.deployer-XXXX-SNAPSHOT.tar.gz ##

[https://github.com/alibaba/canal/releases][https_github.com_alibaba_canal_releases]

## 3. 查看wiki ##

ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)  
所以需要安装mysql5.7以下的版本稳妥些

## 4. 去MySQL官网下载mysql并安装 ##

## 5. 查看mysql安装目录 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70][]

## 6. 复制一个my-default.ini改名叫my.ini ##

修改对应位置：  
添加：

![20200427152511288.png][]

#添加这一行就ok
    log-bin=mysql-bin
    #选择row模式
    binlog-format=ROW
    #配置mysql replaction需要定义，不能和canal的slaveId重复
    server_id=1
    character-set-server=utf8
    collation-server=utf8_general_ci

添加：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 1][]

[mysql]
    default-character-set = utf8
    [mysql.server]
    default-character-set = utf8
    [mysqld_safe]
    default-character-set = utf8
    [client]
    default-character-set = utf8

## 7. 重启mysql ##

.6\bin>sc delete mysql
    \bin>net stop mysql
    \bin>mysqld --install mysql --defaults-file="C:\Program Files\MySQL\MySQL Server 5.6\my.ini"

## 8. 查看是否开启binlog ##

show variables like'log_%';

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 2][]

On：表示已开启

## 9. 创建数据库canal用户 ##

![20200427161238416.png][]

官网是%，%是对所有非本地主机授权，不包括localhost。由于我们是在windows本机上做，所以需要配置为localhost.

## 10. 修改canal-deploy-> conf\\example里的instance.properties ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 3][]

这两个新添的配置可以注解调，还不太明白具体的用处，可能是testdb 的 账号密码

canal.instance.tsdb.dbUsername=canal
    canal.instance.tsdb.dbPassword=canal

注：这里的slaveId=1234不能和my.ini的一样

## 11. 在cmd下启动canal-deploy ##

![20200427161431923.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 4][]

如果没有报错那就是启动成功了

## 12. 创建canal-client服务 ##

Pom或者gradle。主要依赖：

compile group: 'org.jetbrains', name: 'annotations', version: '13.0'
    compile group: 'com.alibaba.otter', name: 'canal.client', version: '1.0.25'

## 13. 编写客户端类 ##

package com.shao.demo.canalclient;
    import java.net.InetSocketAddress;
    import java.util.List;
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.common.utils.AddressUtils;
    import com.alibaba.otter.canal.protocol.Message;
    import com.alibaba.otter.canal.protocol.CanalEntry.Column;
    import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
    import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
    import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
    import com.alibaba.otter.canal.client.*;
    import org.jetbrains.annotations.NotNull;
    /**
     * @author zhiqi.shao
     * @Date 2018/6/4 18:29
     */
    public class ClientSample {
     
     public static void main(String args[]) {
     // 创建链接
    CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
     11111), "example", "", "");
     int batchSize = 1000;
     int emptyCount = 0;
      try {
        connector.connect();
       connector.subscribe(".*\\..*");
       connector.rollback();
       int totalEmtryCount = 1200;
       while (emptyCount < totalEmtryCount) {
       Message message = connector.getWithoutAck(batchSize); 
        // 获取指定数量的数据 
        long batchId = message.getId();
        int size = message.getEntries().size();
       if (batchId == -1 || size == 0) {
           emptyCount++;
          System.out.println("empty count : " + emptyCount);
          try {
             Thread.sleep(1000);
             } catch (InterruptedException e) {
             e.printStackTrace();
            }
        } else {
            emptyCount = 0;
           // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
           printEntry(message.getEntries());
       }
        connector.ack(batchId); // 提交确认 // connector.rollback(batchId); // 处理失败, 回滚数据 }
        System.out.println("empty too many times, exit");
     } finally {
         connector.disconnect();
     }
     }
     
     private static void printEntry(@NotNull List<Entry> entrys) {
     for (Entry entry : entrys) {
     if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
     continue;
     }
     
     RowChange rowChage = null;
              try {
     rowChage = RowChange.parseFrom(entry.getStoreValue());
     } catch (Exception e) {
     throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
               e);
     }
     
     EventType eventType = rowChage.getEventType();
     System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
     eventType));
     
     for (RowData rowData : rowChage.getRowDatasList()) {
         if (eventType == EventType.DELETE) {
            printColumn(rowData.getBeforeColumnsList());
         } else if (eventType == EventType.INSERT) {
           printColumn(rowData.getAfterColumnsList());
         } else {
            System.out.println("-------> before");
            printColumn(rowData.getBeforeColumnsList());
           System.out.println("-------> after");
           printColumn(rowData.getAfterColumnsList());
        }
       }
      }
     }
     
     private static void printColumn(@NotNull List<Column> columns) {
         for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
         }
       }
    }

## 14. 启动客户端 ##

use canal_test;
    CREATE TABLE user (
     uid INT(4) PRIMARY KEY NOT NULL AUTO_INCREMENT,
     name VARCHAR(10) NOT NULL
    );
     
    insert into user (name) values('shaoshao');

![20200427161645549.png][]

## 15. 关于update ##

rowData.getAfterColumnsList()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 5][]

## 16. 常见错误 ##

1.  服务端：com.alibaba.otter.canal.parse.exception.CanalParseException: can't find start position for example  
    是由于你改了配置文件，导致meta.dat 中保存的位点信息和数据库的位点信息不一致；导致canal抓取不到数据库的动作；  
    解决方法：删除meta.dat删除，再重启canal，问题解决；
2.  客户端：java.lang.OutOfMemoryError: Java heap space  
    canal消费端挂了太久，在zk对应conf下节点的  
    /otter/canal/destinations/test\_db/1001/cursor 位点信息是很早以前，导致重启canal时，从很早以前的位点开始消费，导致canal服务器内存爆掉
3.  服务端ERROR c.a.otter.canal.server.netty.handler.SessionHandler - something goes wrong with channel:\[id: 0x0191aafd, /192.168.10.68:49502 => /192.168.10.68:11111\], exception=java.io.IOException:  
    当客户端停掉后，canal服务端会报此异常  
    客户端：com.alibaba.otter.canal.protocol.exception.CanalClientException: something goes wrong with reason: something goes wrong with channel:\[id: 0x01311037, /192.168.10.68:52086 => /192.168.10.68:11111\], exception=com.alibaba.otter.canal.server.exception.CanalServerException: destination:example should start first  
    当服务端停掉或者重启中，客户端连不上就会抛出此异常。场景修改了服务点的配置文件此时服务端会重启，客户端就会报次异常

## 17. 消费过滤 ##

canalConnector.subscribe("canal\_test\\..*");//客户端只消费canal\_test库的数据变化  
subscribe(filter)方法；有的话，filter需要和instance.properties的canal.instance.filter.regex一致，否则subscribe的filter会覆盖instance的配置，如果subscribe的filter是.*\\..\*，那么相当于你消费了所有的更新数据。

## 18. 配置 ##

1.  【instance.properties配置定义优先级高于canal.properties】
2.  修改了服务端配置文件，服务器会自动重启

## 19. 关于HA机制的设计 ##

canal server: 为了减少对mysql dump的请求，不同server上的instance要求同一时间只能有一个处于running，其他的处于standby状态.  
canal client: 为了保证有序性，一份instance同一时间只能由一个canal client进行get/ack/rollback操作，否则客户端接收无法保证有序。

## 20. Canal的工作原理 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 6][]

原理相对比较简单：

1.  canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
2.  mysql master收到dump请求，开始推送binary log给slave(也就是canal)
3.  canal解析binary log对象(原始为byte流)

## 21. 链接方式（参考：[http://www.importnew.com/25189.html][http_www.importnew.com_25189.html]） ##

### 1. HA配置架构图 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 7][]

### 2. 单连 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 8][]

### 3. 两个client+两个instance+1个mysql ###

当mysql变动时，两个client都能获取到变动

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 9][]

### 4. 一个server+两个instance+两个mysql+两个client ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 10][]

### 5. instance\*\*\*\*的standby配置 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 11][]

Standby：备库

## 22. 总结 ##

这里总结了一下Canal的一些点，仅供参考：

1.  原理：模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议；mysql master收到dump请求，开始推送binary log给slave(也就是canal)；解析binary log对象(原始为byte流)
2.  重复消费问题：在消费端解决。
3.  采用开源的open-replicator来解析binlog
4.  canal需要维护EventStore，可以存取在Memory, File, zk
5.  canal需要维护客户端的状态，同一时刻一个instance只能有一个消费端消费
6.  数据传输格式：protobuff
7.  支持binlog format 类型:statement, row, mixed. 多次附加功能只能在row下使用，比如otter
8.  binlog position可以支持保存在内存，文件，zk中
9.  instance启动方式：rpc/http; 内嵌
10. 有ACK机制
11. 无告警，无监控，这两个功能都需要对接外部系统
12. 方便快速部署。

## 23. 我调试成功的代码地址 ##

[https://gitee.com/zhiqishao/canal-client][https_gitee.com_zhiqishao_canal-client]

[https_github.com_alibaba_canal]: https://github.com/alibaba/canal
[https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70]: /images/20210527/3e15d4fef06b4c9b97fb8cd038433868.png
[20200427152511288.png]: /images/20210527/7c64f4cac2fd4c46a340508bb8fdf9d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 1]: /images/20210527/a6a1b5cb194441cba5cfdbee10228b94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 2]: /images/20210527/37902fb6cb4a4dc48ade3de5b40a203b.png
[20200427161238416.png]: /images/20210527/531f8d2cb7744376bbd3e73ed4317fbb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 3]: /images/20210527/2646b6d9ae7f437a9e6fe7552cde67d0.png
[20200427161431923.png]: /images/20210527/12dfb171c1cf4b5bb4202072a8ace44b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 4]: /images/20210527/20701efad126432fa54a7842dc4c276a.png
[20200427161645549.png]: /images/20210527/5248435520f84debb9a4b2c9e963da36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 5]: /images/20210527/00ceb2d63e8e486e8f6829efca60f6f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 6]: /images/20210527/f43874bb175b47ea97968fc0e932ec49.png
[http_www.importnew.com_25189.html]: http://www.importnew.com/25189.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 7]: /images/20210527/c4509c9567aa4dfdb78939fe37686e59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 8]: /images/20210527/581b64c49a8e4c18a1ff7d1c94f08337.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 9]: /images/20210527/016c8e4a737c43599d1bb8c26ddc0ec5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 10]: /images/20210527/fc301f6b72b843b291c57e56770ce78e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3o2OTE4Mzc4Nw_size_16_color_FFFFFF_t_70 11]: /images/20210527/3fd23431b026430db443d050aa260d6b.png
[https_gitee.com_zhiqishao_canal-client]: https://gitee.com/zhiqishao/canal-client