最近在工作中需要处理一些大数据量同步的场景,正好运用到了canal这款数据库中间件,因此特意花了点时间来进行该中间件的的学习和总结。

背景介绍

早期,阿里巴巴B2B公司因为存在杭州和美国双机房部署,存在跨机房同步的业务需求。不过早期的数据库同步业务,主要是基于trigger的方式获取增量变更,不过从2010年开始,阿里系公司开始逐步的尝试基于数据库的日志解析,获取增量变更进行同步,由此衍生出了增量订阅&消费的业务,从此开启了一段新纪元。

适用版本

支持mysql5.7及以下版本

传统的主从同步原理

master将数据记录到了binlog日志里面,然后slave会通过一个io线程去读取master那边指定位置点开始的binlog日志内容,并将相应的信息写会到slave这边的relay日志里面,最后slave会有单独的sql线程来读取这些master那边执行的sql语句记录,达成两端的数据同步。

传统的mysql主从同步实现的原理图如下所示:

Canal中间件功能

基于纯java语言开发,可以用于做增量数据订阅和消费功能。

相比于传统的数据同步,我们通常需要进行先搭建主从架构,然后使用binlog日志进行读取,然后指定需要同步的数据库,数据库表等信息。但是随着我们业务的不断复杂,这种传统的数据同步方式以及开始变得较为繁琐,不够灵活。

canal模拟mysql slave的交互协议,伪装自己为mysql slave,向mysql master发送dump协议mysql master收到dump请求,开始推送binary log给slave(也就是canal),canal解析binary log对象(原始为byte流),通过对binlog数据进行解析即可获取需要同步的数据,在进行同步数据的过程中还可以加入开发人员的一些额外逻辑处理,比较开放。

Binlog的三种基本类型分别为:

STATEMENT模式只记录了sql语句,但是没有记录上下文信息,在进行数据恢复的时候可能会导致数据的丢失情况

ROW模式除了记录sql语句之外,还会记录每个字段的变化情况,能够清楚的记录每行数据的变化历史,但是会占用较多的空间,需要使用mysqlbinlog工具进行查看。

MIX模式比较灵活的记录,例如说当遇到了表结构变更的时候,就会记录为statement模式。当遇到了数据更新或者删除情况下就会变为row模式

Canal环境搭建

需要先登录mysql数据库,检查binlog功能是否有开启。

mysql> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | OFF |
+---------------+-------+
1 row in set (0.00 sec)

如果显示状态为OFF表示该功能未开启,那么这个时候就需要到my.ini里面进行相关配置了,在原来的my.ini配置底部插入以下内容:

server-id=192
log-bin=mysql-bin
binlog_format = ROW

当再次通过客户端查看log_bin状态为ON的时候,就表示binlog已经开启:

mysql> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+
1 row in set (0.00 sec)

然后在mysql里面添加以下的相关用户和权限:

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
GRANT SHOW VIEW, SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

开启之后,我们可以前往canal的官方地址进行相应版本的安装包进行下载:
https://github.com/alibaba/canal/releases

下载好指定的版本之后,找到里面的bin目录底下的startup脚本,启动。

启动之后会发现黑窗停止在这样一行的内容上,然后就不动了

Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0
Listening for transport dt_socket at address: 9099

这时候需要前往日志文件夹底下canallogs,查看canal日志文件是否已经开启,如果显示以下内容,就表示启动已经成功

2019-05-06 10:41:56.116 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2019-05-06 10:41:56.144 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2019-05-06 10:41:56.145 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2019-05-06 10:41:56.233 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.164.1:11111]
2019-05-06 10:41:58.179 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now .....

canal server的默认端口号为:11111,如果需要调整的话,可以去到conf目录底下的canal.properties文件中进行修改。

启动了canal的server之后,便是基于java的客户端搭建了。

首先在canalconf目录底下创建一个独立的文件夹(文件命名 idea_user_data),用于做额外的数据源配置:

