使用canal实现数据实时同步

canal
在这里插入图片描述

canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费

早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。

基于日志增量订阅和消费的业务包括

数据库镜像 数据库实时备份 索引构建和实时维护(拆分异构索引、倒排索引等) 业务 cache 刷新 带业务逻辑的增量数据处理 当前的
canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

工作原理
在这里插入图片描述

  • MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log
    events，可以通过 show binlog events 进行查看)

  • MySQL slave 将 master 的 binary
    log events 拷贝到它的中继日志(relay log) MySQL slave 重放 relay log

  • MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
    canal工作原理

  • MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
    canal工作原理

  • MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )

  • canal 解析 binary log 对象(原始为 byte 流)

我自己的应用场景是在统计分析功能中，采用了微服务调用的方式获取统计数据，但是这样耦合度很高，效率相对较低，我现在采用Canal数据库同步工具，通过实时同步数据库的方式实现，例如我们要统计每天注册与登录人数，我们只需要把会员表同步到统计库中，实现本地统计就可以了，这样效率更高，耦合度更低。
Canal是阿里巴巴旗下的一款开源项目，纯Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。

Canal环境搭建
canal的原理是基于mysql binlog技术，所以这里要开启mysql的binlog写入功能
在linux系统中，开启mysql服务：systemctl start mysqld或者service mysql start

检查binlog功能是否开启
在这里插入图片描述
开启binlog功能
如果显示状态为OFF表示该功能尚未开启，开启binlog功能

修改mysql的配置文件my.cnf

vim /etc/my.cnf

追加内容

log-bin=mysql-bin     #binlog文件名
binlog_format=ROW     #选择row模式
server_id=1           #mysql实例id,不能和canal的slaveId重复

在这里插入图片描述

重启mysql

systemctl restart mysqld

再次登录mysql客户端，查看log_bin变量
在这里插入图片描述
显示为ON表示该功能已开启。

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

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

这其实是添加了能远程访问mysql数据库的用户，账号和密码都是canal,由于我的虚拟机本来就添加过root用户，这里我就不再添加这个canal了，你根据自己情况。

下载安装Canal服务
下载canal地址

下载之后，放到目录中，解压文件
在这里插入图片描述
解压

tar zxvf canal.deployer-1.1.4.tar.gz -C /usr/local/canal/

在这里插入图片描述
修改配置文件

vim conf/example/instance.properties

在这里插入图片描述

这里是引用注： mysql 数据解析关注的表，Perl正则表达式. 多个正则之间以逗号(,)分隔，转义符需要双斜杠()
1.常见例子：所有表：.* or .\…
2.canal schema下所有表： canal\…*
3.canal下的以canal打头的表：canal\.canal.*
4…canal schema下的一张表：canal.test1
5.多个规则组合使用：canal\…*,mysql.test1,mysql.test2 (逗号分隔) 注意：此过滤条件只针对row模式的数据有效(ps.
6.mixed/statement因为不解析sql，所以无法准确提取tableName进行过滤)

进入bin目录下启动

./startup.sh

在这里插入图片描述
代码整合 创建canal_client模块

引入相关依赖

创建application.properties配置文件

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--mysql--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>commons-dbutils</groupId><artifactId>commons-dbutils</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId></dependency>
</dependencies>

# 服务端口
server.port=10000
# 服务名
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://localhost:3306/guli?serverTimezone=GMT%2B8
spring.datasource.username=root
spring.datasource.password=123456

编写canal客户端类

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.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import javax.sql.DataSource;
import java.net.InetSocketAddress;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;@Component
public class CanalClient {//sql队列private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();@Resourceprivate DataSource dataSource;/*** canal入库方法*/public void run() {CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.159.33",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<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 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().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 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().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<RowData> rowDatasList = rowChange.getRowDatasList();for (RowData rowData : rowDatasList) {List<Column> 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);}}
}

这个地方的ip你改成你自己虚拟机或者服务器上的。
在这里插入图片描述

