使用Canal组件订阅MySQL binlog数据增量

向右看齐 2024-03-26 14:34 60阅读 0赞

#### 一、简介 ####

Canal是一款强大的开源组件，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

**1. canal 工作原理**

*  canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
 *  MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
 *  canal 解析 binary log 对象(原始为 byte 流)  
    ![工作原理简略图][33c648a8da4a4dadb2d2bad31c36febc.png]  
    **2. 基于日志增量订阅和消费的业务包括**
 *  数据库镜像
 *  数据库实时备份
 *  索引构建和实时维护(拆分异构索引、倒排索引等)
 *  业务 cache 刷新
 *  带业务逻辑的增量数据处理  
    当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

#### 二、环境准备 ####

**1. 开启MySQL的Binlog日志写入功能**  
找到MySQL服务器上的MySQL配置文件`my.ini`，增添以下配置：

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

注意：修改完配置之后，一定要重启MySQL！！！  
以Windows系统为例：  
打开桌面，右键点击“此电脑”—> 管理 —> 服务和应用程序 —>服务，找到MySQL服务，右键，选择重启。  
![在这里插入图片描述][d39ecaf8af574f3db11db239e9df3f2e.png]

**2. 在MySQL中创建一个名为 ‘canal’ 的用户，并授予该用户在MySQL所有数据库中的所有表上进行 SELECT、复制从服务器和复制客户端操作的权限**:

CREATE USER canal IDENTIFIED BY 'canal';  -- 创建了一个名为 canal 的用户，并设置其密码为 'canal'
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    FLUSH PRIVILEGES;

第2条授权语句是指：授予了 canal 用户在所有数据库中的所有表上进行 SELECT、复制从服务器和复制客户端操作的权限。其中`\*.*`表示所有数据库中的所有表，`'canal'@'%'`表示在所有的 IP 地址上都可以使用这个用户进行操作。这个授权语句可以根据实际需求进行修改，比如指定特定的数据库和表，或者只在某个特定的 IP 地址上授权等。

**3. 下载Canal组件**  
下载 canal, 访问[release页面][release]，以1.1.6版本为例：  
![在这里插入图片描述][36ed4f098e78461aac89a56355d86a9f.png]  
将下载下来的`canal.deployer-1.1.6.tar.gz`安装包进行解压缩：  
![在这里插入图片描述][f6152f8f87a44fcab3ec26739cf5f364.png]  
打开`/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 = .\*\\\\..\*

可以根据实际需要修改这些配置项，配置完后将配置文件保存为 .properties 文件，并将文件放置在 Canal 的 `conf` 目录下，然后启动 Canal 即可开始监控 MySQL 实例的 `binlog`。

**4. 启动Canal服务**  
Linux：

sh bin/startup.sh

Windows：  
双击`startup.sh`脚本文件。

注意：如果在Windows中启动canal下的`startup.sh`后出现闪退，且`/canal/logs/canal_stdout.log`日志为：

Unrecognized VM option 'AggressiveOpts'
    Error: Could not create the Java Virtual Machine.
    Error: A fatal exception has occurred. Program will exit.

这个问题可能是因为 `startup.sh` 脚本中使用了在 Windows 平台下不支持的 JVM 参数，导致 JVM 启动失败。

解决办法：  
可以尝试修改 `startup.sh` 文件，将其中的 JVM 参数改为 Windows 下支持的格式。具体来说，可以将 `startup.sh` 中的 `-XX:+AggressiveOpts` 参数改为 `-XX:-UseAggressiveOpts` 或者去掉这个参数。

#### 三、通过Java客户端监听 ####

方式一：拉取canal源代码

https://github.com/alibaba/canal.git

直接打开`com.alibaba.otter.canal.example.SimpleCanalClientTest`类。

方式二、自行创建项目工程

1.  新建maven工程，修改pom文件，并引入maven依赖：

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

1.  编写简单的Canal客户端demo：

package com.alibaba.otter.canal.sample;
    import java.net.InetSocketAddress;
    import java.util.List;
    
    
    import com.alibaba.otter.canal.client.CanalConnectors;
    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;
    
    public class SimpleCanalClientExample {
        
    	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 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.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<Column> columns) {
        
    	    for (Column column : columns) {
        
    	        System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    	    }
    	}
    }

1.  运行client。

#### 四、触发数据库变更 ####

1.  在MySQL中，选择一个数据表进行更新操作：  
    ![在这里插入图片描述][6b8ce7257a044e74bcbb99262f9fc3eb.png]
2.  client客户端控制台输出如下信息：  
    ![在这里插入图片描述][c0fa408aa2cd4db0ad46c31e8f6640b8.png]

[33c648a8da4a4dadb2d2bad31c36febc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/40f980bfc9d14a458b724a0c017e383c.png
[d39ecaf8af574f3db11db239e9df3f2e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/52e9a621ddd04ca7868dad7ca9d3c4d9.png
[release]: https://github.com/alibaba/canal/releases
[36ed4f098e78461aac89a56355d86a9f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/a3bc8539a55747499586e07785b56b77.png
[f6152f8f87a44fcab3ec26739cf5f364.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/848bdbcb899f44938f71e0a60e21d93f.png
[6b8ce7257a044e74bcbb99262f9fc3eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/746c475e99474627ab9d4ac596023c2a.png
[c0fa408aa2cd4db0ad46c31e8f6640b8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/9d9fd6a20ce740549ce01f22674cb72a.png