我要评论前言
在分布式系统中，数据同步是一个常见的需求。例如，我们可能需要将主库的数据实时同步到多个从库，或者将数据从一个数据库集群同步到另一个集群。本篇内容通过一个实际案例，介绍如何使用 spring boot 和 canal 实现 mysql 数据库之间的数据同步。
一、背景
假设我们有以下数据库架构：
- 两个主库：db_1 和 db_2。
- 每个主库对应两个从库：db_1_bk_1、db_1_bk_2 和 db_2_bk_1、db_2_bk_2。
- 我们的目标是：
- 将 db_1 的数据同步到 db_1_bk_1 和 db_1_bk_2。
- 将 db_2 的数据同步到 db_2_bk_1 和 db_2_bk_2。
二、canal 简介
canal 是阿里巴巴开源的一款基于 mysql binlog 的增量数据订阅与分发工具。它通过模拟 mysql 的从节点，实时捕获主库的 binlog 日志，并将数据变更事件推送给下游消费者。canal 支持多种下游适配器，如 kafka、rabbitmq 和直接消费。
三、主库数据库配置
1.主库配置
为了使 canal 能够正常解析 binlog 日志，主库需要进行以下配置：
- 开启 binlog 日志：确保主库开启了 binlog 日志，并且设置为 row 模式。
- 配置 server-id：为每个主库设置唯一的 server-id。
- 创建 canal 用户并授予权限：创建一个用户供 canal 使用，并授予必要的权限。
编辑主库的配置文件（my.cnf 或 my.ini），添加以下内容：
[mysqld] # 开启 binlog 日志 log-bin=mysql-bin # 设置 binlog 格式为 row 模式 binlog-format=row # 设置唯一的 server-id server-id=1
注意：
- 如果你有多个主库，每个主库的 server-id 必须是唯一的。
- 修改配置后，需要重启 mysql 服务以使配置生效。
2.创建 canal 用户并授予权限
canal 需要一个具有读取 binlog 权限的 mysql 用户。以下是创建用户并授予权限的步骤：
# 登录 mysql mysql -u root -p # 创建用户 create user 'canal'@'%' identified by 'canal'; # 授予权限 grant replication slave, replication client on *.* to 'canal'@'%' identified by 'canal'; # 刷新权限 flush privileges;
说明：
- canal 用户需要足够的权限来读取 binlog 数据，但不需要对数据库进行写操作。
- 如果你的 mysql 版本较新（8.x），可能需要使用 alter user 命令来设置密码：
alter user 'canal'@'%' identified by 'canal';
四.配置 canal server
canal server 是 canal 的核心组件，负责连接主库并解析 binlog 数据。我们需要为每个主库配置一个 canal 实例。
1.canal server 配置文件
在 canal server 的配置目录下，创建两个实例配置文件：conf/db_1/instance.properties 和 conf/db_2/instance.properties。conf/db_1/instance.properties：
# 主库的地址和端口 canal.instance.master.address=db_1_ip:3306 # canal 连接主库的用户名和密码 canal.instance.dbusername=canal canal.instance.dbpassword=canal # 需要同步的表正则表达式，这里表示同步 db_1 数据库的所有表 canal.instance.filter.regex=db_1\\..*
conf/db_2/instance.properties：
# 主库的地址和端口 canal.instance.master.address=db_2_ip:3306 # canal 连接主库的用户名和密码 canal.instance.dbusername=canal canal.instance.dbpassword=canal # 需要同步的表正则表达式，这里表示同步 db_2 数据库的所有表 canal.instance.filter.regex=db_2\\..*
2.启动 canal server
使用以下命令启动 canal server：
nohup sh bin/canal.sh start &
注意：
- 确保主库的 binlog 位置和文件名正确。如果不确定，可以通过 show master status; 命令查看。
- 如果主库已经运行了一段时间，需要指定 binlog 的起始位置，避免重复同步旧数据。
五.开发 spring boot 客户端
spring boot 客户端作为 canal 的消息消费者，负责接收数据变更事件并同步到目标从库。
1. 引入依赖
在 spring boot 项目的 pom.xml文件中，引入 canal 客户端依赖：
<dependency>
<groupid>com.alibaba.otter</groupid>
<artifactid>canal.client</artifactid>
<version>1.1.8</version>
</dependency>2. 配置 canal 客户端
在 application.yml 文件中，配置 canal server 的地址：
canal: server.ip: canal_server_ip server.port: 11111
3. 实现数据同步逻辑
创建一个 canal 客户端服务类，用于接收和处理数据变更事件。canalclientservice.java：
@service
public class canalclientservice {
private final canalconnector canalconnector;
public canalclientservice(@value("${canal.server.ip}") string canalserverip, @value("${canal.server.port}") int canalserverport) {
this.canalconnector = canalconnectors.newsingleconnector(new inetsocketaddress(canalserverip, canalserverport), "example", "", "");
}
@postconstruct
public void start() {
canalconnector.connect();
canalconnector.subscribe("db_1..*, db_2..*"); // 订阅 db_1 和 db_2 的所有表
new thread(this::process).start();
}
private void process() {
while (true) {
message message = canalconnector.getwithoutack(100);
long batchid = message.getid();
if (batchid == -1 || message.getentries().isempty()) {
continue;
}
for (entry entry : message.getentries()) {
handledata(entry);
}
canalconnector.ack(batchid);
}
}
private void handledata(entry entry) {
string schemaname = entry.getheader().getschemaname(); // 数据库名
string tablename = entry.getheader().gettablename(); // 表名
eventtype eventtype = entry.getheader().geteventtype(); // 数据变更类型
system.out.println("schema: " + schemaname + ", table: " + tablename + ", type: " + eventtype);
// 根据来源数据库同步到对应的从库
if ("db_1".equals(schemaname)) {
synctobackupdbs(entry, "db_1_bk_1", "db_1_bk_2");
} else if ("db_2".equals(schemaname)) {
synctobackupdbs(entry, "db_2_bk_1", "db_2_bk_2");
}
}
private void synctobackupdbs(entry entry, string... backupdbs) {
// 根据事件类型同步到从库
if (entry.getheader().geteventtype() == eventtype.insert) {
for (string db : backupdbs) {
syncinsert(entry, db);
}
} else if (entry.getheader().geteventtype() == eventtype.update) {
for (string db : backupdbs) {
syncupdate(entry, db);
}
} else if (entry.getheader().geteventtype() == eventtype.delete) {
for (string db : backupdbs) {
syncdelete(entry, db);
}
}
}
private void syncinsert(entry entry, string backupdb) {
// 使用 mybatis 将数据插入到对应的从库
system.out.println("insert into " + backupdb);
}
private void syncupdate(entry entry, string backupdb) {
// 使用 mybatis 将数据更新到对应的从库
system.out.println("update into " + backupdb);
}
private void syncdelete(entry entry, string backupdb) {
// 使用 mybatis 将数据从对应的从库删除
system.out.println("delete from " + backupdb);
}
}六.启动并测试
- 启动 canal server。
- 启动 spring boot 应用。
- 在主库 db_1 或 db_2 中插入、更新或删除数据。
- 观察从库 db_1_bk_1、db_1_bk_2、db_2_bk_1 和 db_2_bk_2 是否同步成功。
七.注意事项
- 数据一致性：确保从库的数据与主库保持一致。可以通过事务或锁机制来避免冲突。
- 性能优化：如果数据量较大，建议结合中间件（如 kafka）进行缓冲和负载均衡。
- 错误处理：在同步过程中，需要处理网络异常、数据库连接异常等情况。
- canal server 高可用：在生产环境中，建议部署 canal server 的集群，以提高系统的可用性。
八.总结
通过 spring boot 和 canal，我们可以实现 mysql 数据库之间的高效数据同步。canal 提供了强大的 binlog 解析能力，而 spring boot 则提供了灵活的开发框架，两者结合可以轻松应对复杂的分布式数据同步需求。希望本文对你有所帮助，如果有任何问题，欢迎在评论区留言。
到此这篇关于使用 spring boot 和 canal 实现 mysql 数据库同步的文章就介绍到这了,更多相关spring boot mysql 数据库同步内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论