然后创建一份特定的properties文件:(名称最好为:instance.properties),这里面只需要创建properties文件即可,其余几份文件会自动生成,instance.properties可以直接从example文件夹里面进行copy。

首先是导入相应的依赖文件:

<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency>

单机版本的canal连接案例

单机版本的环境比较好搭建,相应的代码如下:

首先是canal客户端的配置类

/**
* @author idea
* @date 2019/5/6
* @Version V1.0
*/
public class CanalConfig { public static String CANAL_ADDRESS="127.0.0.1"; public static int PORT=11111; public static String DESTINATION="idea_user_data"; public static String FILTER=".*\..*";
}

客户端代码:

package com.sise.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 java.net.InetSocketAddress;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue; import static com.sise.config.CanalConfig.*; /**
* @author idea
* @date 2019/5/6
* @Version V1.0
*/
public class CanalClient { private static Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>(); public static void main(String args[]) { CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_ADDRESS,
PORT), DESTINATION, "", "");
int batchSize = 1000;
try {
connector.connect();
connector.subscribe(FILTER);
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() >= 10) {
executeQueueSql();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
} finally {
connector.disconnect();
}
} /**
* 模拟执行队列里面的sql语句
*/
public static void executeQueueSql() {
int size = SQL_QUEUE.size();
for (int i = 0; i < size; i++) {
String sql = SQL_QUEUE.poll();
System.out.println("[sql]----> " + sql);
}
} /**
* 数据处理
*
* @param entrys
*/
private static 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 static 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().getSchemaName() + "." + 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 static 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().getSchemaName() + "." + 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 static 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().getSchemaName() + "." + 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();
}
} }

启动程序之后,我们对数据库表进行10次左右的修改操作之后,便可以从控制台中看到sql的打印信息。

关于canal集群搭建的一些坑

在实际开发中,如果只有一台canal机器作为server,当该台机器挂掉之后,服务就会终止,那么这个时候我们便需要引入集群部署的方式了。

搭建canal集群的环境需要先搭建好相应的zk集群模式。zk的集群搭建网上资料很多,这里就不进行讲解了。

canal搭建集群的一些资料可以参考以下链接:
https://github.com/alibaba/canal/wiki/AdminGuide

canal在搭建HA模式的时候有几个容易掉坑的步骤:
canal.properties配置里面需要添加zk的地址,同时canal.instance.global.spring.xml

需要修改为classpath:spring/default-instance.xml

每台机子的canal里面的具体instance所在目录的名称需要统一,每个实例都有对应的slaveId,他们的id需要保证不重复。搭建好了canal集群环境之后,然后代码部分需要在链接的那个模块进行稍微的调整:

CanalConnector connector = CanalConnectors.newClusterConnector(CLUSTER_ADDRESS, DESTINATION, "", "");

为了保证master在某些特殊场景下挂掉,mysql需要搭建为双M模式,那么我们这个时候可以在每个canal机器的instance配置文件中加入master的地址和standby的地址:

canal.instance.master.address=******
canal.instance.standby.address = ******

同时对于detecing也需要进行配置修改

canal.instance.detecting.enable = true ## 需要开启心跳检查
canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() ##心跳检查sql
canal.instance.detecting.interval.time = 3 ##心跳检查频率
canal.instance.detecting.retry.threshold = 3 ## 心跳检查失败次数阀值,当超过这个次数之后,就会自动切换到standby上边的机器进行binlog的订阅读取
canal.instance.detecting.heartbeatHaEnable = true ## 是否开启master和standby的主动切换

ps: master和standby进行切换机器的时候可能会有时间延迟。

启动2台canal机器,可以在zk里面查看到canal注册的节点信息:

通过模拟测试,关闭当前端口为11111的canal机器,节点信息会自动更换为第二台canal进行替换:

ClusterCanalConnector和SimpleCanalConnector类发现了username和password的参数,但是似乎具体配置中并没有做具体的设置,这是为什么呢?

后来也在github上边查看到了一些网友的相关讨论:

