canal快速入门(小白必看)

拼搏现实的明天。 2023-01-18 02:21 157阅读 0赞

### 文章目录 ###

*   *  一：canal简介
     *  二：工作原理
     *  三：环境准备
     *  四：CanalClient编写
     *  五：测试canal
     *  六：canal相关文档
     *  七：开心一刻

## 一：canal简介 ##

`canal [kə'næl]`，译意为水道/管道/沟渠，**主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70]

通过上图可以看出，canal支持 `Mysql`、`Kafka`、`ElasticSearch`、`RocketMq` 等，常见的**业务场景**：**数据同步**。**解析binlog，同步数据到ElasticSearch、RocketMq 等**。有了它，我们不需要再专门写一个框架来完成这件事情，完全交给`canal`就足够了。

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

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

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

## 二：工作原理 ##

**MySQL主备复制原理**

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

**canal 工作原理**

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

## 三：环境准备 ##

**3.1 准备一套Mysql主从集群环境**

既然`canal`是模拟成一台`slave`服务，来拉取`binlog`。那肯定得有一套Mysql集群环境（Mysql 主从集群环境），如果有童鞋还没有此环境，可以参考下 [Mysql主从集群搭建][Mysql]

**3.2 开启binlog**

先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

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

**3.3 创建canal用户，用来数据同步**

授权 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;

**3.4 canal下载**

再来下载个`canal`包，[点击下载][Link 1]，我用的是最新版1.1.5。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 1]  
上传至linux 上的`Mysql master` 节点上，如下：  
![在这里插入图片描述][20210430110723250.png]  
解压tar

mkdir -p /usr/local/canal
     tar -zxvf canal.deployer-1.1.5.tar.gz  -C  /usr/local/canal

目录如下：  
![在这里插入图片描述][20210430111310525.png]  
**3.5 查看mysql master 节点状态**

**3.6 修改canal配置**

vi conf/example/instance.properties

## mysql serverId , v1.0.26+ will autoGen
    # canal.instance.mysql.slaveId=0
    
    # enable gtid use true/false
    canal.instance.gtidon=false
    
    # 注意：标☆☆☆的都是必须修改的，同时也给了注释
    
    # position info 改成自己的数据库信息
    # ☆☆☆ canal.instance.master.address：数据库master节点地址
    # ☆☆☆ canal.instance.master.journal.name：也就是master中的File
    # ☆☆☆ canal.instance.master.position：也就是master中的Position
    
    canal.instance.master.address=192.168.29.134:3306
    canal.instance.master.journal.name=mysql-bin.000002
    canal.instance.master.position=4936
    canal.instance.master.timestamp=
    canal.instance.master.gtid=
    
    # rds oss binlog
    canal.instance.rds.accesskey=
    canal.instance.rds.secretkey=
    canal.instance.rds.instanceId=
    
    # table meta tsdb info
    canal.instance.tsdb.enable=true
    #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
    #canal.instance.tsdb.dbUsername=canal
    #canal.instance.tsdb.dbPassword=canal
    
    #canal.instance.standby.address =
    #canal.instance.standby.journal.name =
    #canal.instance.standby.position =
    #canal.instance.standby.timestamp =
    #canal.instance.standby.gtid=
    
    # username/password
    # ☆☆☆ canal.instance.dbUsername：数据库账号
    # ☆☆☆ canal.instance.dbPassword：数据库密码
    
    canal.instance.dbUsername=canal
    canal.instance.dbPassword=canal
    canal.instance.connectionCharset = UTF-8
    # enable druid Decrypt database password
    canal.instance.enableDruid=false
    #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
    
    # table regex
    canal.instance.filter.regex=.*\\..*

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

**3.7 启动**

sh bin/startup.sh

**3.8 查看启动日志**

cat logs/canal/canal.log

2021-04-28 14:37:13.323 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
    2021-04-28 14:37:13.381 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
    2021-04-28 14:37:13.402 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
    2021-04-28 14:37:13.507 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.29.134(192.168.29.134):11111]
    2021-04-28 14:37:17.415 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......

