cannel mysql_mysql、canal、kafka、数据同步系列(四)canal的安装和配置
历经了前面的各种安装,终于到canal的安装和配置了。这里依然采用docker-compose安装.
1、编写docker-compose.yml文件
version: '3'
services:
canal-server:
image: canal/canal-server:v1.1.3
container_name: canal-server
ports:
- 11111:11111
environment:
- canal.instance.mysql.slaveId=12
- canal.auto.scan=false
- canal.destinations=test
- canal.instance.master.address=192.168.1.9:3306
- canal.instance.dbUsername=canal
- canal.instance.dbPassword=canal
- canal.mq.topic=test
- canal.instance.filter.regex=esen_approval.apt_approval
volumes:
- ./conf:/canal-server/conf
- ./logs:/canal-server/logs
2、说明
canal.instance.mysql.slaveId:slaveId不能与mysql的serverId一样
canal.instance.master.address:mysql地址
canal.instance.dbUsername:mysql账号
canal.instance.dbPassword:mysql密码
3、启动
docker-compose up -d
4、到这一步,我们就实现了canal同步mysql binlog数据了
5、我们来测试下
1)我们采用maven,构建一个项目。加入pom
com.alibaba.otter
canal.client
1.1.0
2)在数据库里面有一个库test,我们新建一个表
CREATE TABLE `user` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB
3)编写测试类
public class Cannel {
public static void main(String args[]) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.10.10.10",
11111), "test", "xxxx", "xxxx");
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++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
private static void printEntry(List 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("================> 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()) {
if (eventType == CanalEntry.EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == CanalEntry.EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List columns) {
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
4)运行main方法,我们在数据库表user中insert、update,查看控制台的信息
,我们会收到binlogs信息。完美,我们实现了初步的同步。
cannel mysql_mysql、canal、kafka、数据同步系列(四)canal的安装和配置相关推荐
- 系统重构数据同步利器之Canal实战篇
一.背景 二话不说,先上图 上图来自于官网(https://github.com/alibaba/canal),基本上涵盖了目前生产环境使用场景了,众所周知,Canal做数据同步已经是行业内标杆了.我 ...
- canal mq数据同步
canal mq数据同步 官网:https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart ************** ...
- linux数据同步技术比较,linux下实现web数据同步的四种方式(性能比较)教程.docx
linux下实现web数据同步的四种方式(性能比较)教程 实现web数据同步的四种方式=======================================1.nfs实现web数据共享2.rs ...
- 数据同步的解决方案Canal
Canal实现数据同步的原理: 1.是根据模拟mysql slave的主从交互协议,伪装自己是mysql slave,向mysql master发送dump请求. 2.mysql master收到du ...
- linux下实现web数据同步的四种方式(性能比较)
实现web数据同步的四种方式 ======================================= 1.nfs实现web数据共享 2.rsync +inotify实现web数据同步 3.rs ...
- 实现web数据同步的四种方式
实现web数据同步的四种方式 ======================================= 1.nfs实现web数据共享 2.rsync +inotify实现web数据同步 3.rs ...
- 客快物流大数据项目(六十一):将消费的kafka数据同步到Kudu中
目录 将消费的kafka数据同步到Kudu中 一.导入表名映射关系类
- GBase 8a Kafka 数据同步
数据同步系统通过 Oracle Goldengate.GBase RTSync 等工具复制 Oracle.GBase 8s等数据库的业务数据到 GBase 8a MPP Cluster,为了应对业务系 ...
- Python量化数据仓库搭建系列1:数据库安装与操作
Python量化数据仓库搭建系列1:数据库安装与操作 本系列教程为量化开发者,提供本地量化金融数据仓库的搭建教程与全套源代码.我们以恒有数(UDATA)金融数据社区为数据源,将金融基础数据落到本地数据 ...
- Greenplum【环境搭建 04】使用GPKafka实现Kafka数据导入Greenplum数据库(扩展安装文件网盘分享)
分享资源地址及文件列表: 链接:https://pan.baidu.com/s/1XVTxKLkOYrL4pCZpFfs-Tg 提取码:sq90 包含文件: # 命令执行 gpkafka # 扩展安装 ...
最新文章
- python读写二进制文件的方法
- 简述推荐系统中的矩阵分解
- WiFi的基本调制过程
- SQL2005存储过程解密
- 巴铁 无人驾驶_巴铁?Clip-air?请查收汽车的最新玩法
- 笔记本启动关闭小键盘NUM LOCK的N种方法
- securecrt 中文横着显示解决
- 如何用java实现阶乘倒数求和_JAVA 阶乘 的倒数求和public class Jiecheng {public static void main(...
- 苹果6严重卡顿_苹果手机变慢变卡,怎么破?
- 人人网(cookie登录)
- 深度学习基础----GAE和VGAE
- L1-061 新胖子公式 (10 分)C语言
- JAVA EE面试重点
- 有趣的表情包购物网站
- 升级鸿蒙系统最新计划,华为鸿蒙系统更新升级方法分享
- 基于模型预测控制MPC的光伏并网系统设计|太阳能发电|模型预测控制
- C语言C++编程学习:排序原理分析
- clover EFI bootloaderfor mac(10.12)官方版下载
- 【frp】使用 frp 实现内网穿透(CentOS 云服务器 + Win10)
- Spring Cloud 配置(E.SR5)
热门文章
- PHP输出星座,php根据日期显示所在星座的方法
- 【电子取证:镜像仿真篇】Linux镜像仿真、E01镜像取证
- 2年Android7K,2月Android手机性能榜:神机K40冲榜功成 天玑820力压麒麟中端芯
- loj2038「SHOI2015」超能粒子炮・改(卢卡斯的理解和延伸+数位dp)
- hive:字符串转int
- 怎么在eclipse中调试java程序?
- mybatis快速入门
- unc 目录不受支持_DOS批处理不支持将UNC 路径作为当前目录的解决方法
- 覆盖率测试工具(coverage tool)
- 【调剂】西安微电子技术研究所2023年硕士研究生入学考试复试分数线及复试安排...