关于SpringBoot整合Canal数据同步的问题
目录
- 1、CentOS7编译安装MySQL5.7.24
- 2、Mysql设置binLog配置
- 3、Linux下载安装Canal服务
- 4、Boot项目中引入依赖
- 5、修改properties配置文件
- 6、修改Application启动类
- 7、创建Canal配置类自动监听
1、CentOS7编译安装MySQL5.7.24
CentOS7编译安装MySQL5.7.24的教程详解
链接地址：https://www.jb51.net/article/152246.htm
2、Mysql设置binLog配置
（1）检查binlog功能是否有开启
（2）如果显示状态为OFF表示该功能未开启，开启binlog功能
mysql> show variables like 'log_bin'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin | OFF | +---------------+-------+ 1 row in set (0.00 sec)
1，修改 mysql 的配置文件 my.cnf
vi /etc/my.cnf
追加内容：
log-bin=mysql-bin #binlog文件名 binlog_format=ROW #选择row模式 server_id=1 #mysql实例id,不能和canal的slaveId重复
2，重启 mysql：
service mysql restart
3，登录 mysql 客户端，查看 log_bin 变量
mysql> show variables like 'log_bin'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin | ON| +---------------+-------+ 1 row in set (0.00 sec) 如果显示状态为ON表示该功能已开启
（3）在mysql里面添加以下的相关用户和权限
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal'; GRANT SHOW VIEW, SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
3、Linux下载安装Canal服务
下载地址：
https://github.com/alibaba/canal/releases
（1）下载之后，放到目录中，解压文件
mkdir /usr/local/canal cd /usr/local/canal canal.deployer-1.1.4.tar.gz tar zxvf canal.deployer-1.1.4.tar.gz
（2）修改配置文件
vi conf/example/instance.properties
#需要改成自己的数据库信息 canal.instance.master.address=39.106.224.236:3306 #需要改成自己的数据库用户名与密码 canal.instance.dbUsername=canal canal.instance.dbPassword=canal #需要改成同步的数据库表规则 #1、同步所有的表 canal.instance.filter.regex=.*\\..* #2、需要同步的那个库中的那个表 #canal.instance.filter.regex=guli_ucenter.ucenter_member
mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\)
常见例子：
- 所有表：.* or .\…
- canal schema下所有表： canal\…*
- canal下的以canal打头的表：canal\.canal.*
- canal schema下的一张表：canal.test1
- 多个规则组合使用：canal\…*,mysql.test1,mysql.test2 (逗号分隔)
- 注意：此过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql，所以无法准确提
- 取tableName进行过滤).
（3）进入bin目录下启动
./startup.sh
4、Boot项目中引入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--mysql-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>commons-dbutils</groupId>
<artifactId>commons-dbutils</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
</dependency>
</dependencies>
5 、修改properties配置文件
# mysql数据库连接 spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver spring.datasource.url=jdbc:mysql://localhost:3306/guli?serverTimezone=GMT%2B8 spring.datasource.username=root spring.datasource.password=123456
6、修改Application启动类
7、创建Canal配置类自动监听
注意： 在run()方法中自行修改Linux虚拟机Ip地址后直接使用！
package com.atguigu.canal.client;
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 com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.net.InetSocketAddress;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
@Component
public class CanalClient {
//sql队列
private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();
@Resource
private DataSource dataSource;
/**
* canal入库方法
*/
public void run() {
CanalConnector connector = CanalConnectors.newSingleConnector(new
/*
* 此处Ip地址为linux虚拟机地址
* 端口号为固定 11111
* 其他不用修改
*/
InetSocketAddress("39.106.224.236",
11111), "example", "", "");
int batchSize = 1000;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
try {
while (true) {
//尝试从master那边拉去数据batchSize条记录，有多少取多少
Message message = connector.getWithoutAck(batchSize);
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
Thread.sleep(1000);
} else {
dataHandle(message.getEntries());
}
connector.ack(batchId);
//当队列里面堆积的sql大于一定数值的时候就模拟执行
if (SQL_QUEUE.size() >= 1) {
executeQueueSql();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
} finally {
connector.disconnect();
}
}
/**
* 模拟执行队列里面的sql语句
*/
public void executeQueueSql() {
int size = SQL_QUEUE.size();
for (int i = 0; i < size; i++) {
String sql = SQL_QUEUE.poll();
System.out.println("[sql]----> " + sql);
this.execute(sql.toString());
}
}
/**
* 数据处理
* @param entrys
*/
private void dataHandle(List<Entry> entrys) throws
InvalidProtocolBufferException {
for (Entry entry : entrys) {
if (EntryType.ROWDATA == entry.getEntryType()) {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
EventType eventType = rowChange.getEventType();
if (eventType == EventType.DELETE) {
saveDeleteSql(entry);
} else if (eventType == EventType.UPDATE) {
saveUpdateSql(entry);
} else if (eventType == EventType.INSERT) {
saveInsertSql(entry);
}
}
}
}
/**
* 保存更新语句
* @param entry
*/
private void saveUpdateSql(Entry entry) {
try {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
List<RowData> rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List<Column> newColumnList = rowData.getAfterColumnsList();
StringBuffer sql = new StringBuffer("update " +
entry.getHeader().getTableName() + " set ");
for (int i = 0; i < newColumnList.size(); i++) {
sql.append(" " + newColumnList.get(i).getName()
+ " = '" + newColumnList.get(i).getValue() + "'");
if (i != newColumnList.size() - 1) {
sql.append(",");
}
}
sql.append(" where ");
List<Column> oldColumnList = rowData.getBeforeColumnsList();
for (Column column : oldColumnList) {
if (column.getIsKey()) {
//暂时只支持单一主键
sql.append(column.getName() + "=" + column.getValue());
break;
}
}
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 保存删除语句
*
* @param entry
*/
private void saveDeleteSql(Entry entry) {
try {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
List<RowData> rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List<Column> columnList = rowData.getBeforeColumnsList();
StringBuffer sql = new StringBuffer("delete from " +
entry.getHeader().getTableName() + " where ");
for (Column column : columnList) {
if (column.getIsKey()) {
//暂时只支持单一主键
sql.append(column.getName() + "=" + column.getValue());
break;
}
}
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 保存插入语句
*
* @param entry
*/
private void saveInsertSql(Entry entry) {
try {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
List<RowData> rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List<Column> columnList = rowData.getAfterColumnsList();
StringBuffer sql = new StringBuffer("insert into "+entry.getHeader().getTableName() + " (");
for (int i = 0; i < columnList.size(); i++) {
sql.append(columnList.get(i).getName());
if (i != columnList.size() - 1) {
sql.append(",");
}
}
sql.append(") VALUES (");
for (int i = 0; i < columnList.size(); i++) {
sql.append("'" + columnList.get(i).getValue() + "'");
if (i != columnList.size() - 1) {
sql.append(",");
}
}
sql.append(")");
SQL_QUEUE.add(sql.toString());
}
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
}
/**
* 入库
* @param sql
*/
public void execute(String sql) {
Connection con = null;
try {
if(null == sql) return;
con = dataSource.getConnection();
QueryRunner qr = new QueryRunner();
int row = qr.execute(con, sql);
System.out.println("update: "+ row);
} catch (SQLException e) {
e.printStackTrace();
} finally {
DbUtils.closeQuietly(con);
}
}
}
到此这篇关于SpringBoot整合Canal数据同步的文章就介绍到这了,更多相关SpringBoot Canal数据同步内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！
相关推荐
-
springboot 整合canal实现示例解析
目录 前言 环境准备 一.快速搭建canal服务 搭建步骤 1.服务器使用docker快速安装一个mysql并开启binlog日志 2.上传canal安装包并解压 3.进入到第二步解压后的文件目录,并修改配置文件 4.启动canal服务 二.与springboot整合 1.Java中使用canal 2.编写一个demo 3.与springboot整合 4.application.yml 配置文件 5.核心工具类 6.提供一个配置类,在程序启动后监听数据变化 7.启动类 前言 在Mysql到Ela
-
Springboot2.3.x整合Canal的示例代码
目录 一.故事背景 二.什么是Canal 三.Canal安装 (1)事前准备 (1)数据库开启binlog (2)数据库新建账号,开启MySQLslav权限 (2)CanalAdmin安装 (3)CanalServer安装 (4)springbootdemo示例 一.故事背景 前言… 最近工作中遇到了一个数据同步的问题 我们这边系统的一个子业务需要依赖另一个系统的数据,当另一个系统数据变更时,我们这边的数据库要对数据进行同步… 那么我自己想到的同步方式呢就两种: 1.MQ订阅,另一个系统数据变更
-
关于SpringBoot整合Canal数据同步的问题
目录 1.CentOS7编译安装MySQL5.7.24 2.Mysql设置binLog配置 3.Linux下载安装Canal服务 4.Boot项目中引入依赖 5.修改properties配置文件 6.修改Application启动类 7.创建Canal配置类自动监听 1.CentOS7编译安装MySQL5.7.24 CentOS7编译安装MySQL5.7.24的教程详解 链接地址:https://www.jb51.net/article/152246.htm 2.Mysql设置binLog配置
-
SpringBoot整合canal实现数据同步的示例代码
目录 一.前言 二.docker-compose部署canal 三.canal-admin可视化管理 四.springboot整合canal实现数据同步 五.canal-spring-boot-starter 一.前言 canal:阿里巴巴 MySQL binlog 增量订阅&消费组件https://github.com/alibaba/canal tips: 环境要求和配置参考 https://github.com/alibaba/canal/wiki/AdminGuide 这里额外提下Red
-
SpringBoot整合Canal与RabbitMQ监听数据变更记录
目录 需求 步骤 环境搭建 canal.properties instance.properties 修改canal配置文件 整合SpringBoot Canal实现客户端 Canal整合RabbitMQ SpringBoot整合RabbitMQ 需求 我想要在SpringBoot中采用一种与业务代码解耦合的方式,来实现数据的变更记录,记录的内容是新数据,如果是更新操作还得有旧数据内容. 经过调研发现,使用Canal来监听MySQL的binlog变化可以实现这个需求,可是在监听到变化后需要马上保
-
SpringBoot定时任务实现数据同步的方法
本文实例为大家分享了SpringBoot定时任务实现数据同步的具体代码,供大家参考,具体内容如下 前言 业务的需求是,通过中台调用api接口获得,设备数据,要求现实设备数据的同步. 方案一:通过轮询接口的方式执行 pullData() 方法实现数据同步 该方式的原理是先清空之前的所有数据,然后重新插入通过api调用获取的最新数据.该方法的优点,逻辑简单.缺点是,频繁删除.插入数据.再调用查询数据时候,某一时刻,数据全部删除,还没及时插入的时候.数据可能有异常. 方案二:通过轮询接口的方式执行 p
-
MySQL高性能实现Canal数据同步神器
目录 简介 配置MySQL centos7 安装 canal java客户端 简介 Canal是阿里巴巴基于Java开源的数据同步工具.平时业务场景使用比较多的如下: 同步数据到ES.Redis缓存中. 数据同步. 业务需要监听数据. 图片来源阿里巴巴github仓库:https://github.com/alibaba/canal 配置MySQL 跟配置主从复制类似. 数据执行如下命令: -- 给canal单独创建用户 用户名:canal 密码:123456 create user 'cana
-
SpringBoot 整合 Elasticsearch 实现海量级数据搜索功能
目录 一.简介 二.代码实践 2.1.导入依赖 2.2.配置环境变量 2.3.创建 elasticsearch 的 config 类 2.4.索引管理 2.5.文档管理 三.小结 今天给大家讲讲 SpringBoot 框架 整合 Elasticsearch 实现海量级数据搜索. 一.简介 在上篇ElasticSearch 文章中,我们详细的介绍了 ElasticSearch 的各种 api 使用. 实际的项目开发过程中,我们通常基于某些主流框架平台进行技术开发,比如 SpringBoot,今天我
-
Springboot整合Redis与数据持久化
目录 Springboot整合Redis 使用json方式存储 序列化方式存储数据 MySQL与Redis一致性解决同步问题 Redis持久化机制 全量同步与增量同步 RDB与AOF RDB AOF Springboot整合Redis 有两种存储数据的方式: 方案1:在Redis存放一个对象 使用json序列化与反序列化 方案2:直接使用redis自带序列化方式存储对象 maven依赖 <parent> <groupId>org.springframework.boot</g
-
SpringBoot整合Redis入门之缓存数据的方法
目录 前言 为什么要使用Redis呢? 相关依赖 配置 数据库 实体类 RedisConfig Mapper Service接口 Service实现类 测试Redis Controller 前言 Redis是一个开源的使用ANSI C语言编写.支持网络.可基于内存亦可持久化的日志型.Key-Value数据库,并提供多种语言的API.从2010年3月15日起,Redis的开发工作由VMware主持.从2013年5月开始,Redis的开发由Pivotal赞助. 为什么要使用Redis呢? 举个例子,
-
springboot整合redis进行数据操作(推荐)
redis是一种常见的nosql,日常开发中,我们使用它的频率比较高,因为它的多种数据接口,很多场景中我们都可以用到,并且redis对分布式这块做的非常好. springboot整合redis比较简单,并且使用redistemplate可以让我们更加方便的对数据进行操作. 1.添加依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starte