canal rdb适配器

女爷i 2022-10-13 01:57 236阅读 0赞

canal rdb适配器

官网：https://github.com/alibaba/canal/wiki/Sync-RDB

应用：**实现数据库之间的增量同步、数据备份 **

*********************

配置文件

application.yml：启动器配置文件

  1. server:
  2.   port: 8081
  3. spring:
  4.   jackson:
  5.     date-format: yyyy-MM-dd HH:mm:ss
  6.     time-zone: GMT+8
  7.     default-property-inclusion: non_null
  9. canal.conf:
  10.   mode: tcp #tcp kafka rocketMQ rabbitMQ
  11.   flatMessage: true
  12.   zookeeperHosts:
  13.   syncBatchSize: 1000
  14.   retries: 0
  15.   timeout:
  16.   accessKey:
  17.   secretKey:
  18.   consumerProperties:
  19.     # canal tcp consumer
  20.     canal.tcp.server.host: 127.0.0.1:11111
  21.     canal.tcp.zookeeper.hosts:
  22.     canal.tcp.batch.size: 500
  23.     canal.tcp.username:
  24.     canal.tcp.password:
  25.     # kafka consumer
  26.     kafka.bootstrap.servers: 127.0.0.1:9092
  27.     kafka.enable.auto.commit: false
  28.     kafka.auto.commit.interval.ms: 1000
  29.     kafka.auto.offset.reset: latest
  30.     kafka.request.timeout.ms: 40000
  31.     kafka.session.timeout.ms: 30000
  32.     kafka.isolation.level: read_committed
  33.     kafka.max.poll.records: 1000
  34.     # rocketMQ consumer
  35.     rocketmq.namespace:
  36.     rocketmq.namesrv.addr: 127.0.0.1:9876
  37.     rocketmq.batch.size: 1000
  38.     rocketmq.enable.message.trace: false
  39.     rocketmq.customized.trace.topic:
  40.     rocketmq.access.channel:
  41.     rocketmq.subscribe.filter:
  42.     # rabbitMQ consumer
  43.     rabbitmq.host:
  44.     rabbitmq.virtual.host:
  45.     rabbitmq.username:
  46.     rabbitmq.password:
  47.     rabbitmq.resource.ownerId:
  49.   srcDataSources:
  50.     defaultDS:
  51.       url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
  52.       username: root
  53.       password: 121212
  54.   canalAdapters:
  55.   - instance: example # canal instance Name or mq topic name
  56.     groups:
  57.     - groupId: g1
  58.       outerAdapters:
  59.       - name: rdb
  60.         key: mysql1
  61.         properties:
  62.           jdbc.driverClassName: com.mysql.jdbc.Driver
  63.           jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
  64.           jdbc.username: root
  65.           jdbc.password: 121212
  66. #      - name: rdb
  67. #        key: oracle1
  68. #        properties:
  69. #          jdbc.driverClassName: oracle.jdbc.OracleDriver
  70. #          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
  71. #          jdbc.username: mytest
  72. #          jdbc.password: m121212
  73. #      - name: rdb
  74. #        key: postgres1
  75. #        properties:
  76. #          jdbc.driverClassName: org.postgresql.Driver
  77. #          jdbc.url: jdbc:postgresql://localhost:5432/postgres
  78. #          jdbc.username: postgres
  79. #          jdbc.password: 121212
  80. #          threads: 1
  81. #          commitSize: 3000

mytest_user.yml：canal adapter会自动加载conf/rdb下的所有以.yml结尾的适配器配置文件

  1. dataSourceKey: defaultDS
  2. destination: example
  3. groupId: g1
  4. outerAdapterKey: mysql1
  5. concurrent: true
  6. dbMapping:
  7.   database: mytest            # 源库
  8.   table: user                 # 源表
  9.   targetTable: mytest2.user   # 目标数据库、表
  10.   targetPk:
  11.     id: id
  12.   #mapAll: true       # 是否整表映射, 要求源表和目标表字段名一模一样 
  13.                       # 如果targetColumns也配置了映射,则以targetColumns配置为准
  14.   targetColumns:      # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填
  15.     id:
  16.     name:
  17.     role_id:
  18.     c_time:
  19.     test1:
  20.   etlCondition: "where c_time>={}"
  21.   commitBatch: 3000 # 批量提交的大小
  24. ## Mirror schema synchronize config
  25. #dataSourceKey: defaultDS
  26. #destination: example
  27. #groupId: g1
  28. #outerAdapterKey: mysql1
  29. #concurrent: true
  30. #dbMapping:
  31. #  mirrorDb: true              #镜像配置，需要源库、目标库一模一样
  32. #  database: mytest

*********************

适配器配置类

