SpringBoot整合canal实现数据同步的示例代码

2024-04-02 19:04:59 280人浏览泡泡鱼

Python 官方文档：入门教程 => 点击学习

摘要

目录一、前言二、Docker-compose部署canal三、canal-admin可视化管理四、SpringBoot整合canal实现数据同步五、canal-spring-boot

一、前言

canal：阿里巴巴 Mysql binlog 增量订阅&消费组件
https://GitHub.com/alibaba/canal

tips: 环境要求和配置参考 Https://github.com/alibaba/canal/wiki/AdminGuide

这里额外提下Redis缓存和mysql数据一致性解决方案

延时双删策略
异步更新缓存(基于订阅binlog的同步机制)
…

我们的canal即可作为Mysql binlog增量订阅消费组件+MQ 消息队列将增量数据更新到redis

二、docker-compose部署canal

tips: 详情可查看 https://gitee.com/zhengqingya/docker-compose

# 准备
git clone https://gitee.com/zhengqingya/docker-compose.git
cd docker-compose/Liunx

# 导入初始化SQL
Liunx/canal/canal_admin/canal_manager.sql

# 运行  (tips:先修改配置文件信息)
docker-compose -f docker-compose-canal.yml -p canal up -d

三、canal-admin可视化管理

tips: 页面使用自行多点点就会了^_^
将数据投递到RabbitMQ配置为例，附带些其中重要的配置信息
mq的交换机和队列可通过后面java代码自动生成，无需自己手动建立

1、canal.properties

# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9

# tcp, kafka, RocketMQ, rabbitMQ
canal.serverMode = rabbitMQ

rabbitmq.host = www.zhengqingya.com:5672
rabbitmq.virtual.host = my_vhost
rabbitmq.exchange = canal.exchange
rabbitmq.username = admin
rabbitmq.password = admin

2、example/instance.propertiOS

canal.instance.master.address=www.zhengqingya.com:3306
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8

# table regex 只同步test数据库下的t_user表
canal.instance.filter.regex=test\\.t_user
# table black regex
canal.instance.filter.black.regex=mysql\\.slave_.*

# mq config
canal.mq.topic=canal_routing_key

四、springboot整合canal实现数据同步

1、pom.xml引入mq依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、application.yml配置

# RabbitMQ配置
spring:
  rabbitmq:
    addresses: www.zhengqingya.com:5672,www.zhengqingya.com:5673 # 指定client连接到的server的地址，多个以逗号分隔
    # 填写自己安装rabbitmq时设置的账号密码，默认账号密码为`guest`
    username: admin
    password: admin
    virtual-host: my_vhost # 填写自己的虚拟机名，对应可查看 `127.0.0.1:15672/#/users` 下Admin中的`Can access virtual hosts`信息

3、mq监听canal消息数据

@Slf4j
@Component
public class CanalRabbitMqListener {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = MqConstant.CANAL_QUEUE, durable = "true"),
                    exchange = @Exchange(value = MqConstant.CANAL_EXCHANGE),
                    key = MqConstant.CANAL_ROUTING_KEY
            )
    })
    public void handleCanalDataChange(String message) {
        log.info("[canal] 接收消息: {}", JSON.tojsONString(message));
    }

}

public interface MqConstant {

    String CANAL_EXCHANGE = "canal.exchange";
    String CANAL_QUEUE = "canal_queue";
    String CANAL_ROUTING_KEY = "canal_routing_key";

}

4、测试

先启动项目让程序自动建立所需mq中的交换机和队列

在这里插入图片描述

再修改canal监听的表数据

在这里插入图片描述

查看程序监听的mq消息数据如下，拿到数据就可以进行数据解析处理了…

在这里插入图片描述

五、canal-spring-boot-starter

tips: 可参考 https://github.com/NORManGyllenhaal/canal-client

此方式需将canal.properties配置文件中的canal.serverMode属性值修改为tcp

在这里插入图片描述

1、pom.xml中引入依赖

<!-- https://mvnrepository.com/artifact/top.javatool/canal-spring-boot-starter -->
<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

2、application.yml配置

canal:
  server: www.zhengqingya.com:11111
  destination: example # tips：canal-admin中Instance管理下需存在example实例配置

3、监听canal数据 – 订阅数据库增删改操作

@Slf4j
@Component
@CanalTable(value = "t_user")
public class UserHandler implements EntryHandler<User> {

    @Override
    public void insert(User user) {
        log.info("insert message  {}", user);
    }

    @Override
    public void update(User before, User after) {
        log.info("update before {} ", before);
        log.info("update after {}", after);
    }

    @Override
    public void delete(User user) {
        log.info("delete  {}", user);
    }

}

@Data
@Table(name = "t_user")
public class User implements Serializable {

    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "user_id")
    private Integer userId;

    
    @Column(name = "username")
    private String username;

    
    @Column(name = "password")
    private String password;

    
    @Column(name = "sex")
    private Integer sex;

    
    private String remark;

    
    private Date date;

}

经测试发现这个jar存在一些bug，ex：针对表字段，数据原本为空，修改为有值的时候，如果java这边用非String字段类型去接收会报错！

在这里插入图片描述