SpringBoot集成Canal实现MySQL实时同步数据到Redis

MySQL增量数据同步利器Canal环境搭建流程

软件环境

• JDK17.0.12

• canal-server1.1.7

• canal-client1.1.7

• MySQL5.7

• IDEA2024.2.0.2

我们先看Canal1.1.7源码对应的项目结构

1、基于源码编译打包

# 源码下载地址
https://github.com/alibaba/canal
# 执行以下命令，打包编译
mvn clean install -Dmaven.test.skip=true

2、搭建canal-admin
2.1 安装Ebean enhancer插件
安装和编译时启用，如下图

2.2 创建数据库
创建canal_manager数据库和执行对应脚本，脚本在\canal-canal-1.1.7\admin\admin-web\src\main\resources目录下

2.3 修改配置文件
按照下图修改数据库配置信息

2.4 启动canal管理后台
基于源码启动管理后台

访问以下地址 http://127.0.0.1:8089/

默认用户名及密码 admin/123456

3、搭建canal-server
3.1 canal-server端配置
使用canal_local.properties的配置覆盖canal.properties

# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name = 

3.2 启动canal-server
基于源码启动canal-server，启动成功后，在管理后台查看对应server

3.3 修改MySQL配置信息
对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

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

授权 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 修改instance.properties

## mysql serverId
canal.instance.mysql.slaveId = 1234
#position info，需要改成自己的数据库信息
canal.instance.master.address = 192.168.0.104: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 = .\*\\\\..\*

3.5 启动instance

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

• 数据库镜像

• 数据库实时备份

• 索引构建和实时维护(拆分异构索引、倒排索引等)

• 业务 cache 刷新

• 带业务逻辑的增量数据处理

软件环境

• JDK17.0.12

• SpringBoot3.4.0

• redisson-spring-boot-starter3.38.1

• Redis6.x

• Canal-Server1.1.7

• Canal-Admin1.1.7

• Canal-Client1.1.7

• IDEA2024.2.0.2

项目结构

1、项目搭建
1.1 Canal项目依赖项

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>cn.itbeien</groupId>
        <artifactId>springboot3-labs-master</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <artifactId>springboot-canal</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <canal.client-version>1.1.7</canal.client-version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>${canal.client-version}</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.protocol</artifactId>
            <version>${canal.client-version}</version>
        </dependency>
       <!-- <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>-->
    </dependencies>
</project>

1.2 配置信息

#application.properties
server.port=2001
server.servlet.context-path=/canal
#canal
canal-monitor-mysql.host=192.168.0.105
#canal.properties  canal.port
canal-monitor-mysql.port=11111

spring.data.redis.host=192.168.0.104
spring.data.redis.port=6379
spring.data.redis.password=Rootpwd20240809
# redis数据库编号
spring.data.redis.database=8

1.3 代码实现
canal实时从mysql获取数据，同步到分布式缓存redis，完成业务缓存刷新

package cn.itbeien.canal.util;


import cn.itbeien.canal.entity.SysUser;
import com.alibaba.fastjson.JSON;
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 lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

@Slf4j
@Component
public class CanalUtil {


    @Value("${canal-monitor-mysql.host}")
    String canalMonitorHost;

    @Value("${canal-monitor-mysql.port}")
    Integer canalMonitorPort;

    @Autowired
    private RedisClient redisClient;

    private final static int BATCH_SIZE = 10000;

    /**
     * 启动服务
     */
    // @Async("TaskPool")
    public void startMonitorSQL() {
        while (true) {
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalMonitorHost, canalMonitorPort), "0.104", "", "");
            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++;
                        log.info("empty count :{} " , emptyCount);
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                        }
                    } else {
                        emptyCount = 0;
                        printEntry(message.getEntries());
                    }

                    connector.ack(batchId); // 提交确认
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                }

                log.info("empty too many times, exit");

            } catch (Exception e) {
                log.error("成功断开监测连接!尝试重连:{}",e);
            } finally {
                connector.disconnect();
                //防止频繁访问数据库链接: 线程睡眠 10秒
                try {
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) {
                    log.error("成功断开监测连接!尝试重连:{}",e);
                }
            }
        }
    }


    private  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("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                //canal获取mysql数据库删除事件
                if (eventType == CanalEntry.EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {//canal获取mysql数据库新增事件
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    log.info("-------&gt; before");
                    printColumn(rowData.getBeforeColumnsList());
                    log.info("-------&gt; after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private  void printColumn(List<CanalEntry.Column> columns) {
        SysUser sysUser = new SysUser();
        for (CanalEntry.Column column : columns) { //一行数据库数据=一个对象
            log.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            //获取字段名称和字段值,设置到实体类中
            if(column.getName().equalsIgnoreCase("id")){
                sysUser.setId(column.getValue());
            }else if(column.getName().equalsIgnoreCase("name")){
                sysUser.setName(column.getValue());
            }else if(column.getName().equalsIgnoreCase("age")){
                sysUser.setAge(Integer.valueOf(column.getValue()));
            }else if(column.getName().equalsIgnoreCase("email")){
                sysUser.setEmail(column.getValue());
            }
        }
        if(sysUser.getId()!=null && !"".equals(sysUser.getId())){
            String userJson = JSON.toJSONString(sysUser);
            redisClient.set(sysUser.getId(),userJson);//保存用户数据
        }
        log.info(sysUser.toString());
    }

}

2、MySQL数据同步到Redis
2.1 测试代码

package cn.itbeien.canal.test;

import cn.itbeien.canal.util.CanalUtil;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;


@SpringBootTest
public class CanalApplication {
    @Autowired
    private CanalUtil canalUtil;
    @Test
    public void test(){
        this.canalUtil.startMonitorSQL();
    }
}

2.2 环境准备
2.2.1 启动canal-admin

2.2.2 启动canal-server

2.2.3 启动canal-instance

2.2.4 启动canal-client
启动canal-client监听mysql增量数据，运行cn.itbeien.canal.test.CanalApplication

3、整体流程测试
在MySQL中新增一条数据

在canal-client端进行数据变更的监听

最后我们查询redis分布式缓存是否有id为88的这条数据