devbox
小舞很执着
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

canal 同步mysql数据_使用canal同步mysql

作者：小舞很执着 | 2024-08-18 11:12:14

赞

踩

使用canal同步mysql

1、Canal 是一个开源的数据库数据同步工具，用于监控数据库变更并将变更数据传递到指定的消费端，通常用于将MySQL数据库的变更数据传递到其他系统进行处理

2、使用 Canal 同步 MySQL 数据的基本步骤：

  1. 安装和配置 Canal 服务器：

    • 下载 Canal 服务器安装包或从源代码构建。
    • 配置 Canal 服务器的参数，包括数据库连接信息、监听的数据表、认证配置等。这些配置通常在 canal.properties 文件中设置。

  2. 启动 Canal 服务器：

    • 启动已配置好的 Canal 服务器，它会连接到 MySQL 数据库，监控数据变更。

  3. 编写消费端应用程序：

    • 消费端应用程序可以是您自己编写的任何程序，它会从 Canal 服务器获取数据库变更数据并进行处理。
    • 使用 Canal 提供的客户端 SDK（如 Java 客户端）来连接 Canal 服务器，订阅特定的数据库和数据表。
    • 编写逻辑来处理接收到的变更数据，可以根据业务需求将数据写入其他系统、存储、消息队列等。
  4. 客户端连接canal-server，订阅mysql数据变更逻辑，以及将binglog日志解析成相应的sql语句，可以直接在mysql中执行
  5. 引入的pom 
    • <dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.4</version>
