cancl对接mysql安装及使用

「爱情、让人受尽委屈。」 2022-05-24 09:13 296阅读 0赞

**1、环境准备**  
    （1）安装mysql，要求5.0版本以上,本环境中安装路径为/opt/mysql/。  
    （2）安装环境jdk，要求1.7以上。  
    （3）使用root用户启动mysql

>service mysqld start     //启动mysql
        >service mysqld status   //检查mysql启动状态

（4）查看canal相关资料  
        https://github.com/alibaba/canal  
        里面有包括canal的文档，server端 client端的 例子 源码包等等。  
  
**2、cancl概述**  
    canal是应阿里巴巴存在杭州和美国的双机房部署，存在跨机房同步的业务需求而提出的。  
    早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。ps. 目前内部使用的同步，已经支持mysql5.x和oracle部分版本的日志解析。  
    基于日志增量订阅&消费支持的业务：  
        数据库镜像  
        数据库实时备份  
        多级索引 (卖家和买家各自分库索引)  
        search build  
        业务cache刷新  
        价格变化等重要业务消息

keyword：数据库同步，增量订阅&消费。

**3、工作原理**

(1) mysql主备复制实现

**![70][]**

从上层来看，复制分成三步：

1.  master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
2.  slave将master的binary log events拷贝到它的中继日志(relay log)；
3.  slave重做中继日志中的事件，将改变反映它自己的数据。

（2）canal的工作原理：

**![70 1][]**

原理相对比较简单：

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

**4、cancl部署**  
  
（1）开启mysql的binlog功能，并配置binlog模式为row。  
    在my.cnf 加入如下：

vi /etc/my.cnf
        log-bin=mysql-bin #添加这一行就ok  ，表示打开二进制日志功能。
        binlog-format=ROW #选择row模式  
        server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复。

（2）在mysql中 配置canal数据库管理用户，配置相应权限（replication权限）

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

（3）下载canal  
    https://github.com/alibaba/canal/releases    
    解压到相应的文件夹下

>cd /opt/cancl/
     >tar -zxvf canal.deployer-1.0.26-SNAPSHOT.tar.gz -C /opt/cancl/

修改配置 instance.properties

vim canal/conf/example/instance.properties

#################################################  
    ## mysql serverId  
    #查看mysql的slaveId，需要修改成自己的真实的slaveId
    canal.instance.mysql.slaveId = 1  
     
    # 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 = test  
    canal.instance.connectionCharset = UTF-8  
     
    # table regex  
    canal.instance.filter.regex = .*\\..*  
     
    #################################################

然后cd到bin目录  启动和停止canal-server

>./startup.sh

停止

>./stop.sh

验证启动状态，查看log文件  
    >cat canal/log/canal/canal.log

2018-05-24 10:52:08.655 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.  
     2018-05-24 10:52:08.854 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.137.131:11111]  
     2018-05-24 10:52:09.209 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

上述日志信息显示启动canal成功

**5、运行canal实例**

**  **  IDE ：intellij IEDA

系统：Windows10

mvn ：Apache Maven 3.5.3

（1）建立实例maven工程

在win系统打开cmd，cd到对应的workspace目录中，创建mvn工程。

mvn archetype:create -DgroupId=com.alibaba.otter -DartifactId=canal.sample

![70 2][]

这里直接回车，选择maven版本为1.1

![70 3][]

（2）使用IDE打开创建好的mvn工程，添加pom.xml依赖：

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

（3）更新依赖 mvn install

（4）创建ClientSample.java文件进行连接linux对应机器和对应的IP

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

（5）运行java实例，启动后看到控制端信息：

empty count : 1  
    empty count : 2  
    empty count : 3  
    empty count : 4

（6）在mysql中操作一张表，触发数据库变更：

mysql> insert into test5 values(5,6);
    mysql> insert into test5 values(8,9);
    mysql> update test5 set a=19 where a=5;

（7）查看启动的后台控制端口，已经有相应信息打印：

================> binlog[replicalog.000044:1689] , name[test,test5] , eventType : UPDATE
    -------> before
    a : 5    update=false
    b : 6    update=false
    -------> after
    a : 19    update=true
    b : 6    update=false

[70]: /images/20220524/96dcbe4e63fe4ba9a29e809feee53a3f.png
[70 1]: /images/20220524/dcbf7bf0cdec436eaf45e4d3f8196ee3.png
[70 2]: /images/20220524/7290560da3fa40058fd46e6f50a2df28.png
[70 3]: /images/20220524/29b7f64160f64bfaa7ab50e97fc9e771.png