• mysql的数据表同步工具 canal的使用

    一  canal的扫盲

    1.1 canal的介绍

    canal是阿里巴巴旗下的一款开源项目，使用java语言进行开发，基于数据库增量日志解析，提供增量数据订阅与消费的功能。是一款很好用的数据库同步工具。目前只支持mysql。

    二  canal的搭建

    2.1 架构流程

     2.2 配置服务器mysql

    canal的原理是基于mysql binlog技术，所以，这里一定要开启mysql的binlog写入的功能。
    1.开启mysql服务：service mysqld  start 或  service mysql start
    2.检测binlog功能是否开启，如果是off，则没有开启，如果是on表示开启
    show variables like 'log_bin';

    3.如果binlog的显示为off，则修改配置文件  my.cnf 进行配置开启

    vi   /etc/my.cnf

    1. # set zhucongfuzhi
    2. server_id = 86               # 设置服务器编号
    3. log_bin = master-bin        # 启用二进制日志，并设置二进制日志文件前缀 
    4. expire_logs_days=7          #自动清理 7 天前的log文件,可根据需要修改
    5. binlog_format=ROW           #选择row模式

    4.重启mysql数据库

    切换到 mysql的 隶属用户：hd-mysql

    [root@localhost local]# su hd-mysql
    [hd-mysql@localhost etc]$ service mysql start
    Starting MySQL. SUCCESS! 



    重启后，再查看binlog的值，为on，则表示已经开启了。

    5.创建远程访问用户，并授权访问

    进入mysql的命令模式：

    1. create user 'canal'@'%'IDENTIFIED BY 'boc123'
    2. grant all on *.* to 'canal'@'%'
    3. flush privileges;

    mysql> create user 'canal'@'%'IDENTIFIED BY 'boc123';
    Query OK, 0 rows affected (0.01 sec)

    mysql> grant all on *.* to 'canal'@'%';
    Query OK, 0 rows affected (0.01 sec)

    mysql> flush privileges;
    Query OK, 0 rows affected (0.02 sec)

    mysql> 
     

     2.3 配置安装canal同步工具

    1.软件包下载地址

    Releases · alibaba/canal · GitHub

    2.上传软件包到服务器 

     3.解压并修改配置文件

    将软件安装到：/usr/local/ 目录下 ，完整路径为 /usr/local/canal  这个目录

    [root@localhost local]# mkdir -p canal
    [root@localhost local]# cd canal/
    [root@localhost canal]# ls
    [root@localhost canal]# pwd
    /usr/local/canal
    [root@localhost canal]# tar -zxvf /root/export/servers/canal.deployer-1.1.6.tar.gz -C .
    bin/startup.bat
    bin/restart.sh
    bin/startup.sh
    bin/stop.sh
    conf/metrics/
    conf/example/
    4.修改配置文件

    vi conf/example/instance.properties

    [root@localhost example]# pwd
    /usr/local/canal/conf/example
    [root@localhost example]# vi instance.properties 
    修改内容如下

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

    3.进入bin目录下启动

    1.进入到安装目录： /usr/local/canal

    2.启动命令:  sh bin/startup.sh

    [root@localhost canal]# ./bin/startup.sh
    cd to /usr/local/canal/bin for workaround relative path
    LOG CONFIGURATION : /usr/local/canal/bin/../conf/logback.xml
    canal conf : /usr/local/canal/bin/../conf/canal.properties

    截图如下

      2.4  关闭防火墙

     2.5  编写接收java程序

    1.项目结构

     2.pom文件

    1. "1.0" encoding="UTF-8"?>
    2.  
    3. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.   <modelVersion>4.0.0modelVersion>
    6.   <parent>
    7.     <groupId>org.springframework.bootgroupId>
    8.     <artifactId>spring-boot-starter-parentartifactId>
    9.     <version>2.2.1.RELEASEversion>
    10.     <relativePath/> 
    11.   parent>
    12.   <groupId>com.ljfgroupId>
    13.   <artifactId>canal-demoartifactId>
    14.   <version>1.0-SNAPSHOTversion>
    15.  
    16.   <name>canal-demoname>
    17.   
    18.   <url>http://www.example.comurl>
    19.  
    20.   <properties>
    21.     <project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
    22.     <maven.compiler.source>1.8maven.compiler.source>
    23.     <maven.compiler.target>1.8maven.compiler.target>
    24.   properties>
    25.  
    26.   <dependencies>
    27.     <dependency>
    28.       <groupId>org.springframework.bootgroupId>
    29.       <artifactId>spring-boot-starter-webartifactId>
    30.     dependency>
    31.  
    32.     
    33.     <dependency>
    34.       <groupId>mysqlgroupId>
    35.       <artifactId>mysql-connector-javaartifactId>
    36.     dependency>
    37.  
    38.     <dependency>
    39.       <groupId>commons-dbutilsgroupId>
    40.       <artifactId>commons-dbutilsartifactId>
    41.       <version>1.7version>
    42.     dependency>
    43.  
    44.     <dependency>
    45.       <groupId>org.springframework.bootgroupId>
    46.       <artifactId>spring-boot-starter-jdbcartifactId>
    47.     dependency>
    48.  
    49.     <dependency>
    50.       <groupId>com.alibaba.ottergroupId>
    51.       <artifactId>canal.clientartifactId>
    52.       <version>1.1.0version>
    53.     dependency>
    54.   dependencies>
    55.  
    56.   <build>
    57.  
    58.   build>
    59. project>

    3.配置文件

    1. # 服务端口
    2. server.port=10010
    3. # 服务名
    4. spring.application.name=canal-client-t14
    5.  
    6. # 环境设置：dev、test、prod
    7. spring.profiles.active=dev
    8.  
    9. # mysql数据库连接
    10. spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    11. spring.datasource.url=jdbc:mysql://localhost:3306/xx_db?serverTimezone=GMT%2B8
    12. spring.datasource.username=root
    13. spring.datasource.password=cloudiip

    4.处理类

    1. package com.ljf.canal;
    2.  
    3. import com.alibaba.otter.canal.client.CanalConnector;
    4. import com.alibaba.otter.canal.client.CanalConnectors;
    5. import com.alibaba.otter.canal.protocol.CanalEntry.*;
    6. import com.alibaba.otter.canal.protocol.Message;
    7. import com.google.protobuf.InvalidProtocolBufferException;
    8. import org.apache.commons.dbutils.DbUtils;
    9. import org.apache.commons.dbutils.QueryRunner;
    10. import org.springframework.stereotype.Component;
    11.  
    12. import javax.annotation.Resource;
    13. import javax.sql.DataSource;
    14. import java.net.InetSocketAddress;
    15. import java.sql.Connection;
    16. import java.sql.SQLException;
    17. import java.util.Iterator;
    18. import java.util.List;
    19. import java.util.Queue;
    20. import java.util.concurrent.ConcurrentLinkedQueue;
    21.  
    22. @Component
    23. public class CanalClient {
    24.  
    25.     //sql队列
    26.     private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();
    27.  
    28.     @Resource
    29.     private DataSource dataSource;
    30.  
    31.     /**
    32.      * canal入库方法
    33.      */
    34.     public void run() {
    35.  
    36.         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.152.141",
    37.                 11111), "example", "canal", "boc123");
    38.         int batchSize = 1000;
    39.         try {
    40.             connector.connect();
    41.             connector.subscribe(".*\\..*");
    42.             connector.rollback();
    43.             try {
    44.                 while (true) {
    45.                     //尝试从master那边拉去数据batchSize条记录，有多少取多少
    46.                     Message message = connector.getWithoutAck(batchSize);
    47.                     long batchId = message.getId();
    48.                     int size = message.getEntries().size();
    49.                     if (batchId == -1 || size == 0) {
    50.                         Thread.sleep(1000);
    51.                     } else {
    52.                         dataHandle(message.getEntries());
    53.                     }
    54.                     connector.ack(batchId);
    55.  
    56.                     //当队列里面堆积的sql大于一定数值的时候就模拟执行
    57.                     if (SQL_QUEUE.size() >= 1) {
    58.                         executeQueueSql();
    59.                     }
    60.                 }
    61.             } catch (InterruptedException e) {
    62.                 e.printStackTrace();
    63.             } catch (InvalidProtocolBufferException e) {
    64.                 e.printStackTrace();
    65.             }
    66.         } finally {
    67.             connector.disconnect();
    68.         }
    69.     }
    70.  
    71.     /**
    72.      * 模拟执行队列里面的sql语句
    73.      */
    74.     public void executeQueueSql() {
    75.         int size = SQL_QUEUE.size();
    76.         for (int i = 0; i < size; i++) {
    77.             String sql = SQL_QUEUE.poll();
    78.             System.out.println("[sql]----> " + sql);
    79.  
    80.             this.execute(sql.toString());
    81.         }
    82.     }
    83.  
    84.     /**
    85.      * 数据处理
    86.      *
    87.      * @param entrys
    88.      */
    89.     private void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException {
    90.         for (Entry entry : entrys) {
    91.             if (EntryType.ROWDATA == entry.getEntryType()) {
    92.                 RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    93.                 EventType eventType = rowChange.getEventType();
    94.                 if (eventType == EventType.DELETE) {
    95.                     saveDeleteSql(entry);
    96.                 } else if (eventType == EventType.UPDATE) {
    97.                     saveUpdateSql(entry);
    98.                 } else if (eventType == EventType.INSERT) {
    99.                     saveInsertSql(entry);
    100.                 }
    101.             }
    102.         }
    103.     }
    104.  
    105.     /**
    106.      * 保存更新语句
    107.      *
    108.      * @param entry
    109.      */
    110.     private void saveUpdateSql(Entry entry) {
    111.         try {
    112.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    113.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    114.             for (RowData rowData : rowDatasList) {
    115.                 List<Column> newColumnList = rowData.getAfterColumnsList();
    116.                 StringBuffer sql = new StringBuffer("update " + entry.getHeader().getTableName() + " set ");
    117.                 for (int i = 0; i < newColumnList.size(); i++) {
    118.                     sql.append(" " + newColumnList.get(i).getName()
    119.                             + " = '" + newColumnList.get(i).getValue() + "'");
    120.                     if (i != newColumnList.size() - 1) {
    121.                         sql.append(",");
    122.                     }
    123.                 }
    124.                 sql.append(" where ");
    125.                 List<Column> oldColumnList = rowData.getBeforeColumnsList();
    126.                 for (Column column : oldColumnList) {
    127.                     if (column.getIsKey()) {
    128.                         //暂时只支持单一主键
    129.                         sql.append(column.getName() + "=" + column.getValue());
    130.                         break;
    131.                     }
    132.                 }
    133.                 SQL_QUEUE.add(sql.toString());
    134.             }
    135.         } catch (InvalidProtocolBufferException e) {
    136.             e.printStackTrace();
    137.         }
    138.     }
    139.  
    140.     /**
    141.      * 保存删除语句
    142.      *
    143.      * @param entry
    144.      */
    145.     private void saveDeleteSql(Entry entry) {
    146.         try {
    147.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    148.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    149.             for (RowData rowData : rowDatasList) {
    150.                 List<Column> columnList = rowData.getBeforeColumnsList();
    151.                 StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getTableName() + " where ");
    152.                 for (Column column : columnList) {
    153.                     if (column.getIsKey()) {
    154.                         //暂时只支持单一主键
    155.                         sql.append(column.getName() + "=" + column.getValue());
    156.                         break;
    157.                     }
    158.                 }
    159.                 SQL_QUEUE.add(sql.toString());
    160.             }
    161.         } catch (InvalidProtocolBufferException e) {
    162.             e.printStackTrace();
    163.         }
    164.     }
    165.  
    166.     /**
    167.      * 保存插入语句
    168.      *
    169.      * @param entry
    170.      */
    171.     private void saveInsertSql(Entry entry) {
    172.         try {
    173.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    174.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    175.             for (RowData rowData : rowDatasList) {
    176.                 List<Column> columnList = rowData.getAfterColumnsList();
    177.                 StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getTableName() + " (");
    178.                 for (int i = 0; i < columnList.size(); i++) {
    179.                     sql.append(columnList.get(i).getName());
    180.                     if (i != columnList.size() - 1) {
    181.                         sql.append(",");
    182.                     }
    183.                 }
    184.                 sql.append(") VALUES (");
    185.                 for (int i = 0; i < columnList.size(); i++) {
    186.                     sql.append("'" + columnList.get(i).getValue() + "'");
    187.                     if (i != columnList.size() - 1) {
    188.                         sql.append(",");
    189.                     }
    190.                 }
    191.                 sql.append(")");
    192.                 SQL_QUEUE.add(sql.toString());
    193.             }
    194.         } catch (InvalidProtocolBufferException e) {
    195.             e.printStackTrace();
    196.         }
    197.     }
    198.  
    199.     /**
    200.      * 入库
    201.      * @param sql
    202.      */
    203.     public void execute(String sql) {
    204.         Connection con = null;
    205.         try {
    206.             if(null == sql) return;
    207.             con = dataSource.getConnection();
    208.             QueryRunner qr = new QueryRunner();
    209.             int row = qr.execute(con, sql);
    210.             System.out.println("update: "+ row);
    211.         } catch (SQLException e) {
    212.             e.printStackTrace();
    213.         } finally {
    214.             DbUtils.closeQuietly(con);
    215.         }
    216.     }
    217. }
    218.

    5.启动类

    1. package com.ljf;
    2.  
    3. import com.ljf.canal.CanalClient;
    4. import org.springframework.boot.CommandLineRunner;
    5. import org.springframework.boot.SpringApplication;
    6. import org.springframework.boot.autoconfigure.SpringBootApplication;
    7.  
    8. import javax.annotation.Resource;
    9.  
    10. /**
    11.  * Hello world!
    12.  *
    13.  */
    14. @SpringBootApplication
    15. public class  App implements CommandLineRunner
    16. {
    17.     @Resource
    18.     private CanalClient canalClient;
    19.     public static void main( String[] args )
    20.     {
    21.         System.out.println( "Hello World!" );
    22.         SpringApplication.run(App.class, args);
    23.     }
    24.  
    25.     @Override
    26.     public void run(String... strings) throws Exception {
    27.         //项目启动，执行canal客户端监听
    28.         canalClient.run();
    29.     }
    30. }

    6.启动服务

      2.6 测试验证

    1.在目的端的数据库，新建一个同样名字的数据库，同样名字的数据表。

    如这里源数据库 xx_db, 数据表tb_student;

     目的端：

     2.在源表中新增数据

     3.在目的库中查看

    4.查看console

    总结： 可以看到新增数据已经同步过来了！ 

    源代码见:   https://gitee.com/jurf-liu/canal-demo.git

  • 相关阅读:
    Python的文件操作
    从 Google 离职，前Go 语言负责人跳槽小公司
    深度学习发展下的“摩尔困境”，人工智能又将如何破局？
    java计算机毕业设计ssm基于C程序课程的题库在线平台(源码+系统+mysql数据库+Lw文档）
    数据结构第二课 -----线性表之顺序表
    CSDN每日一练 |『坐公交』『盗版解锁密码』『n边形划分』2023-09-17
    神经网络有哪些基本功能,常见的神经网络有哪些
    世界坐标系、相机坐标系和图像坐标系的转换
    难点解释-理解寄主机通过虚拟网络连接到虚拟机的概念
    并查集学习笔记
  • 原文地址：https://blog.csdn.net/u011066470/article/details/126734578