canal 入门实践

Canal 的组成部分

简单来说，Canal 会将自己伪装成 MySQL 从节点（Slave），并从主节点（Master）获取 Binlog，解析和贮存后供下游消费端使用。Canal 包含两个组成部分：服务端和客户端。服务端负责连接至不同的 MySQL 实例，并为每个实例维护一个事件消息队列；客户端则可以订阅这些队列中的数据变更事件，处理并存储到数据仓库中。下面我们来看如何快速搭建起一个 Canal 服务。

MySQL 服务器

本人使用的mysql版本为5.6.33

1. 修改 mysql 配置文件 my.conf

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

   • 注意：针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步

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

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

  重启下mysql服务器 service mysql restart

canal服务器

• 下载 canal, 访问 release 页面 , 选择需要的包下载, 如以 1.1.5 版本为例

  wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz

• 解压缩

  mkdir /tmp/canal
  tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal
  

  • 解压完成后，进入 /tmp/canal 目录，可以看到如下结构

    drwxr-xr-x 2 jianghang jianghang  136 2013-02-05 21:51 bin
    drwxr-xr-x 4 jianghang jianghang  160 2013-02-05 21:51 conf
    drwxr-xr-x 2 jianghang jianghang 1.3K 2013-02-05 21:51 lib
    drwxr-xr-x 2 jianghang jianghang   48 2013-02-05 21:29 logs
    

• 配置修改

  vi conf/example/instance.properties
  

  ## mysql serverId
  canal.instance.mysql.slaveId = 1234
  #position info，需要改成自己的数据库信息
  canal.instance.master.address = 127.0.0.1:3306 
  canal.instance.master.journal.name = 
  canal.instance.master.position = 
  canal.instance.master.timestamp = 
  #canal.instance.standby.address = 
  #canal.instance.standby.journal.name =
  #canal.instance.standby.position = 
  #canal.instance.standby.timestamp = 
  #username/password，需要改成自己的数据库信息
  canal.instance.dbUsername = canal  
  canal.instance.dbPassword = canal
  canal.instance.defaultDatabaseName =
  canal.instance.connectionCharset = UTF-8
  #table regex
  canal.instance.filter.regex = .\*\\\\..\*

  • canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1
  • 如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false

• 启动

  sh bin/startup.sh
  

• 查看 server 日志

  vi logs/canal/canal.log
  
  

  2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
  2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]
  2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
  

• 查看 instance 的日志

  vi logs/example/example.log
  

  2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
  2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
  2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
  2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....
  

   

• 关闭

  sh bin/stop.sh
  
  canal客户端
  
  

  Canal 客户端

  java maven配置

          
              com.alibaba.otter
              canal.client
              1.1.5
          
          
              com.alibaba.otter
              canal.protocol
              1.1.5
          

  package canal;
  
  import com.alibaba.otter.canal.client.CanalConnector;
  import com.alibaba.otter.canal.client.CanalConnectors;
  import com.alibaba.otter.canal.protocol.CanalEntry.*;
  import com.alibaba.otter.canal.protocol.Message;
  
  import java.net.InetSocketAddress;
  import java.util.List;
  
  public class ClientSample {
      
      public static void main(String args[]) {
          // 创建链接
          CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                  11111), "example", "", "");
          int batchSize = 1000;
          int emptyCount = 0;
          try {
              connector.connect();
              connector.subscribe(".*\\..*");
              connector.rollback();
              int totalEmptyCount = 1200;
              while (emptyCount < totalEmptyCount) {
                  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) {
                      }
                  } 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(List 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(List columns) {
          for (Column column : columns) {
              System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
          }
      }
  
  }

   运行之后，往数据库表中插入一条数据了。打印出的日志如下
  

  ================> binlog[mysql-bin.000042:694] , name[huagang,users] , eventType : INSERT
  id : 14 update=true
  username : AA update=true
  passowrd : AA update=true
  create_date : 2022-01-24 14:40:16 update=true

引用：https://github.com/alibaba/canal/wiki/QuickStart

         https://blog.csdn.net/u013837825/article/details/120838244