canal结合kafka发送sql数据案例

pom依赖:

     <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>1.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.0.1</version>
</dependency>

kafka的配置类:

public class KafkaProperties
{
public final static String ZK_CONNECTION = "XXX.XXX.XXX.XXX:2181";
public final static String BROKER_LIST_ADDRESS = "XXX.XXX.XXX.XXX:9092";
public final static String GROUP_ID = "group1";
public final static String TOPIC = "USER-DATA";
}

关于kafka的环境搭建步骤比较简单,网上有很多的资料,这里就不多一一介绍了。
首先是kafka的producer部分代码:

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.log4j.Logger; import java.util.Properties; import static com.sise.kafka.KafkaProperties.TOPIC; /**
* @author idea
* @date 2019/5/7
* @Version V1.0
*/
public class KafkaProducerDemo extends Thread { public static Logger log = Logger.getLogger(KafkaProducerDemo.class); //kafka的链接地址要使用hostname 默认9092端口
private static final String BROKER_LIST = BROKER_LIST_ADDRESS; private static KafkaProducer<String, String> producer = null; static {
Properties configs = initConfig();
producer = new KafkaProducer<String, String>(configs);
} /*
初始化配置
*/
private static Properties initConfig() {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
return properties;
} public static void sendMsg(String msg) {
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, msg);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (null != e) {
log.info("send error" + e.getMessage());
} else {
System.out.println("send success");
}
}
});
} }

接着是consumer部分的代码:

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector; import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties; /**
* @author idea
* @date 2019/5/7
* @Version V1.0
*/
public class KafkaConsumerDemo extends Thread { private final ConsumerConnector consumer;
private final String topic; public KafkaConsumerDemo(String topic) {
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
createConsumerConfig());
this.topic = topic;
} private static ConsumerConfig createConsumerConfig() {
Properties props = new Properties();
props.put("zookeeper.connect", KafkaProperties.ZK_CONNECTION);
props.put("group.id", KafkaProperties.GROUP_ID);
props.put("zookeeper.session.timeout.ms", "40000");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
return new ConsumerConfig(props);
} @Override
public void run() {
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, new Integer(1));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) {
System.out.println("【receive】" + new String(it.next().message()));
}
} }

然后需要在CanalClient 的executeQueueSql函数出进行部分功能的修改:

 /**
* 给kafka发送sql语句
*/
public static void executeQueueSql() {
int size = SQL_QUEUE.size();
for (int i = 0; i < size; i++) {
String sql = SQL_QUEUE.poll();
//发送sql给kafka
KafkaProducerDemo.sendMsg(sql);
}
}

为了验证程序是否正常,启动canal和kafka之后,对canal监听的数据库里面的表进行数据信息的修改,然后canal会将修改的binlog里面的sql放入队列中,当队列满了之后便向kafka中进行发送:

consumer端接受到数据之后控制台便打印出相应内容:

阿里Canal框架(数据同步中间件)初步实践的更多相关文章

  1. 增量数据同步中间件DataLink分享(已开源)

    项目介绍 名称: DataLink['deitə liŋk]译意: 数据链路,数据(自动)传输器语言: 纯java开发(JDK1.8+)定位: 满足各种异构数据源之间的实时增量同步,一个分布式.可扩展 ...

  2. 网站运维之JAVA-SSH框架数据同步问题

    一.环境 SSH环境,查询用的是基于Hibernate的配置文件构建了一个SessionFactory,主要代码如下 public class HibernateUtil { private stat ...

  3. 微服务之数据同步Porter

    Porter是一款数据同步中间件,主要用于解决同构/异构数据库之间的表级别数据同步问题. 背景 在微服务架构模式下深刻的影响了应用和数据库之间的关系,不像传统多个服务共享一个数据库,微服务架构下每个服 ...

  4. Mysql、ES 数据同步

    数据同步中间件 不足:不支持 ES6.X 以上.Mysql 8.X 以上 ime 标识最大时间 logstash全量.增量同步解决方案 https://www.elastic.co/cn/downlo ...

  5. 开源数据同步神器——canal

    前言 如今大型的IT系统中,都会使用分布式的方式,同时会有非常多的中间件,如redis.消息队列.大数据存储等,但是实际核心的数据存储依然是存储在数据库,作为使用最广泛的数据库,如何将mysql的数据 ...

  6. 美团DB数据同步到数据仓库的架构与实践

    背景 在数据仓库建模中,未经任何加工处理的原始业务层数据,我们称之为ODS(Operational Data Store)数据.在互联网企业中,常见的ODS数据有业务日志数据(Log)和业务DB数据( ...

  7. DB 数据同步到数据仓库的架构与实践

    背景 在数据仓库建模中,未经任何加工处理的原始业务层数据,我们称之为ODS(Operational Data Store)数据.在互联网企业中,常见的ODS数据有业务日志数据(Log)和业务DB数据( ...

  8. 基于 MySQL Binlog 的 Elasticsearch 数据同步实践 原

    一.背景 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品.订单等数据的多维度检索. 使用 Elasticsearch 存储业务数据可以 ...

  9. 阿里HBase的数据管道设施实践与演进

    摘要:第九届中国数据库技术大会,阿里巴巴技术专家孟庆义对阿里HBase的数据管道设施实践与演进进行了讲解.主要从数据导入场景. HBase Bulkload功能.HImporter系统.数据导出场景. ...

随机推荐

  1. 探究toString()和valueOf()

    1.用法如下:toString()方法:返回对象的字符串表示. 对象 操作 Array 将 Array 的元素转换为字符串.结果字符串由逗号分隔,且连接起来. Boolean 如果 Boolean 值 ...

  2. 为 WSUS 服务器定期运行清理向导

    在 WSUS 的管理界面的 Options 里面,可以找到 Server Cleanup Wizard 然后运行.后来想了一下,为什么不把它弄成定期运行呢! 找了一下,从 Windows Server ...

  3. Gradient Boosting Decision Tree学习

    Gradient Boosting Decision Tree,即梯度提升树,简称GBDT,也叫GBRT(Gradient Boosting Regression Tree),也称为Multiple ...

  4. C语言中static作用

    在C语言中,static的字面意思很容易把我们导入歧途,其实它的作用有三条. (1)先来介绍它的第一条也是最重要的一条:隐藏. 当我们同时编译多个文件时,所有未加static前缀的全局变量和函数都具有 ...

  5. 支持MySelf

    编程思路分享,BUG上报,主推Java Web方向与软件架构设计,不定期推出系列针对性基础教程,项目均放置于GitHub,个人运营精力有限,感谢支持. 交流群:628793702 个人技术公众,欢迎关 ...

  6. mysqlfrm

    mysqlfrm可基于frm文件生成对应的表结构.常用于数据恢复场景. 其有两种操作模式. 1. 创建一个临时实例来解析frm文件. 2. 使用诊断模式解析frm文件. 以下表进行测试,看看, 1.  ...

  7. Hibernate Validator注解大全

    hibernate Validator 是 Bean Validation 的参考实现 .Hibernate Validator 提供了 JSR 303 规范中所有内置 constraint 的实现, ...

  8. error_page 改变状态码为新的状态码,并显示指定内容

    server { listen 80; server_name www.espressos.com; location / { root /data0/www/bbs; index index.htm ...

  9. busybox hexdump 命令使用

    http://blog.sina.com.cn/s/blog_a6559d920101gvlk.html hexdump命令是Linux下的打印16进制的利器,它可以按我们指定的格式输出16进制,特别 ...

  10. 使用多字节字符集的跨平台(PC、Android、IOS、WP)编码/解码方法

    随着移动端的发展,跨平台已成为通讯架构设计的重要考虑因素,PC.Android.IOS.WP等跨多平台间的数据通讯,必然要解决字符编码/解码的问题. 多字节字符集MBCS不是跨平台的首选字符集,面向跨 ...