</dependency>
    1. package cn.liu.demo.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.  
    9. import java.net.InetSocketAddress;
    10. import java.util.List;
    11. import java.util.Queue;
    12. import java.util.concurrent.ConcurrentLinkedQueue;
    13.  
    14. import static com.alibaba.otter.canal.protocol.CanalEntry.*;
    15. import static com.alibaba.otter.canal.protocol.CanalEntry.EntryType.*;
    16.  
    17. /**
    18.  * @Author：
    19.  * @Date：2023/8/22 14:42
    20.  * @Description：
    21.  */
    22. public class CanalClient {
    23.     private static Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();
    24.  
    25.     public static void main(String[] args) {
    26.         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "instance1", "admin", "admin");
    27.         try {
    28.             connector.connect();
    29.             // 监听的表，格式为数据库.表名,数据库.表名
    30.             connector.subscribe("user.*");
    31.             //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
    32.             connector.rollback();
    33.             while (true){
    34.                 Message message = connector.getWithoutAck(100); // 获取指定数量的数据
    35.                 //获取批量ID
    36.                 long batchId = message.getId();
    37.                 //获取批量的数量
    38.                 int size = message.getEntries().size();
    39.  
    40.                 //如果没有数据
    41.                 if (batchId == -1 || size == 0) {
    42.                     try {
    43.                         //线程休眠2秒
    44.                         Thread.sleep(2000);
    45.                     } catch (InterruptedException e) {
    46.                         e.printStackTrace();
    47.                     }
    48.                 } else {
    49.                     //如果有数据,处理数据
    50.                     printEntry(message.getEntries());
    51.                     dataHandle(message.getEntries());
    52.                 }
    53.                 //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
    54.                 connector.ack(batchId);
    55.  
    56.  
    57.                 //当队列里面堆积的sql大于一定数值的时候就模拟执行
    58.                 if (SQL_QUEUE.size() >= 1) {
    59.                     executeQueueSql();
    60.                 }
    61.             }
    62.  
    63.         }catch (Exception e){
    64.             e.printStackTrace();
    65.         }finally {
    66.             connector.disconnect();
    67.         }
    68.  
    69.     }
    70.  
    71.     /**
    72.      * 模拟执行队列里面的sql语句
    73.      */
    74.     public static 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.     }
    81.  
    82.  
    83.     /**
    84.      * 打印canal server解析binlog获得的实体类信息
    85.      */
    86.     private static void printEntry(List<Entry> entrys) {
    87.         for (Entry entry : entrys) {
    88.             if (entry.getEntryType() == TRANSACTIONBEGIN || entry.getEntryType() == TRANSACTIONEND) {
    89.                 //开启/关闭事务的实体类型，跳过
    90.                 continue;
    91.             }
    92.             //RowChange对象，包含了一行数据变化的所有特征
    93.             //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
    94.             CanalEntry.RowChange rowChage;
    95.             try {
    96.                 rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
    97.             } catch (Exception e) {
    98.                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
    99.             }
    100.             //获取操作类型：insert/update/delete类型
    101.             CanalEntry.EventType eventType = rowChage.getEventType();
    102.             //打印Header信息
    103.             System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
    104.                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    105.                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
    106.                     eventType));
    107.             //判断是否是DDL语句
    108.             if (rowChage.getIsDdl()) {
    109.                 System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
    110.             }
    111.             //获取RowChange对象里的每一行数据，打印出来
    112.             for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
    113.                 //如果是删除语句
    114.                 if (eventType == CanalEntry.EventType.DELETE) {
    115.                     printColumn(rowData.getBeforeColumnsList());
    116.                     //如果是新增语句
    117.                 } else if (eventType == CanalEntry.EventType.INSERT) {
    118.                     printColumn(rowData.getAfterColumnsList());
    119.                     //如果是更新的语句
    120.                 } else {
    121.                     //变更前的数据
    122.                     System.out.println("------->; before");
    123.                     printColumn(rowData.getBeforeColumnsList());
    124.                     //变更后的数据
    125.                     System.out.println("------->; after");
    126.                     printColumn(rowData.getAfterColumnsList());
    127.                 }
    128.             }
    129.         }
    130.     }
    131.  
    132.     /**
    133.      * 数据处理
    134.      *
    135.      * @param entrys
    136.      */
    137.     private static void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException {
    138.         for (Entry entry : entrys) {
    139.             if (ROWDATA == entry.getEntryType()) {
    140.                 CanalEntry.RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    141.                 EventType eventType = rowChange.getEventType();
    142.                 if (eventType == EventType.DELETE) {
    143.                     saveDeleteSql(entry);
    144.                 } else if (eventType == EventType.UPDATE) {
    145.                     saveUpdateSql(entry);
    146.                 } else if (eventType == EventType.INSERT) {
    147.                     saveInsertSql(entry);
    148.                 }
    149.             }
    150.         }
    151.     }
    152.  
    153.     private static void saveUpdateSql(Entry entry) {
    154.         try {
    155.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    156.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    157.             for (RowData rowData : rowDatasList) {
    158.                 List<Column> newColumnList = rowData.getAfterColumnsList();
    159.                 StringBuffer sql = new StringBuffer("update " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " set ");
    160.                 for (int i = 0; i < newColumnList.size(); i++) {
    161.                     sql.append(" " + newColumnList.get(i).getName()
    162.                             + " = '" + newColumnList.get(i).getValue() + "'");
    163.                     if (i != newColumnList.size() - 1) {
    164.                         sql.append(",");
    165.                     }
    166.                 }
    167.                 sql.append(" where ");
    168.                 List<Column> oldColumnList = rowData.getBeforeColumnsList();
    169.                 for (Column column : oldColumnList) {
    170.                     if (column.getIsKey()) {
    171.                         //暂时只支持单一主键
    172.                         sql.append(column.getName() + "=" + column.getValue());
    173.                         break;
    174.                     }
    175.                 }
    176.                 SQL_QUEUE.add(sql.toString());
    177.             }
    178.         } catch (InvalidProtocolBufferException e) {
    179.             e.printStackTrace();
    180.         }
    181.     }
    182.  
    183.     /**
    184.      * 保存删除语句
    185.      *
    186.      * @param entry
    187.      */
    188.     private static void saveDeleteSql(Entry entry) {
    189.         try {
    190.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    191.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    192.             for (RowData rowData : rowDatasList) {
    193.                 List<Column> columnList = rowData.getBeforeColumnsList();
    194.                 StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " where ");
    195.                 for (Column column : columnList) {
    196.                     if (column.getIsKey()) {
    197.                         //暂时只支持单一主键
    198.                         sql.append(column.getName() + "=" + column.getValue());
    199.                         break;
    200.                     }
    201.                 }
    202.                 SQL_QUEUE.add(sql.toString());
    203.             }
    204.         } catch (InvalidProtocolBufferException e) {
    205.             e.printStackTrace();
    206.         }
    207.     }
    208.  
    209.     /**
    210.      * 保存插入语句
    211.      *
    212.      * @param entry
    213.      */
    214.     private static void saveInsertSql(Entry entry) {
    215.         try {
    216.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
    217.             List<RowData> rowDatasList = rowChange.getRowDatasList();
    218.             for (RowData rowData : rowDatasList) {
    219.                 List<Column> columnList = rowData.getAfterColumnsList();
    220.                 StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " (");
    221.                 for (int i = 0; i < columnList.size(); i++) {
    222.                     sql.append(columnList.get(i).getName());
    223.                     if (i != columnList.size() - 1) {
    224.                         sql.append(",");
    225.                     }
    226.                 }
    227.                 sql.append(") VALUES (");
    228.                 for (int i = 0; i < columnList.size(); i++) {
    229.                     sql.append("'" + columnList.get(i).getValue() + "'");
    230.                     if (i != columnList.size() - 1) {
    231.                         sql.append(",");
    232.                     }
    233.                 }
    234.                 sql.append(")");
    235.                 SQL_QUEUE.add(sql.toString());
    236.             }
    237.         } catch (InvalidProtocolBufferException e) {
    238.             e.printStackTrace();
    239.         }
    240.     }
    241.  
    242.  
    243.     private static void printColumn(List<CanalEntry.Column> columns) {
    244.         for (CanalEntry.Column column : columns) {
    245.             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    246.         }
    247.     }
    248.  
    249.  
    250.  
    251. }

 总结：

总结起来，使用 Canal 同步 MySQL 数据的过程涉及到配置 Canal 服务器、编写消费端应用程序以及处理接收到的变更数据。这使您能够将 MySQL 数据库的变更同步到其他系统中，实现数据的实时传输和处理。具体的实现方式会根据您的业务需求和技术栈而有所不同

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号