转载来源：https://juejin.cn/post/6844904073213247496

大约两年以前，笔者在一个项目中遇到了数据同步的难题。
当时，系统部署了几十个实例，分为1个中心平台和N个分中心平台，而每一个系统都对应一个单独的数据库实例。
在数据库层面，有这样一个需求：

中心平台数据库要包含所有系统平台的数据。
分中心数据库只包含本系统平台的数据。
在中心平台可以新增或修改分中心平台的数据，但要讲数据实时同步到对应的分中心平台数据库。

这几十个数据库实例之间，没有明确的主从关系，是否同步还要看数据的来源，所以并不能用MySQL的主从同步来做。
当时，笔者实验了几种方式，最后采用的方式是基于Mybatis拦截器机制 + 消息队列的方式来做的。
大概原理是通过Mybatis拦截器，拦截到事务操作，比如新增、修改和删除，根据自定义的数据主键(标识数据来源和去向)，封装成对象，投递到消息队列对应的topic中去。然后，每个系统监听不同的topic，消费数据并同步到数据库。
在此后的一段时间里，知道了canal这个开源组件。发现它更直接，它可以从MySQL的binlog中解析数据，投递到消息队列或其它地方。
一、canal简介
说起canal，也是阿里巴巴存在数据同步的业务需求。所以从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。
基于日志增量订阅&消费支持的业务：

数据库镜像
数据库实时备份
多级索引 (卖家和买家各自分库索引)
search build
业务cache刷新
价格变化等重要业务消息

我们正可以基于canal的机制，来完成一系列如数据同步、缓存刷新等业务。
二、启动canal
1、修改MySQL配置
对于自建的MySQL服务, 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下：
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
复制代码然后创建一个账户，用来链接MySQL，作为 MySQL slave 的权限。
CREATE USER canal IDENTIFIED BY ‘canal’;
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON . TO ‘canal’@’%’;
– GRANT ALL PRIVILEGES ON . TO ‘canal’@’%’ ;
FLUSH PRIVILEGES;
复制代码2、下载
下载canal非常简单，访问 releases页面选择需要的包下载，然后将下载的包解压到指定的目录即可。
tar -zxvf canal.deployer-1.1.4.tar.gz -C /canal
解压完成后，我们可以看到这样一个目录：

3、修改配置
在启动之前，还需要修改一些配置信息。
首先，定位到canal/conf/example ，编辑instance.properties配置文件，重点有几项：
canal.instance.mysql.slaveId=1234 # canal模拟slaveid
canal.instance.master.address=127.0.0.1:3306 # MySQL数据库地址
canal.instance.dbUsername=canal # 作为slave角色的账户
canal.instance.dbPassword=canal # 作为slave角色的账户密码
canal.instance.connectionCharset = UTF-8 # 数据库编码方式对应Java中的编码类型
canal.instance.filter.regex=.\… # 表过滤的表达式
canal.mq.topic=example # MQ 主题名称
复制代码我们希望canal监听到的数据，要发送到消息队列中，还需要修改canal.properties文件，在这里主要是MQ的配置。在这里笔者使用的是阿里云版RocketMQ，参数如下：

配置ak/sk

canal.aliyun.accessKey = XXX
canal.aliyun.secretKey = XXX

配置topic

