手摸手实现 Canal 如何接入 MySQL 实现数据写操作监听

作者：知识浅谈

2022 年 7 月 29 日
本文字数：3419 字
阅读完需：约 11 分钟

🍁 作者：知识浅谈，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.log2022-03-31T04:20:25.133810Z 6 [Note] [MY-010454] [Server] A temporary password isgenerated for root@localhost: Jby&XTOc.7iNmysql ‐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);          }        }      }    }  }}