php阿里的同步工具canal,基于阿里的Canal实现数据同步
一、开启同步数据库的binlog功能
(1)开启同步数据端的数据库服务(比如我的将一号虚拟机上的mysql数据库作为同步操作数据库)
systemctl start mysql.service
mysql -h192.168.137.100 -uroot -p123
(2)检查mysql的binlog功能是否开启(可见是OFF，关闭的)
MySQL [(none)]> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | OFF |
+---------------+-------+
(3)开启binlog功能，修改my.cnf配置文件：vim /etc/my.cnf
在文件中添加如下内容后，重启mysql服务，在查看mysql的binlog功能。可见已经开启。
log-bin=mysql-bin
binlog_format=ROW
server_id=1
MySQL [(none)]> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+
二、安装canal工具
(1)下载地址：https://github.com/alibaba/canal/releases
(2)下载文件：canal.deployer-1.1.4.tar.gz
(3)文件传到 /usr/local/canal文件夹下，解压
tar zxvf canal.deployer-1.1.4.tar.gz
(4)修改配置文件vi conf/example/instance.properties
修改如下内容为数据库连接：
canal.instance.master.address=192.168.137.100:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=123
(5)启动canal数据同步工具 cd bin
[root@localhost bin]# ./startup.sh
三、测试项目准备
(1)新建项目，引入如下依赖
org.springframework.boot
spring-boot-starter-web
mysql
mysql-connector-java
commons-dbutils
commons-dbutils
org.springframework.boot
spring-boot-starter-jdbc
com.alibaba.otter
canal.client
(2)项目配置文件配置本地同步数据库
# 服务端口
server.port=1000
# 服务名
spring.application.name=canal-client
# 环境设置：dev、test、prod
spring.profiles.active=dev
# mysql数据库连接
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://192.168.137.130:3306/TeachOnLine?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=123
(3)创建如下启动类
@SpringBootApplication
public class CanalApplication implements CommandLineRunner {
@Resource
private CanalClient canalClient;
// canalClient为数据库同步操作执行类
public static void main(String[] args) {
SpringApplication.run(CanalApplication.class, args);
}
@Override
public void run(String... strings) throws Exception {
//项目启动，执行canal客户端监听
canalClient.run();
}
}
(4)创建CanalClient数据库同步操作执行类内容如下(记得保证192.168.137.100服务器的11111端口可通)
@Component
public class CanalClient {
//sql队列
private Queue SQL_QUEUE = new ConcurrentLinkedQueue<>();
@Resource
private DataSource dataSource;
/**
* canal入库方法
*/
public void run() {
// 创建远程数据库连接对象
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.137.100",
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 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 rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List 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 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 rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List 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 rowDatasList = rowChange.getRowDatasList();
for (RowData rowData : rowDatasList) {
List 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);
}
}
}
四、测试
(1)启动类开启，在192.168.137.100(远程)数据库TeachOnLine的表members中插入数据
insert into members values(2,‘candy’,22);
(2)查看控制台打印信息
[sql]----> insert into members (id,username,age) VALUES ('2','candy','22')
update: 1
(3)在192.168.137.130(本地)数据库TeachOnLine的表members中出现相同数据。
注：在此例中本地数据库会同步远程数据库的变化，反过来修改本地数据库内容，远程数据库不会同步本地数据库的加粗样式变化
标签：Canal,canal,同步,数据库,mysql,阿里,sql,entry,size
来源： https://blog.csdn.net/weixin_44187615/article/details/111726524
php阿里的同步工具canal,基于阿里的Canal实现数据同步相关推荐
- elasticsearch 数据类型_基于 MySQL Binlog 的 Elasticsearch 数据同步实践
来源;马蜂窝 一.背景 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品.订单等数据的多维度检索. 使用 Elasticsearch 存 ...
- 基于 MySQL Binlog 的 Elasticsearch 数据同步实践
一.为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品.订单等数据的多维度检索. 使用 Elasticsearch 存储业务数 ...
- 基于 MySQL Binlog 的 Elasticsearch 数据同步实践 原
一.背景 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品.订单等数据的多维度检索. 使用 Elasticsearch 存储业务数据可以 ...
- canal+Kafka实现mysql与redis数据同步
前言 上篇文章简单介绍canal概念,本文结合常见的缓存业务去讲解canal使用.在实际开发过程中,通常都会把数据往redis缓存中保存一份,做下简单的查询优化.如果这时候数据库数据发生变更操作,就不 ...
- 基于Java方式如何实现数据同步
本篇内容介绍了"基于Java方式如何实现数据同步"的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读, ...
- datax 持续数据同步_采用DataX实现多表增量数据同步
这两天验证了一下阿里的DataX数据同步工具,觉得DataX可以用来做管理数据的多级数据同步.DataX用来做批量数据迁移很适合,能够保证数据的一致性,性能也很好,结合时间戳字段,用来实现数据定时增量 ...
- 数据同步之采用时间戳的方法进行增量数据同步（一）
本文主要介绍源表为单表时,增量数据同步的情况.当源表为多表时,后面的文章会继续介绍. 一.数据同步情况说明 将源数据库S中的A表(将此表称为源表),通过ETL工具同步至目标数据库T的A表(将此表称为目 ...
- 大数据常用同步工具（DataX/Sqoop/Nifi/Canal等）
一.离线数据同步 DataX 阿里的Datax是比较优秀的产品,基于python,提供各种数据村塾的读写插件,多线程执行,使用起来也很简单,操作简单通常只需要两步: 创建作业的配置文件(json格式配 ...
- 如何将项目发布到阿里云_尚硅谷基于阿里云搭建数据仓库（实时）项目视频发布...
11月,尚大发布了基于阿里云搭建数据仓库离线项目视频,不少爱学习的小伙伴表示,对实时项目视频翘首以盼,听了大海哥的课,腰不酸了,腿不疼了,一口气从头听到尾还不过瘾,沉迷于学习无法自拔-- 说曹操,大海 ...
最新文章
- 2016年度工作总结
- python子进程通信_python执行子进程实现进程间通信的方法
- 汇编语言--串处理指令
- Docker 容器时区时间不一致问题解决
- 将M进制的数转换为N进制的数（java）
- .mvn 需要放git上吗_汽车行李架上可以放东西吗？放了东西可以跑高速吗？
- UnicodeEncodeError：#39;ascii#39;编解码器无法在位置20编码字符u#39;\\ xa0#39;：序数不在范围内（128）
- tensorflow精进之路(二十五)——Object Detection API目标检测(下)（VOC数据集训练自己的模型进行目标检测）
- springMvc将对象json返回时自动忽略掉对象中的特定属性的注解方式
- 泪目！java项目源码免费下载
- 实验三 vi编辑器(Linux基础教程）
- 记录一个报错问题 关于调用win32com打开文档
- 『前端面试资料』 前端面试考题汇总 摘自校内某内推群群公告
- vue-lazyload图片懒加载的简单使用
- 第十二周《支持SQl Server内部,第二版》CHAR1数据页面和数据行 作者Dmitri Korotkevitch
- switch条件语句与跳转语句（附分数分层案例）
- 【商业信息】PNP ID注册名单 2019-05-21
- getchar()函数的作用
- 移动互联网你需要了解35家潜力企业
- android checkbox 文字和图标间距,CheckBox选择框和文字设置间距