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); } } } } } }
