阿里开源数据同步组件Canal
一.简介
canal是阿里开源的数据同步组件
这个是是git地址
二.使用步骤
1.安装配置mysql
- 安装一个数据库(这个数据库是被监听的对象,我这里用的是mysql5.7)
- 创建一个用户专门用于数据同步(这里也可以直接用root用户)
①创建用户 用户名:canal 密码:canal
create user 'canal'@'%' identified by 'canal';
②授予外部访问权限
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'canal';
- 修改mysql配置文件
# 打开binlog
log-bin=mysql-bin
# 选择ROW(行)模式
binlog-format=ROW
# 配置MySQL replaction需要定义,不要和canal的slaveId重复(canal1.1.4自动生成slaveId)
server_id=1
- 修改完配置文件之后,重启mysql,再查看是否修改成功
show VARIABLES like 'log_bin';
- 创建数据库
create database user character set utf8;
- 创建表
CREATE TABLE `user`.`tb_user` (`id` int(32) NOT NULL,`username` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,`age` int(32) NULL DEFAULT NULL,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;
2.安装配置canal
下载canal,我这里用的是1.1.4,github下载可能非常慢,这里再贴上我的百度云,提取码:ewji
将canal放到/usr/local/canal目录下,解压
tar -zxvf canal.deployer-1.1.4.tar.gz
- 打开配置文件
vi conf/example/instance.properties
只需要修改这行代码配置文件就可以了,如果用户不是canal,还需要修改下dbUsername和dbPassword,
canal.instance.filter.regex不用修改,默认监听整个数据库,也可以写具体的表名,用逗号隔开
注意:如果这里使用的是docker安装的mysql,有可能没有vi命令,需要手动安装
apt-get update
apt-get install vim
如果上面的命令执行失败,我之前用centos 8的时候报错了,最后没有解决,更换成centos 7就可以了
3.启动canal
- 启动canal
- 查看日志文件,查看canal是否启动成功
启动成功
注意,这里启动的时候有可能会有很多问题出现
比如:数据库编码问题
如果是用navicat创建的数据库,这样看上去用的好像是utf8,但是在canal这里不行
我们可以看一下该数据库的编码格式,如果是这样的就会有问题
show variables like "%character%";
修改一下数据的编码格式
set character_set_client = utf8;set character_set_server = utf8;set character_set_connection = utf8;set character_set_database = utf8;set character_set_results = utf8;set collation_connection = utf8_general_ci;set collation_database = utf8_general_ci;set collation_server = utf8_general_ci;
这样再重启就不会有编码问题了
三.Java canal客户端
- 引入maven依赖
<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.4</version>
</dependency>
- 修改启动类
需要在项目启动时便持续监听
package com.xx;import com.xx.client.CanalClient;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import javax.annotation.Resource;/*** @author aqi* DateTime: 2021/2/24 4:43 下午* Description: canal启动类*/
@SpringBootApplication
public class CanalApplication implements CommandLineRunner {@Resourceprivate CanalClient canalClient;public static void main(String[] args) {SpringApplication.run(CanalApplication.class, args);}@Overridepublic void run(String... args) throws Exception {// 项目启动,执行canal客户端监听canalClient.run();}
}
- 编写canal客户端
这里操作数据库用的是JdbcTemplate
package com.xx.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.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;/*** @author aqi* DateTime: 2021/2/24 4:48 下午* Description: No Description*/
@Component
public class CanalClient {//sql队列private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();@Resourceprivate JdbcTemplate jdbcTemplate;/*** canal入库方法*/public void run() {CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("服务器ipd地址", 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 | 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);}}/*** 数据处理*/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);}}}}/*** 保存更新语句*/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();StringBuilder sql = new StringBuilder("update " + entry.getHeader().getTableName() + " set ");for (int i = 0; i < newColumnList.size(); i++) {sql.append(" ").append(newColumnList.get(i).getName()).append(" = '").append(newColumnList.get(i).getValue()).append("'");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()).append("=").append(column.getValue());break;}}SQL_QUEUE.add(sql.toString());}} catch (InvalidProtocolBufferException e) {e.printStackTrace();}}/*** 保存删除语句*/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();StringBuilder sql = new StringBuilder("delete from " + entry.getHeader().getTableName() + " where ");for (Column column : columnList) {if (column.getIsKey()) {//暂时只支持单一主键sql.append(column.getName()).append("=").append(column.getValue());break;}}SQL_QUEUE.add(sql.toString());}} catch (InvalidProtocolBufferException e) {e.printStackTrace();}}/*** 保存插入语句*/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();StringBuilder sql = new StringBuilder("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("'").append(columnList.get(i).getValue()).append("'");if (i != columnList.size() - 1) {sql.append(",");}}sql.append(")");SQL_QUEUE.add(sql.toString());}} catch (InvalidProtocolBufferException e) {e.printStackTrace();}}/*** 入库*/public void execute(String sql) {jdbcTemplate.execute(sql);}
}
注意:
如果项目启动失败,并且抛出canal连接失败,这个大概率是防火墙的问题,因为canal默认的端口号是11111,这个默认是关闭的,所以需要手动开启
查看已经开启的端口号
firewall-cmd --zone=public --list-ports
开启11111端口
firewall-cmd --zone=public --add-port=11111/tcp --permanent
重启防火墙
firewall-cmd --reload
- 往数据库中插入数据
INSERT INTO tb_user (username, age) VALUES ('Tom', 20)
- 查看数据是否同步,可以看到数据同步成功
三.总结
- canal对于代码没有侵入性,基于监听mysql binlog日志去实现数据同步,实时性也非常的不错,这里有官方发出的性能测试结果,日常工作一般是够用了
阿里开源数据同步组件Canal相关推荐
- 开源oracle同步图形工具,阿里开源数据同步工具--DataX
阿里开源数据同步工具--DataX 是啥?: 是异构数据源离线同步工具 能干啥?: 能够将MySQL sqlServer Oracle Hive HBase FTP 之间进行稳定高效的数据同步. 设 ...
- 阿里开源数据同步神器DataX异构数据源间数据同步同步MySQL与HDFS相互实战
Datax 实战使用 继上一篇 阿里开源数据同步神器DataX异构数据源间数据同步基础介绍与快速入门之后的实战篇 1.MySQL-To-HDFS 环境 & 准备说明: 描述: 为了快速搭建测试 ...
- mysql获取最好成绩对应数据的其他项_开源数据同步神器——canal
前言 如今大型的IT系统中,都会使用分布式的方式,同时会有非常多的中间件,如redis.消息队列.大数据存储等,但是实际核心的数据存储依然是存储在数据库,作为使用最广泛的数据库,如何将mysql的数据 ...
- 开源数据同步神器——canal
前言 如今大型的IT系统中,都会使用分布式的方式,同时会有非常多的中间件,如redis.消息队列.大数据存储等,但是实际核心的数据存储依然是存储在数据库,作为使用最广泛的数据库,如何将mysql的数据 ...
- 大数据任务调度和数据同步组件初探
本文个人博客地址 本文公众号地址 背景 数据从最原始的状态,可能是一个 excel,一个文本,或者是来自业务数据库的数据,格式各种各样,落地到数据仓库.数据湖中,数据的同步过程 是必不可少的 图片来源 ...
- DataBus(数据同步组件)
DataBus(数据同步组件) github: https://github.com/linkedin/databus/wiki Databus是一个低延迟.可靠的.支持事务的.保持一致性的数据变更抓 ...
- 开源数据同步备份工具(MySQL、Oracle、SqlServer、PostgreSQL)
真正的大师,永远都怀着一颗学徒的心! 一.项目简介 今天说的是一个开源数据同步备份工具,他支持mysql.oracle.kafka.PostgreSQL.sqlserver等 二.实现功能 数据同步与 ...
- k8s集群下搭建数据同步工具-canal:canal-admin篇
k8s集群下搭建数据同步工具-canal:canal-admin篇 前言 容器化 canal-admin 环境准备 k8s集群创建pod canal-admin 前言 本文使用v1.1.4版本的can ...
- 接口推送数据原理_数据同步组件(Canal)在珍爱网的应用与实践
本文作者:珍爱网技术团队 二爷 随着公司业务的不断发展,公司对于实时报表的需求越来越旺盛,原则上来说,实时报表最好的实现方式的通过Spark,storm这类的技术去支撑,由于人手原因,并不能很好的支撑 ...
最新文章
- MPB:南土所冯有智组-基于微量热曲线的微生物群落代谢特征分析
- nginx监听事件流程
- 【OpenCV 例程200篇】78. 频率域图像滤波基础
- P1287 盒子与球(python3实现)
- 千寻的计算机字符,转义字符变量与赋值
- 两种方法清空memcache
- 篮球战术谈之经典配合
- 计算机的桌面助手,正规的电脑桌面一键整理助手
- 简单的朴素贝叶斯算法实现英文文本分类(Python实现)
- python--spilt和strip用法
- html英文日期js,JS网页上显示中英文版日期时间(根据电脑上的时间)
- ABBYY FineReader 15标准版OCR文字识别及PDF编辑软工具
- 自带设备(BYOD)能用零信任框架吗?
- 1468:OKR-Periods of Words(kmp算法)
- 【大疆无人机OnboardSDK(三)妙算ssh远程控制台系统搭建】
- Elasticsearch实战-实现Hotel索引库的自动补全、拼音搜索功能
- 多边形快速凸包算法(Melkman‘s Algorithm)
- Linux JDK 卸载安装
- vue项目实现前端打印功能
- Eclipse下执行Python文件出现SyntaxError: Non-UTF-8 code starting with '\xb4' in file