canal.mq.accessChannel = cloud
canal.mq.servers = 内网接入点
canal.mq.producerGroup = GID_**group（在后台创建）
canal.mq.namespace = rocketmq实例id
canal.mq.topic=（在后台创建）
复制代码4、启动
直接运行启动脚本即可运行：./canal/bin/startup.sh 。然后打开logs/canal/canal.log文件，可以看到启动效果。
2020-02-26 21:12:36.715 [main] INFO com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
2020-02-26 21:12:36.746 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.44.128(192.168.44.128):11111]
2020-02-26 21:12:37.406 [main] INFO com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now …
复制代码三、启动MQ监听
我们把canal监听到的数据，投送到了消息队列中，那么接下来就是写个监听程序来消费其中的数据。
为了方便，笔者直接使用的是阿里云版RocketMQ，测试代码如下：
public static void main(String[] args) {
Properties properties = new Properties();
// 您在控制台创建的 Group ID
properties.put(PropertyKeyConst.GROUP_ID, “GID_CANAL”);
// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.AccessKey, “accessKey”);
// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, “secretKey”);
// 设置 TCP 接入域名，到控制台的实例基本信息中查看
properties.put(PropertyKeyConst.NAMESRV_ADDR,“http://MQ_INST_xxx.mq-internet.aliyuncs.com:80”);
// 集群订阅方式（默认）
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe(“example”,"*",new CanalListener());
consumer.start();
logger.info(“Consumer Started”);
}
复制代码四、测试
把环境都部署好之后，我们进入测试阶段来看一看实际效果。
我们以一张t_account表为例，这里面记录着账户id和账户余额。
首先，我们新增一条记录，insert into t_account (id,user_id,amount) values (4,4,200);
此时，MQ消费到数据如下：
{
“data”: [{
“id”: “4”,
“user_id”: “4”,
“amount”: “200.0”
}],
“database”: “seata”,
“es”: 1582723607000,
“id”: 2,
“isDdl”: false,
“mysqlType”: {
“id”: “int(11)”,
“user_id”: “varchar(255)”,
“amount”: “double(14,2)”
},
“old”: null,
“pkNames”: [“id”],
“sql”: “”,
“sqlType”: {
“id”: 4,
“user_id”: 12,
“amount”: 8
},
“table”: “t_account”,
“ts”: 1582723607656,
“type”: “INSERT”
}
复制代码通过数据可以看到，这里面详细记录了数据库的名称、表的名称、表的字段和新增数据的内容等。
然后，我们还可以把这条数据修改一下：update t_account set amount = 150 where id = 4;
此时，MQ消费到数据如下：
{
“data”: [{
“id”: “4”,
“user_id”: “4”,
“amount”: “150.0”
}],
“database”: “seata”,
“es”: 1582724016000,
“id”: 3,
“isDdl”: false,
“mysqlType”: {
“id”: “int(11)”,
“user_id”: “varchar(255)”,
“amount”: “double(14,2)”
},
“old”: [{
“amount”: “200.0”
}],
“pkNames”: [“id”],
“sql”: “”,
“sqlType”: {
“id”: 4,
“user_id”: 12,
“amount”: 8
},
“table”: “t_account”,
“ts”: 1582724016353,
“type”: “UPDATE”
}
复制代码可以看到，除了修改后的内容，canal还用old字段记录了修改前字段的值。
最后，我们删除这条数据：delete from t_account where id = 4;
相应的，MQ消费到数据如下：
{
“data”: [{
“id”: “4”,
“user_id”: “4”,
“amount”: “150.0”
}],
“database”: “seata”,
“es”: 1582724155000,
“id”: 4,
“isDdl”: false,
“mysqlType”: {
“id”: “int(11)”,
“user_id”: “varchar(255)”,
“amount”: “double(14,2)”
},
“old”: null,
“pkNames”: [“id”],
“sql”: “”,
“sqlType”: {
“id”: 4,
“user_id”: 12,
“amount”: 8
},
“table”: “t_account”,
“ts”: 1582724155370,
“type”: “DELETE”
}
复制代码监听到数据库表的变化之后，就可以根据自己的业务场景，对这些数据进行业务上的处理啦。
五、总结
可以看到，利用canal组件可以很方便的完成对数据变化的监听。如果利用消息队列来做数据同步的话，只有一点需要格外注意，即消息顺序性的问题。
binlog本身是有序的，但写入到mq之后如何保障顺序是值得关注的问题。
在mq顺序性问题这里，可以看到canal的消费顺序性相关解答。

作者：清幽之地
链接：https://juejin.cn/post/6844904073213247496
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

ES MQ canal同步mysql相关推荐

Elasticsearch7.9集群部署，head插件，canal同步mysql数据到es，亲自测试，无坑
Elasticsearch集群部署 1.服务器规划 10.4.7.11 node1 10.4.7.12 node2 10.4.7.13 node3 1. 集群相关一个运行中的 Elastics ...
canal同步mysql到es
canal 主要用途是基于 **MySQL 数据库增量日志解析**,提供**增量数据订阅和消费**. 可以简单地把canal理解为一个用来**同步增量数据的一个工具**. 工作原理 canal的工作 ...
使用canal同步MySQL数据到Elasticsearch(ES)
目录 1.功能及使用场景 1.1.功能介绍 1.2.使用场景 2.需求引入 3.canal文件下载及准备 3.1 下载文件 3.2 准备文件 4.deployer安装及效果测试 4.1.deploye ...
Flink 最佳实践之使用 Canal 同步 MySQL 数据至 TiDB
摘要:本文将介绍如何将 MySQL 中的数据,通过 Binlog + Canal 的形式导入到 Kafka 中,继而被 Flink 消费的案例.内容包括: 背景介绍环境介绍部署 TiDB Clus ...
canal同步mysql数据到rocketmq集群
rockermq多主多从异步复制部署参考 canal github 早期阿里巴巴因为杭州和美国双机房部署,存在跨机房同步的业务需求,实现方式主要是基于业务 trigger 获取增量变更.从 2010 ...
Canal同步mysql binlog至pulsar
Canal 一.简介 canal [kə'næl],主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费基于日志增量订阅和消费的业务包括数据库镜像数据库实时备份索引构建和实时 ...
canal同步mysql数据
canal简介 canal官网文档 canal [kə'næl],译意为水道/管道/沟渠,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费早期阿里巴巴因为杭州和美国双机房部署, ...
全网最详细、最简单 canal同步mysql 监听单实例、监听多实例配置
修改涉及的文件 deployer /conf/canal.properties /conf/example/instance.properties canal-adapter /conf/applic ...
canal同步mysql到kafka_使用Canal同步MySQL数据到Kafka 得到的数据中sql字段无值-问答-阿里云开发者社区-阿里云...
这个应该跟你的binlog记录模式有关系,binlog有3中模式,ROW(行模式), Statement(语句模式), Mixed(混合模式)三种模式的用法如下: ROW(行模式):记录那条数据修改了 ...

ES MQ canal同步mysql

配置ak/sk

配置topic

ES MQ canal同步mysql相关推荐

最新文章

热门文章