出现上面说明启动成功

## 四：CanalClient编写 ##

**4.1 创建maven项目，并引入依赖：**

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

**4.2 创建类CanalClientExample**

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.CanalEntry;
    import com.alibaba.otter.canal.protocol.Message;
    
    import java.net.InetSocketAddress;
    import java.util.List;
    
    public class CanalClientExample {
    
        public static void main(String args[]) {
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.29.134",
                    11111), "example", "", "");
            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<CanalEntry.Entry> entrys) {
            for (CanalEntry.Entry entry : entrys) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    continue;
                }
    
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
    
                CanalEntry.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));
    
                //打印DDL语句
                if (rowChage.getIsDdl()) {
                    System.out.println("================> sql " + rowChage.getSql());
                }
    
    
                //打印DML表字段信息
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == CanalEntry.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<CanalEntry.Column> columns) {
            for (CanalEntry.Column column : columns) {
                System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            }
        }
    
    }

## 五：测试canal ##

要注意下启动顺序：  
mysql -> canal server -> canalClient

**5.1 启动CanalClient**

启动Canal Client后，可以从控制台从看到类似消息：

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

此时代表当前数据库无变更数据

**5.2 在Mysql master 上执行DDL建表语句**

mysql> CREATE TABLE `canal_test_table` ( 
    `ID` int(11) NOT NULL AUTO_INCREMENT, 
    `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  
    PRIMARY KEY (`ID`) 
    ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;
    
    Query OK, 0 rows affected (0.05 sec)
    
    mysql> insert into canal_test_table (id,x) values(null,now());   
    Query OK, 1 row affected (0.01 sec)

**5.3 CanalClient输出信息**

================> binlog[mysql-bin.000002:6757] , name[mytest,canal_test_table] , eventType : CREATE
    ================> sql CREATE TABLE `canal_test_table` ( `ID` int(11) NOT NULL AUTO_INCREMENT, `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  PRIMARY KEY (`ID`) ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8
    empty count : 1
    empty count : 2
    empty count : 3
    empty count : 4
    empty count : 5
    empty count : 6
    ================> binlog[mysql-bin.000002:7242] , name[mytest,canal_test_table] , eventType : INSERT
    ID : 3    update=true
    X : 2021-04-28 15:06:42    update=true
    empty count : 1
    empty count : 2
    empty count : 3

以上就是canal的安装及测试。

## 六：canal相关文档 ##

[canal 消息投递给 kafka/RocketMQ 相关文档][canal _ kafka_RocketMQ]  
[canal 的 docker 模式快速启动][canal _ docker]  
[canal Client API 相关文档][canal Client API]

## 七：开心一刻 ##

> 麻雀和乌鸦一起摆龙门阵。  
> ➤麻雀说：你是啥子鸟哦？  
> ➤乌鸦说：我是凤凰噻！  
> ➤麻雀：哪有你龟儿子这么黑的凤凰哦？  
> ➤乌鸦：你晓得个铲铲，老子是烧锅炉的凤凰噻。。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 2]  
如果觉得不错，帮忙点个赞，您的点赞将是我的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70]: /images/20221021/089fc39f982a414688817a5cbd22cf52.png
[Mysql]: https://blog.csdn.net/weixin_42740530/article/details/116238260
[Link 1]: https://github.com/alibaba/canal/releases
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 1]: /images/20221021/54872886448b497bb656fd5ff58aee07.png
[20210430110723250.png]: /images/20221021/1aa4fbf463f041b792a3e7b73300e337.png
[20210430111310525.png]: /images/20221021/5b86681fdd104aadb9e065acf7556586.png
[canal _ kafka_RocketMQ]: https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
[canal _ docker]: https://github.com/alibaba/canal/wiki/Docker-QuickStart
[canal Client API]: https://github.com/alibaba/canal/wiki/ClientAPI
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc0MDUzMA_size_16_color_FFFFFF_t_70 2]: /images/20221021/a620ac3cacf84f1e86d426ee09d936ec.png