MappingConfig：映射配置类

  1. public class MappingConfig implements AdapterConfig {
  3.     private String    dataSourceKey;      // 数据源key
  5.     private String    destination;        // canal实例或MQ的topic
  7.     private String    groupId;            // groupId
  9.     private String    outerAdapterKey;    // 对应适配器的key
  11.     private boolean   concurrent = false; // 是否并行同步
  13.     private DbMapping dbMapping;          // db映射配置
  16.     。。。
  18.     public void validate() {
  19.         if (dbMapping.database == null || dbMapping.database.isEmpty()) {
  20.             throw new NullPointerException("dbMapping.database");
  21.         }   //database不能为空，否则抛空指针异常
  23.         if (!dbMapping.getMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) {
  24.             throw new NullPointerException("dbMapping.table");
  25.         }   //如果没有开启镜像（dbMapping.mirrorDb=false），则table不能为空，否则抛空指针异常
  27.         if (!dbMapping.getMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) {
  28.             throw new NullPointerException("dbMapping.targetTable");
  29.         }   //如果没有开启镜像（dbMapping.mirrorDb=false），则targetTable不能为空，否则抛空指针异常
  30.     }
  33. ***********
  34. 内部类：DbMapping
  36.     public static class DbMapping implements AdapterMapping {
  38.         private boolean             mirrorDb        = false;                 // 是否镜像库
  39.         private String              database;                                // 数据库名或schema名
  40.         private String              table;                                   // 表名
  41.         private Map<String, String> targetPk        = new LinkedHashMap<>(); // 目标表主键字段
  42.         private boolean             mapAll          = false;                 // 映射所有字段
  43.         private String              targetDb;                                // 目标库名
  44.         private String              targetTable;                             // 目标表名
  45.         private Map<String, String> targetColumns;                           // 目标表字段映射
  47.         private boolean             caseInsensitive = false;                 // 目标表不区分大小写，默认是否
  49.         private String              etlCondition;                            // etl条件sql
  51.         private int                 readBatch       = 5000;
  52.         private int                 commitBatch     = 5000;                  // etl等批量提交大小
  54.         private Map<String, String> allMapColumns;

*********************

示例

创建 mysql实例：源数据库

  1. docker run -it -d --net fixed --ip 172.18.0.2 -p 3306:3306 --privileged=true \
  2. --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
  5. # 创建用户、授权
  6. mysql> create user canal identified with mysql_native_password by "123456";
  7. Query OK, 0 rows affected (0.02 sec)
  9. mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  10. Query OK, 0 rows affected (0.00 sec)
  12. mysql> flush privileges;
  13. Query OK, 0 rows affected (0.00 sec)

创建 canal实例

  1. docker run -it -d --net fixed --ip 172.18.0.3 \
  2. -p 11111:11111 --name canal-server \
  3. -e canal.instance.master.address=172.18.0.2:3306  \
  4. -e canal.instance.dbUsername=canal \
  5. -e canal.instance.dbPassword=123456 canal/canal-server

创建 mysql实例：目标数据库

  1. docker run -it -d --net fixed --ip 172.18.0.5 -p 3307:3306 --privileged=true \
  2. --name mysql2 -e  MYSQL_ROOT_PASSWORD=123456 mysql
  5. # 创建数据库
  6. mysql> create database lihu;
  7. Query OK, 1 row affected (0.00 sec)
  9. mysql> use lihu;
  10. Database changed
  11. mysql> create table user(id int not null auto_increment primary key,custom_name varchar(20) not null, custom_age int not null);
  12. Query OK, 0 rows affected (0.02 sec)

创建 canal adapter

  1. docker run -it -d --net fixed --ip 172.18.0.4 -p 8081:8081 \
  2. -v /usr/canal/canal-adapter/rdb/application.yml:/opt/canal-adapter/conf/application.yml \
  3. -v /usr/canal/canal-adapter/rdb/adapter:/opt/canal-adapter/conf/rdb \
  4. --name canal-adapter slpcat/canal-adapter:v1.1.5-jdk8
  7. *****************
  8. application.yml
  10. server:
  11.   port: 8081
  12. spring:
  13.   jackson:
  14.     date-format: yyyy-MM-dd HH:mm:ss
  15.     time-zone: GMT+8
  16.     default-property-inclusion: non_null
  18. canal.conf:
  19.   mode: tcp #tcp kafka rocketMQ rabbitMQ
  20.   flatMessage: true
  21.   zookeeperHosts:
  22.   syncBatchSize: 1000
  23.   retries: 0
  24.   timeout:
  25.   accessKey:
  26.   secretKey:
  27.   consumerProperties:
  28.     # canal tcp consumer
  29.     canal.tcp.server.host: 172.18.0.3:11111
  30.     canal.tcp.zookeeper.hosts:
  31.     canal.tcp.batch.size: 500
  32.     canal.tcp.username:
  33.     canal.tcp.password:
  34.     # kafka consumer
  35.     kafka.bootstrap.servers: 127.0.0.1:9092
  36.     kafka.enable.auto.commit: false
  37.     kafka.auto.commit.interval.ms: 1000
  38.     kafka.auto.offset.reset: latest
  39.     kafka.request.timeout.ms: 40000
  40.     kafka.session.timeout.ms: 30000
  41.     kafka.isolation.level: read_committed
  42.     kafka.max.poll.records: 1000
  43.     # rocketMQ consumer
  44.     rocketmq.namespace:
  45.     rocketmq.namesrv.addr: 127.0.0.1:9876
  46.     rocketmq.batch.size: 1000
  47.     rocketmq.enable.message.trace: false
  48.     rocketmq.customized.trace.topic:
  49.     rocketmq.access.channel:
  50.     rocketmq.subscribe.filter:
  51.     # rabbitMQ consumer
  52.     rabbitmq.host:
  53.     rabbitmq.virtual.host:
  54.     rabbitmq.username:
  55.     rabbitmq.password:
  56.     rabbitmq.resource.ownerId:
  59.   srcDataSources:
  60.     defaultDS:
  61.       url: jdbc:mysql://172.18.0.2:3306/lihu?useUnicode=true
  62.       username: root
  63.       password: 123456
  64.   canalAdapters:
  65.   - instance: example # canal instance Name or mq topic name
  66.     groups:
  67.     - groupId: g1
  68.       outerAdapters:
  69. #      - name: logger
  70.       - name: rdb
  71.         key: mysql1
  72.         properties:
  73.           jdbc.driverClassName: com.mysql.jdbc.Driver
  74.           jdbc.url: jdbc:mysql://172.18.0.5:3306/lihu?useUnicode=true
  75.           jdbc.username: root
  76.           jdbc.password: 123456
  79. *****************
  80. mytest_user.yml：适配器配置文件
  82. dataSourceKey: defaultDS
  83. destination: example
  84. groupId: g1
  85. outerAdapterKey: mysql1
  86. concurrent: true
  87. dbMapping:
  88.   database: lihu
  89.   table: user
  90.   targetTable: lihu.user
  91.   targetPk:
  92.     id: id
  93. #  mapAll: true
  94.   targetColumns:
  95.     id: id
  96.     custom_name: name
  97.     custom_age: age
  98. #  etlCondition: "where c_time>={}"
  99.   commitBatch: 3000 # 批量提交的大小

******************

使用测试

mysql：源数据库插入数据

  1. mysql> insert into user(id,name,age) values(3,'lihu',20),(4,'lihu2',21);
  2. Query OK, 2 rows affected (0.10 sec)
  3. Records: 2  Duplicates: 0  Warnings: 0

canal-adapter 日志

  1. # rdb创建成功
  2. 2021-07-07 23:19:42.700 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: rdb succeed
  4. 2021-07-07 23:19:42.770 [main] INFO  c.alibaba.otter.canal.connector.core.spi.ExtensionLoader - extension classpath dir: /opt/canal-adapter/plugin
  5. 2021-07-07 23:19:42.886 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Start adapter for canal-client mq topic: example-g1 succeed
  6. 2021-07-07 23:19:42.886 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## the canal client adapters are running now ......
  7. 2021-07-07 23:19:42.899 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <=============
  8. 2021-07-07 23:19:42.925 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8081"]
  9. 2021-07-07 23:19:43.036 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
  10. 2021-07-07 23:19:43.201 [main] INFO  o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8081 (http) with context path ''
  11. 2021-07-07 23:19:43.235 [main] INFO  c.a.otter.canal.adapter.launcher.CanalAdapterApplication - Started CanalAdapterApplication in 21.316 seconds (JVM running for 23.705)
  12. 2021-07-07 23:19:43.332 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Subscribe destination: example succeed <=============
  13. Wed Jul 07 23:20:13 GMT+08:00 2021 WARN: Establishing SSL connection without server's identity verification is not recommended. According to MySQL 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be established by default if explicit option isn't set. For compliance with existing applications not using SSL the verifyServerCertificate property is set to 'false'. You need either to explicitly disable SSL by setting useSSL=false, or set useSSL=true and provide truststore for server certificate verification.
  15. # 插入的行记录：{"id":3,"name":"lihu","age":20}
  16. 2021-07-07 23:20:14.284 [pool-1-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":3,"name":"lihu","age":20},"database":"lihu","destination":"example","old":null,"table":"user","type":"INSERT"}
  18. # 插入的行记录：{"id":4,"name":"lihu2","age":21}
  19. 2021-07-07 23:20:14.285 [pool-2-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":4,"name":"lihu2","age":21},"database":"lihu","destination":"example","old":null,"table":"user","type":"INSERT"}

目标数据库 mysql2

  1. mysql> select * from user;
  2. +----+-------------+------------+
  3. | id | custom_name | custom_age |
  4. +----+-------------+------------+
  5. |  3 | lihu        |         20 |
  6. |  4 | lihu2       |         21 |
  7. +----+-------------+------------+
  8. 2 rows in set (0.00 sec)

源数据库的新增数据同步成功

发表评论

表情：
评论列表 （有 0 条评论，236人围观）

还没有评论，来说两句吧...

相关阅读

    相关 canal简介

    1.简介 canal是纯Java开发的，基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了mysql。 原理相对比较简单： 1. 1.      canal

    妖狐艹你老母妖狐艹你老母/ 0 赞/ 347 阅读