@SpringBootApplication
public class CanalApplication implements CommandLineRunner {@Resourceprivate CanalClient canalClient;public static void main(String[] args) {SpringApplication.run(CanalApplication.class, args);}@Overridepublic void run(String... strings) throws Exception {//项目启动，执行canal客户端监听canalClient.run();}
}

测试之前需要
在linux系统中插入一条数据测试
在这里插入图片描述

看本地控制台
在这里插入图片描述

在Linux中更新以下上面那条数据
在这里插入图片描述

看本地控制台
在这里插入图片描述

看下本地windows的mysql数据库表中数据是否和linux上面的数据一致
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/122010.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

linux安装gitlab-runner最新保姆级教程

linux安装gitlab-runner最新保姆级教程

安装 安装教程来自gitlab官网&#xff0c;本文仅演示CentOS系统下安装gitlab-runner 自动安装 1.添加gitlab官方存储库 curl -L "https://packages.gitlab.com/install/repositories/runner/gitlab-runner/script.rpm.sh" | sudo bash也可以配置yum源安装 vim /…
阅读更多...
Android Studio编译旧的app代码错误及解决方法

Android Studio编译旧的app代码错误及解决方法

‘android.injected.build.density’ is deprecated. The option ‘android.injected.build.density’ is deprecated. It was removed in version 8.0 of the Android Gradle plugin. Density property injection from Android Studio has been removed. 解决 app/build.gr…
阅读更多...
【API篇】八、Flink窗口函数

【API篇】八、Flink窗口函数

文章目录 1、增量聚合之ReduceFunction2、增量聚合之AggregateFunction3、全窗口函数full window functions4、增量聚合函数搭配全窗口函数5、会话窗口动态获取间隔值6、触发器和移除器7、补充 //窗口操作 stream.keyBy(<key selector>).window(<window assigner>)…
阅读更多...
lesson2（补充）关于const成员函数

lesson2（补充）关于const成员函数

个人主页&#xff1a;Lei宝啊 愿所有美好如期而遇 前言&#xff1a; 将const 修饰的 “ 成员函数 ” 称之为 const 成员函数 &#xff0c; const 修饰类成员函数&#xff0c;实际修饰该成员函数 隐含的 this 指针 &#xff0c;表明在该成员函数中不能对类的任何成员进行修改…
阅读更多...
Vue+ElementUI项目打包部署到Ubuntu服务器中

Vue+ElementUI项目打包部署到Ubuntu服务器中

1、修改config/index.js中的assetsPublicPath: /,修改为assetsPublicPath: ./ assetsPublicPath: ./2、在build/utils.js中增加publicPath: ../../ publicPath: ../../3、打开终端&#xff0c;在根目录下执行npm run build进行打包&#xff0c;打包成功后会生成dist npm run…
阅读更多...
从lc560“和为 K 的子数组“带你认识“前缀和+哈希表“的解题思路

从lc560“和为 K 的子数组“带你认识“前缀和+哈希表“的解题思路

1 前缀和哈希表解题的几道题目&#xff1a;建议集中练习 560. 和为 K 的子数组&#xff1a;https://leetcode.cn/problems/subarray-sum-equals-k/ 1248. 统计「优美子数组」: https://leetcode.cn/problems/count-number-of-nice-subarrays/ 1249. 和可被 K 整除的子数组(利用…
阅读更多...
037-第三代软件开发-系统音量设置

037-第三代软件开发-系统音量设置

第三代软件开发-系统音量设置 文章目录 第三代软件开发-系统音量设置项目介绍系统音量设置QML 实现C 实现 总结一下 关键字&#xff1a; Qt、 Qml、 volume、 声音、 GPT 项目介绍 欢迎来到我们的 QML & C 项目&#xff01;这个项目结合了 QML&#xff08;Qt Meta-Obj…
阅读更多...
听GPT 讲Rust源代码--library/std(8)

听GPT 讲Rust源代码--library/std(8)

题图来自Why is Rust programming language so popular?[1] File: rust/library/std/src/sys/sgx/abi/reloc.rs 在Rust源代码中&#xff0c;sgx/abi/reloc.rs文件的作用是定义了针对Intel Software Guard Extensions (SGX)的重定位相关结构和函数。 该文件中的Rela 结构定义了…
阅读更多...
集群节点批量执行 shell 命令

集群节点批量执行 shell 命令

1、SSH 工具本身支持多窗口 比如 MobaXterm&#xff1a; 2、编写脚本通过 ssh 在多台机器批量执行shell命令 创建 ssh_hosts 配置文件&#xff0c;定义需要批量执行的节点&#xff08;必须能够通过 ssh 免密登录&#xff0c;且存在同名用户&#xff09; vim ssh_hostsbig…
阅读更多...
【Codeforces】 CF79D Password

【Codeforces】 CF79D Password

题目链接 CF方向 Luogu方向 题目解法 看到区间异或&#xff0c;一个经典的套路是做差分&#xff0c;我们即在 l l l 处异或一次&#xff0c;在 r 1 r1 r1 处异或一次&#xff0c;然后前缀和起来 于是我们可以将问题转化成&#xff1a;有一个序列初始全 0 0 0&#xff0c…
阅读更多...
Spring定时任务+webSocket实现定时给指定用户发送消息

Spring定时任务+webSocket实现定时给指定用户发送消息

生命无罪&#xff0c;健康万岁&#xff0c;我是laity。 我曾七次鄙视自己的灵魂&#xff1a; 第一次&#xff0c;当它本可进取时&#xff0c;却故作谦卑&#xff1b; 第二次&#xff0c;当它在空虚时&#xff0c;用爱欲来填充&#xff1b; 第三次&#xff0c;在困难和容易之…
阅读更多...
Oracle通过透明网关查询SQL Server 报错ORA-00904

Oracle通过透明网关查询SQL Server 报错ORA-00904

Oracle通过透明网关查询SQL Server 报错ORA-00904 问题描述&#xff1a; 只有全表扫描SELECT * 时SQL语句可以正常执行 添加WHERE条件或指定列名查询&#xff0c;查询语句就报错 问题原因&#xff1a; 字段大小写和SQLSERVER中定义的不一致导致查询异常 解决办法&#xff1a; 给…
阅读更多...
消息队列中间件面试笔记总结RabbitMQ，Kafka，RocketMQ

消息队列中间件面试笔记总结RabbitMQ，Kafka，RocketMQ

文章目录 (一) Rabbit MQRabbitMQ 核心概念消息队列的作用Exchange(交换器)Broker&#xff08;消息中间件的服务节点&#xff09;如何保证消息的可靠性如何保证 RabbitMQ 消息的顺序性如何保证 RabbitMQ 高可用的&#xff1f;如何解决消息队列的延时以及过期失效问题消息堆积问…
阅读更多...
Web服务器与Http协议

Web服务器与Http协议

Web服务器与Http协议 一.Web服务器 1.简介 Web服务器一般指网站服务器&#xff0c;也称之为WWW(World Wide Web)服务器Web服务器是指驻留于因特网上某种类型计算机的程序Web服务器不是硬件服务器&#xff0c;而是软件服务器。Web服务器其主要功能是提供网上信息浏览服务&…
阅读更多...
p5.js 视频播放指南

p5.js 视频播放指南

本文简介 在刚接触 p5.js 时我以为这只是一个艺术方向的 canvas 库&#xff0c;没想到它还支持视频文件和视频流的播放。 本文简单讲讲如何使用 P5.js 播放视频。 播放视频文件 p5.js 除了可以使用 video 元素播放视频外&#xff0c;还支持使用 image 控件播放视频。 方式1&…
阅读更多...
【机器学习可解释性】2.特征重要性排列

【机器学习可解释性】2.特征重要性排列

机器学习可解释性 1.模型洞察的价值2.特征重要性排列3.部分依赖图4.SHAP Value5.SHAP Value 高级使用 正文 前言 你的模型认为哪些特征最重要&#xff1f; 介绍 我们可能会对模型提出的最基本的问题之一是&#xff1a;哪些特征对预测的影响最大&#xff1f; 这个概念被称为…
阅读更多...
【C++】命名空间

【C++】命名空间

目录 1 命名空间的引入 2 命名空间的定义 3 标准命名空间std 头文件和std的关系 4 命名空间的使用 4.1 加命名空间名称及作用域限定符:: 4.2 使用using将命名空间中某个成员引入(最推荐) 4.3 使用using namespace 命名空间名称 5 总结 1 命名空间的引入 为了解决C语言中…
阅读更多...
ffmpeg的下载和编译(vs2022)

ffmpeg的下载和编译(vs2022)

感谢大佬的二创，直接提供了sln编译 ffmpeg二创地址 创建如下目录 build存放代码（build最好改成source，因为作者这么建议，编译完才发现） msvc存放第三方依赖的头文件，这里固定叫msvc，因为大佬的sln里查找的路径是这个，不嫌麻烦也可以自己改 下载代码和编译器 下载源码…
阅读更多...
shell实现部署ftp提供共享yum仓库

shell实现部署ftp提供共享yum仓库

shell 脚本实现脚本测试 脚本实现自动化配置vsftp&#xff0c;并共享opt目录给其它节点机器实现离线源配置 使用脚本需修改以下内容&#xff1a; ftp_ip为ftp服务器ip reponame为repo文件名字 dir为仓库目录&#xff0c;默认opt下 houzui为opt目录下的离线仓库&#xff0c;写入…
阅读更多...
简单的前端语言

简单的前端语言

目录 一.介绍 1.什么是前端 2.什么是后端 3.为什么要学习前端 4.前端学习哪些内容 二.HTTP协议 1.HTTP协议的四大特性&#xff1a; 2.响应状态码 三.HTML介绍 1.什么是html标签 2.html文档介绍 3.如何打开html文档 4.head标签中常用的标签 5.body中常用的标签 一.…
阅读更多...
最新文章

Copyright @ 2022~2023