springboot整合Cannal

川长思鸟来 2023-09-25 18:08 5阅读 0赞

### 简介 ###

canal主要用途是对MySQL数据库增量日志进行解析，提供增量数据的订阅和消费，简单说就是可以对MySQL的增量数据进行实时同步，支持同步到MySQL、Elasticsearch、HBase等数据存储中去。  
canal会模拟MySQL主库和从库的交互协议，从而伪装成MySQL的从库，然后向MySQL主库发送dump协议，MySQL主库收到dump请求会向canal推送binlog，canal通过解析binlog将数据同步到其他存储中去。  
![在这里插入图片描述][1ae0bd6e9ead4aabad9ccc6183ea2d40.png]

### 准备工作 ###

#### 组件下载 ####

我们可以根据自己的需要下载canal的各个组件

*  canal-server（canal-deploy）：可以直接监听MySQL的binlog，把自己伪装成MySQL的从库，只负责接收数据，并不做处理。
    
     *  canal-adapter：相当于canal的客户端，会从canal-server中获取数据，然后对数据进行同步，可以同步到MySQL、Elasticsearch和HBase等存储中去。
     *  canal-admin：为canal提供整体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操作界面，方便更多用户快速和安全的操作。

因为我们这篇文章只需要一个canal-service，所以看看它的建立方式，根据自己的需要去建立即可：

*  [直接下载启动文档][Link 1]
 *  [通过docker启动文档][docker]
 *  [直接传递给mq文档][mq]

#### 修改mysql配置 ####

由于canal是通过订阅MySQL的binlog来实现数据同步的，所以我们需要开启MySQL的binlog写入功能，并设置binlog-format为ROW模式，我的配置文件为/mydata/mysql/conf/my.cnf，改为如下内容即可；

[mysqld]
    ## 设置server_id，同一局域网中需要唯一
    server_id=101 
    ## 指定不需要同步的数据库名称
    binlog-ignore-db=mysql  
    ## 开启二进制日志功能
    log-bin=mall-mysql-bin  
    ## 设置二进制日志使用内存大小（事务）
    binlog_cache_size=1M  
    ## 设置使用的二进制日志格式（mixed,statement,row）
    binlog_format=row  
    ## 二进制日志过期清理时间。默认值为0，表示不自动清理。
    expire_logs_days=7  
    ## 跳过主从复制中遇到的所有错误或指定类型的错误，避免slave端复制中断。
    ## 如：1062错误是指一些主键重复，1032错误是因为主从数据库数据不一致
    slave_skip_errors=1062

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

配置完成后需要重新启动MySQL，重启成功后通过如下命令查看binlog是否启用；

show variables like '%log_bin%'

+---------------------------------+-------------------------------------+
    | Variable_name                   | Value                               |
    +---------------------------------+-------------------------------------+
    | log_bin                         | ON                                  |
    | log_bin_basename                | /var/lib/mysql/mall-mysql-bin       |
    | log_bin_index                   | /var/lib/mysql/mall-mysql-bin.index |
    | log_bin_trust_function_creators | OFF                                 |
    | log_bin_use_v1_row_events       | OFF                                 |
    | sql_log_bin                     | ON                                  |
    +---------------------------------+-------------------------------------+

再查看下MySQL的binlog模式；

show variables like 'binlog_format%';

+---------------+-------+
    | Variable_name | Value |
    +---------------+-------+
    | binlog_format | ROW   |
    +---------------+-------+

### 客户端监听 ###

#### 引入Cannal依赖 ####

在pom.xml文件中加入以下依赖：

<dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.4</version>
    </dependency>
    
    // 数据库依赖
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.17</version>
    </dependency>

#### 设置连接 ####

在application.yml文件中添加以下配置：

# 需要同步数据的MySQL地址
    canal.instance.master.address=127.0.0.1:3306
    canal.instance.master.journal.name=
    canal.instance.master.position=
    canal.instance.master.timestamp=
    canal.instance.master.gtid=
    # 用于同步数据的数据库账号
    canal.instance.dbUsername=canal
    # 用于同步数据的数据库密码
    canal.instance.dbPassword=canal
    # 数据库连接编码
    canal.instance.connectionCharset = UTF-8
    # 需要订阅binlog的表过滤正则表达式
    canal.instance.filter.regex=.*\\..*

#### 创建实体 ####

（忽略表的创建）

@TableName("t_user")
    @Data
    public class User {
        
        @TableId(type = IdType.AUTO)
        @Id
        private Long id;
        @Column(name = "name")
        private String name;
    }

#### 创建监听类 ####

实现CanalEventListener接口，实现监听器方法，例如：

@CanalTable("t_user")
    @Component
    public class UserHandler implements EntryHandler<User> {
        
     
        @Override
        public void insert(User user) {
        
            System.out.println("新增用户");
            System.out.println("user = " + user);
        }
     
        @Override
        public void update(User before, User after) {
        
            System.out.println("修改用户");
            System.out.println("修改用户before：" + before);
            System.out.println("修改用户after：" + after);
        }
     
        @Override
        public void delete(User user) {
        
            System.out.println("删除用户user = " + user);
        }
     
    }

这样的话，就可以实现监听某张表的变动了。

### 参考地址 ###

官方地址：[https://github.com/alibaba/canal/wiki][https_github.com_alibaba_canal_wiki]

[1ae0bd6e9ead4aabad9ccc6183ea2d40.png]: https://img-blog.csdnimg.cn/1ae0bd6e9ead4aabad9ccc6183ea2d40.png
[Link 1]: https://github.com/alibaba/canal/wiki/QuickStart
[docker]: https://github.com/alibaba/canal/wiki/Docker-QuickStart
[mq]: https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
[https_github.com_alibaba_canal_wiki]: https://github.com/alibaba/canal/wiki