手摸手实现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 [email protected]: 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 [email protected]'%' IDENTIFIED with mysql_native_password BY 'canal';
 
 #授权canal用户允许远程到mysql实现主从复制
 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
 CREATE USER [email protected]'%' IDENTIFIED with mysql_native_password BY 'remote';
 grant all privileges on *.* to [email protected]'%';
 
 #初始化数据库
 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(&quot;当次抓取没有数据，休息一会。。。。。。&quot;);
 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(&quot;Table:&quot; + tableName +
 &quot;,EventType:&quot; + eventType +
 &quot;,Before:&quot; + beforeData +
 &quot;,After:&quot; + afterData);
 }
 } else {
 System.out.println(&quot;当前操作类型为：&quot; + entryType);
 }
 }
 }
 }
 }
}


==靓文推荐==

分布式ID的常用解决方案-一把拿下

Mysql主从同步及主从同步延迟解决方案