手摸手实现Canal如何接入MySQL实现数据写操作监听
【摘要】 教你实现Canal如何接入MySQL实现数据写操作监听
🍁 作者：知识浅谈，CSDN签约讲师，CSDN原力作者，后端领域优质创作者，热爱分享创作
💒 公众号：知识浅谈
📌 擅长领域：全栈工程师、爬虫、ACM算法
🔥 联系方式vx：zsqtcc
Canal的使用分为两部分：
- 上部分为canal接入Mysql数据库进行操作监听。
- 下部分为Canal如何接入消息队列、Redis、Canal的高可用HA。
这次拿上部分来做正菜🛴🛴🛴
环境：linux
数据库主机（192.168.31.230）
Canal server主机（192.168.31.231）
🎈安装MySQL 8.0.28（192.168.31.230）
- 防火墙放行3306端口
firewall‐cmd ‐‐zone=public ‐‐add‐port=3306/tcp ‐‐permanent firewall‐cmd ‐‐reload - 安装mysql厂库（注意是mysql厂库，不是mysql）
cd /home/ wget ‐‐no‐check‐certificate https://manongbiji.oss‐cn‐ beijing.aliyuncs.com/ittailkshow/canal/download/world.sql wget ‐‐no‐check‐certificate https://repo.mysql.com/mysql80‐community‐releas e‐el7‐5.noarch.rpm yum localinstall ‐y mysql80‐community‐release‐el7‐5.noarch.rpm - 安装Mysql
#自动安装MySQL 8.0.28 yum install ‐y mysql‐community‐server - 调整配置文件
sudo cat >> /etc/my.cnf <<‐'EOF' server‐id=1 log‐bin=mysql‐bin binlog_format=row binlog‐do‐db=world EOF systemctl start mysqld - 获取初始密码
grep 'temporary password' /var/log/mysqld.log
2022-03-31T04:20:25.133810Z 6 [Note] [MY-010454] [Server] A temporary password is
generated for root@localhost: Jby&XTOc.7iN
mysql ‐uroot ‐pJby&XTOc.7iN
- mysql连接后命令执行
#修改root密码 ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'asA S123456!'; #MySQL降低密码强度 set global validate_password.policy=0; set global validate_password.length=4; #创建canal同步账户 CREATE USER canal@'%' IDENTIFIED with mysql_native_password BY 'canal'; #授权canal用户允许远程到mysql实现主从复制 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; CREATE USER remote@'%' IDENTIFIED with mysql_native_password BY 'remote'; grant all privileges on *.* to remote@'%'; #初始化数据库 source /home/world.sql
🎈安装Canal-Server
- 安装JDK
yum ‐y install java‐1.8.0‐openjdk‐devel.x86_64 sudo cat >> /etc/profile <<‐'EOF' export JAVA_HOME=/usr/lib/jvm/java‐1.8.0‐openjdk export JRE_HOME=$JAVA_HOME/jre export CLASSPATH=$JAVA_HOME/lib:$JRE_HOME/lib:$CLASSPATH export PATH=$JAVA_HOME/bin:$JRE_HOME/bin:$PATH EOF source /etc/profile echo $JAVA_HOME - 下载canal-deployer最新版
# 下载脚本 wget ‐‐no‐check‐certificate https://manongbiji.oss‐cn‐ beijing.aliyuncs.com/ittailkshow/canal/download/canal.deployer‐ 1.1.5.tar.gz mkdir /home/canal tar zxvf canal.deployer‐1.1.5.tar.gz ‐C /home/canal - 修改canal配置文件
vi conf/example/instance.properties #调整serverId canal.instance.mysql.slaveId=10 # master地址 canal.instance.master.address=192.168.31.230:3306 ... # 关闭tsdb canal.instance.tsdb.enable=false #确认canal同步用的用户名、密码 canal.instance.dbUsername=canal canal.instance.dbPassword=canal - canal服务启动并放行相关端口
#启动服务 sh bin/startup.sh #canal admin 端口 开发数据监听程序 pom.xml 编写数据监听程序 firewall‐cmd ‐‐zone=public ‐‐add‐port=11110/tcp ‐‐permanent #canal 监听端口 firewall‐cmd ‐‐zone=public ‐‐add‐port=11111/tcp ‐‐permanent #canal 指标监控端口 firewall‐cmd ‐‐zone=public ‐‐add‐port=11112/tcp ‐‐permanent firewall‐cmd ‐‐reload - 查看canal日志
tail canal.log
🎈开发数据监听程序
pom.xml依赖
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.5</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>1.1.5</version>
</dependency>
</dependencies>
编写数据监听程序
package com.itlaoqi;
import com.alibaba.fastjson.JSONObject;
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.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;
public class AD {
public static void main(String[] args) throws InterruptedException, Invalid
ProtocolBufferException {
//TODO 获取连接
CanalConnector canalConnector = CanalConnectors.newSingleConnector(new Inet
SocketAddress("192.168.31.231", 11111), "example", "", "");
while (true) {
//TODO 连接
canalConnector.connect();
//TODO 订阅数据库
canalConnector.subscribe("world.*");
//TODO 获取数据
Message message = canalConnector.get(100);
//TODO 获取Entry集合
List<CanalEntry.Entry> entries = message.getEntries();
//TODO 判断集合是否为空,如果为空,则等待一会继续拉取数据
if (entries.size() <= 0) {
System.out.println("当次抓取没有数据，休息一会。。。。。。");
Thread.sleep(1000);
} else {
//TODO 遍历entries，单条解析
for (CanalEntry.Entry entry : entries) {
//1.获取表名
String tableName = entry.getHeader().getTableName();
//2.获取类型
CanalEntry.EntryType entryType = entry.getEntryType();
//3.获取序列化后的数据
ByteString storeValue = entry.getStoreValue();
//4.判断当前entryType类型是否为ROWDATA
if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
//5.反序列化数据
CanalEntry.RowChange rowChange =
CanalEntry.RowChange.parseFrom(storeValue);
//6.获取当前事件的操作类型
CanalEntry.EventType eventType = rowChange.getEventType();
//7.获取数据集
List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
//8.遍历rowDataList，并打印数据集
for (CanalEntry.RowData rowData : rowDataList) {
JSONObject beforeData = new JSONObject();
List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
for (CanalEntry.Column column : beforeColumnsList) {
beforeData.put(column.getName(), column.getValue());
}
JSONObject afterData = new JSONObject();
List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
for (CanalEntry.Column column : afterColumnsList) {
afterData.put(column.getName(), column.getValue());
}
//数据打印
System.out.println("Table:" + tableName +
",EventType:" + eventType +
",Before:" + beforeData +
",After:" + afterData);
}
} else {
System.out.println("当前操作类型为：" + entryType);
}
}
}
}
}
}
🎈==靓文推荐==🎈
🚀分布式ID的常用解决方案-一把拿下🚀
🚀Mysql主从同步及主从同步延迟解决方案🚀
【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）