canal mq数据同步

刺骨的言语ヽ痛彻心扉 2022-10-19 12:48 254阅读 0赞

canal mq数据同步

官网：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

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

简 介

tcp mode：客户端直连canal server

  1.          ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

mq mode：canal server将binlog数据发送到mq，1.1.5支持kafka、rocketmq、rabbitmq

  1.          ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1][]

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

canal server mq配置

canal.properties

  1. ##################################################
  2. #########           MQ Properties      #############
  3. ##################################################
  4. # aliyun ak/sk , support rds/mq
  5. canal.aliyun.accessKey =
  6. canal.aliyun.secretKey =
  7. canal.aliyun.uid=
  9. canal.mq.flatMessage = true
  10. canal.mq.canalBatchSize = 50
  11. canal.mq.canalGetTimeout = 100
  12. # Set this value to "cloud", if you want open message trace feature in aliyun.
  13. canal.mq.accessChannel = local
  15. canal.mq.database.hash = true
  16. canal.mq.send.thread.size = 30
  17. canal.mq.build.thread.size = 8
  19. ##################################################
  20. #########              Kafka              #############
  21. ##################################################
  22. kafka.bootstrap.servers = 127.0.0.1:9092
  23. kafka.acks = all
  24. kafka.compression.type = none
  25. kafka.batch.size = 16384
  26. kafka.linger.ms = 1
  27. kafka.max.request.size = 1048576
  28. kafka.buffer.memory = 33554432
  29. kafka.max.in.flight.requests.per.connection = 1
  30. kafka.retries = 0
  32. kafka.kerberos.enable = false
  33. kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
  34. kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
  36. ##################################################
  37. #########             RocketMQ         #############
  38. ##################################################
  39. rocketmq.producer.group = test
  40. rocketmq.enable.message.trace = false
  41. rocketmq.customized.trace.topic =
  42. rocketmq.namespace =
  43. rocketmq.namesrv.addr = 127.0.0.1:9876
  44. rocketmq.retry.times.when.send.failed = 0
  45. rocketmq.vip.channel.enabled = false
  46. rocketmq.tag = 
  48. ##################################################
  49. #########             RabbitMQ         #############
  50. ##################################################
  51. rabbitmq.host =
  52. rabbitmq.virtual.host =
  53. rabbitmq.exchange =
  54. rabbitmq.username =
  55. rabbitmq.password =
  56. rabbitmq.deliveryMode =

example/instance.properties

  1. # 静态topic：消息发送的分区为example
  2. canal.mq.topic=example
  4. # 动态topic：根据数据库、表动态设置发送的topic
  5. canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
  8. # 静态分区：数据发送的分区
  9. canal.mq.partition=0
  11. # 动态分区：根据数据库、表设置返送的分区
  12. canal.mq.partitionsNum=3
  13. canal.mq.partitionHash=test.table:id^name,.*\\..*
  15. # 为不同的topic动态设置分区数
  16. canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6  #test.*：test开头的topic分区数为4
  17.                                                       #mycanal:6 topic mycanal分区数为6
  20. # 动态topic：canal.mq.dynamicTopic
  21. test：test库中的所有数据都发送到test topic上
  22. .*：数据库的数据都发到对应数据库名的topic上
  24. mytest1.user：发送到mytest1_user topic上
  25. mytest2\\..*：发送到mytest2_tableName topic上
  27. topicName:.*：所有数据库的数据都发到topicName上
  28. topicName:test\\..*：test下的所有表都发送到topicName上
  30. test,test1\\.*：数据库test1中的表发送到test1_tableName topic上
  31.                 数据库test中的所有表发送到test topic上
  32.                 其余所有数据发送到canal.mq.topic指定的topic上
  35. # 动态分区：
  36. .*\\..*:id：hash字段为id
  37. .*\\..*:id^name：hash字段为id、name
  38. .*\\..*:$pk$：hash字段为主键（自动查找）
  39. .*\\..*：根据tableName hash
  40. partitionHash为空：发送到默认分区 0
  41. test.test,test.test2:id：test.test根据表名test hash
  42.                          test.test2根据id hash
  43.                          其余发送到对应topic的0分区

CanalKafkaProducer：**kafka消息发送类**

  1. @SuppressWarnings({ "rawtypes", "unchecked" })
  2. @SPI("kafka")
  3. public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQProducer {
  5.     private static final Logger      logger              = LoggerFactory.getLogger(CanalKafkaProducer.class);
  7.     private static final String      PREFIX_KAFKA_CONFIG = "kafka.";
  9.     private Producer<String, byte[]> producer;
  11.     @Override
  12.     public void init(Properties properties) {
  13.         KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig();
  14.         this.mqProperties = kafkaProducerConfig;
  15.         super.init(properties);
  16.         // load properties
  17.         this.loadKafkaProperties(properties);
  19.         Properties kafkaProperties = new Properties();
  20.         kafkaProperties.putAll(kafkaProducerConfig.getKafkaProperties());
  21.         kafkaProperties.put("max.in.flight.requests.per.connection", 1);
  22.         kafkaProperties.put("key.serializer", StringSerializer.class);
  23.         if (kafkaProducerConfig.isKerberosEnabled()) {
  24.             File krb5File = new File(kafkaProducerConfig.getKrb5File());
  25.             File jaasFile = new File(kafkaProducerConfig.getJaasFile());
  26.             if (krb5File.exists() && jaasFile.exists()) {
  27.                 // 配置kerberos认证，需要使用绝对路径
  28.                 System.setProperty("java.security.krb5.conf", krb5File.getAbsolutePath());
  29.                 System.setProperty("java.security.auth.login.config", jaasFile.getAbsolutePath());
  30.                 System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
  31.                 kafkaProperties.put("security.protocol", "SASL_PLAINTEXT");
  32.                 kafkaProperties.put("sasl.kerberos.service.name", "kafka");
  33.             } else {
  34.                 String errorMsg = "ERROR # The kafka kerberos configuration file does not exist! please check it";
  35.                 logger.error(errorMsg);
  36.                 throw new RuntimeException(errorMsg);
  37.             }
  38.         }
  39.         kafkaProperties.put("value.serializer", KafkaMessageSerializer.class);
  40.         producer = new KafkaProducer<>(kafkaProperties);
  41.     }
  43.     private void loadKafkaProperties(Properties properties) {
  44.                                      //加载相关配置
  45.         KafkaProducerConfig kafkaProducerConfig = (KafkaProducerConfig) this.mqProperties;
  46.         Map<String, Object> kafkaProperties = kafkaProducerConfig.getKafkaProperties();
  47.         // 兼容下<=1.1.4的mq配置
  48.         doMoreCompatibleConvert("canal.mq.servers", "kafka.bootstrap.servers", properties);
  49.         doMoreCompatibleConvert("canal.mq.acks", "kafka.acks", properties);
  50.         doMoreCompatibleConvert("canal.mq.compressionType", "kafka.compression.type", properties);
  51.         doMoreCompatibleConvert("canal.mq.retries", "kafka.retries", properties);
  52.         doMoreCompatibleConvert("canal.mq.batchSize", "kafka.batch.size", properties);
  53.         doMoreCompatibleConvert("canal.mq.lingerMs", "kafka.linger.ms", properties);
  54.         doMoreCompatibleConvert("canal.mq.maxRequestSize", "kafka.max.request.size", properties);
  55.         doMoreCompatibleConvert("canal.mq.bufferMemory", "kafka.buffer.memory", properties);
  56.         doMoreCompatibleConvert("canal.mq.kafka.kerberos.enable", "kafka.kerberos.enable", properties);
  57.         doMoreCompatibleConvert("canal.mq.kafka.kerberos.krb5.file", "kafka.kerberos.krb5.file", properties);
  58.         doMoreCompatibleConvert("canal.mq.kafka.kerberos.jaas.file", "kafka.kerberos.jaas.file", properties);

RocketMQConstants：rocketmq producer属性配置

  1. public class RocketMQConstants {
  3.     public static final String ROOT                                  = "rocketmq";
  5.     public static final String ROCKETMQ_PRODUCER_GROUP               = ROOT + "." + "producer.group";
  6.     public static final String ROCKETMQ_ENABLE_MESSAGE_TRACE         = ROOT + "." + "enable.message.trace";
  7.     public static final String ROCKETMQ_CUSTOMIZED_TRACE_TOPIC       = ROOT + "." + "customized.trace.topic";
  8.     public static final String ROCKETMQ_NAMESPACE                    = ROOT + "." + "namespace";
  9.     public static final String ROCKETMQ_NAMESRV_ADDR                 = ROOT + "." + "namesrv.addr";
  10.     public static final String ROCKETMQ_RETRY_TIMES_WHEN_SEND_FAILED = ROOT + "." + "retry.times.when.send.failed";
  11.     public static final String ROCKETMQ_VIP_CHANNEL_ENABLED          = ROOT + "." + "vip.channel.enabled";
  12.     public static final String ROCKETMQ_TAG                          = ROOT + "." + "tag";
  14.     public static final String ROCKETMQ_ACCESS_CHANNEL               = ROOT + "." + "access.channel";
  15.     public static final String ROCKETMQ_BATCH_SIZE                   = ROOT + "." + "batch.size";
  16.     public static final String ROCKETMQ_SUBSCRIBE_FILTER             = ROOT + "." + "subscribe.filter";
  18. }

RabbitMQConstants：rabbitmq producer属性配置

  1. public class RabbitMQConstants {
  3.     public static final String ROOT                      = "rabbitmq";
  5.     public static final String RABBITMQ_HOST             = ROOT + "." + "host";
  6.     public static final String RABBITMQ_EXCHANGE         = ROOT + "." + "exchange";
  7.     public static final String RABBITMQ_VIRTUAL_HOST     = ROOT + "." + "virtual.host";
  8.     public static final String RABBITMQ_USERNAME         = ROOT + "." + "username";
  9.     public static final String RABBITMQ_PASSWORD         = ROOT + "." + "password";
  11.     public static final String RABBITMQ_RESOURCE_OWNERID = ROOT + "." + "rabbitmq.resource.ownerId";
  12. }

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

canal adapter mq属性配置

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: logger

CanalKafkaConsumer：kafka. 开头的属性均可读取

  1. @SPI("kafka")
  2. public class CanalKafkaConsumer implements CanalMsgConsumer {
  4.     private static final String      PREFIX_KAFKA_CONFIG = "kafka.";
  6.     private KafkaConsumer<String, ?> kafkaConsumer;
  7.     private boolean                  flatMessage         = true;
  8.     private String                   topic;
  10.     private Map<Integer, Long>       currentOffsets      = new ConcurrentHashMap<>();
  11.     private Properties               kafkaProperties     = new Properties();
  13.     @Override
  14.     public void init(Properties properties, String topic, String groupId) {
  15.         this.topic = topic;
  17.         Boolean flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
  18.         if (flatMessage != null) {
  19.             this.flatMessage = flatMessage;
  20.         }
  21.         for (Map.Entry<Object, Object> entry : properties.entrySet()) {
  22.             String k = (String) entry.getKey();
  23.             Object v = entry.getValue();
  24.             if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) {
  25.                 kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v);
  26.             }   //读取所有 kafka. 开头的属性
  27.         }
  28.         kafkaProperties.put("group.id", groupId);
  29.         kafkaProperties.put("key.deserializer", StringDeserializer.class);
  30.         kafkaProperties.put("client.id", UUID.randomUUID().toString().substring(0, 6));
  31.     }
  33.     @Override
  34.     public void connect() {
  35.         if (this.flatMessage) {
  36.             kafkaProperties.put("value.deserializer", StringDeserializer.class);
  37.             this.kafkaConsumer = new KafkaConsumer<String, String>(kafkaProperties);
  38.         } else {
  39.             kafkaProperties.put("value.deserializer", KafkaMessageDeserializer.class);
  40.             this.kafkaConsumer = new KafkaConsumer<String, Message>(kafkaProperties);
  41.         }
  42.         kafkaConsumer.subscribe(Collections.singletonList(topic));
  43.     }

CanalRocketMQConsumer：rocketmq 消费相关属性

  1. @SPI("rocketmq")
  2. public class CanalRocketMQConsumer implements CanalMsgConsumer {
  4.     private static final Logger                                logger               = LoggerFactory.getLogger(CanalRocketMQConsumer.class);
  5.     private static final String                                CLOUD_ACCESS_CHANNEL = "cloud";
  7.     private String                                             nameServer;
  8.     private String                                             topic;
  9.     private String                                             groupName;
  10.     private DefaultMQPushConsumer                              rocketMQConsumer;
  11.     private BlockingQueue<ConsumerBatchMessage<CommonMessage>> messageBlockingQueue;
  12.     private int                                                batchSize            = -1;
  13.     private long                                               batchProcessTimeout  = 60 * 1000;
  14.     private boolean                                            flatMessage;
  15.     private volatile ConsumerBatchMessage<CommonMessage>       lastGetBatchMessage  = null;
  16.     private String                                             accessKey;
  17.     private String                                             secretKey;
  18.     private String                                             customizedTraceTopic;
  19.     private boolean                                            enableMessageTrace   = false;
  20.     private String                                             accessChannel;
  21.     private String                                             namespace;
  22.     private String                                             filter               = "*";
  24.     @Override
  25.     public void init(Properties properties, String topic, String groupName) {
  26.         this.topic = topic;
  27.         this.groupName = groupName;
  28.         this.flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
  29.         this.messageBlockingQueue = new LinkedBlockingQueue<>(1024);
  30.         this.accessKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_ACCESS_KEY);
  31.         this.secretKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_SECRET_KEY);
  32.         String enableMessageTrace = properties.getProperty(RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
  33.         if (StringUtils.isNotEmpty(enableMessageTrace)) {
  34.             this.enableMessageTrace = Boolean.parseBoolean(enableMessageTrace);
  35.         }
  36.         this.customizedTraceTopic = properties.getProperty(RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
  37.         this.accessChannel = properties.getProperty(RocketMQConstants.ROCKETMQ_ACCESS_CHANNEL);
  38.         this.namespace = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESPACE);
  39.         this.nameServer = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESRV_ADDR);
  40.         String batchSize = properties.getProperty(RocketMQConstants.ROCKETMQ_BATCH_SIZE);
  41.         if (StringUtils.isNotEmpty(batchSize)) {
  42.             this.batchSize = Integer.parseInt(batchSize);
  43.         }
  44.         String subscribeFilter = properties.getProperty(RocketMQConstants.ROCKETMQ_SUBSCRIBE_FILTER);
  45.         if (StringUtils.isNotEmpty(subscribeFilter)) {
  46.             this.filter = subscribeFilter;
  47.         }
  48.     }

CanalRabbitMQConsumer：rabbitmq 消费相关属性

  1. @SPI("rabbitmq")
  2. public class CanalRabbitMQConsumer implements CanalMsgConsumer {
  4.     private static final Logger                                logger              = LoggerFactory.getLogger(CanalRabbitMQConsumer.class);
  6.     // 链接地址
  7.     private String                                             nameServer;
  8.     // 主机名
  9.     private String                                             vhost;
  10.     private String                                             queueName;
  12.     // 一些鉴权信息
  13.     private String                                             accessKey;
  14.     private String                                             secretKey;
  15.     private Long                                               resourceOwnerId;
  16.     private String                                             username;
  17.     private String                                             password;
  19.     private boolean                                            flatMessage;
  21.     private Connection                                         connect;
  22.     private Channel                                            channel;
  24.     private long                                               batchProcessTimeout = 60 * 1000;
  25.     private BlockingQueue<ConsumerBatchMessage<CommonMessage>> messageBlockingQueue;
  26.     private volatile ConsumerBatchMessage<CommonMessage>       lastGetBatchMessage = null;
  28.     @Override
  29.     public void init(Properties properties, String topic, String groupId) {
  30.         this.nameServer = PropertiesUtils.getProperty(properties, "rabbitmq.host");
  31.         this.vhost = PropertiesUtils.getProperty(properties, "rabbitmq.virtual.host");
  32.         this.queueName = topic;
  33.         this.accessKey = PropertiesUtils.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESS_KEY);
  34.         this.secretKey = PropertiesUtils.getProperty(properties, CanalConstants.CANAL_ALIYUN_SECRET_KEY);
  35.         this.username = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_USERNAME);
  36.         this.password = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_PASSWORD);
  37.         Long resourceOwnerIdPro = (Long) properties.get(RabbitMQConstants.RABBITMQ_RESOURCE_OWNERID);

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

示例

创建 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
  4. # 创建用户、并授权
  5. mysql> create user canal identified with mysql_native_password by "123456";
  6. Query OK, 0 rows affected (0.01 sec)
  8. mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  9. Query OK, 0 rows affected (0.00 sec)
  11. mysql> flush privileges;
  12. Query OK, 0 rows affected (0.00 sec)
  15. # 创建数据库、表
  16. mysql> create database lihu;
  17. Query OK, 1 row affected (0.35 sec)
  19. mysql> use lihu;
  20. Database changed
  22. mysql> create table test(id int not null auto_increment primary key, value int not null);
  23. Query OK, 0 rows affected (0.67 sec)
  25. mysql> create table test2(id int not null auto_increment primary key, value int not null);
  26. Query OK, 0 rows affected (0.27 sec)

创建 canal server

  1. docker run -it -d --net fixed --ip 172.18.0.3 -p 11111:11111 --name canal-server \
  2. -v /usr/canal/mq/conf:/home/admin/canal-server/conf  canal/canal-server
  5. # canal.properties
  6. canal.serverMode = rocketmq
  7. rocketmq.namesrv.addr = 172.18.0.4:9876
  10. # instance.properties
  11. canal.instance.master.address=172.18.0.2:3306
  12. canal.instance.dbUsername=canal
  13. canal.instance.dbPassword=123456
  15. canal.mq.topic=example
  16. canal.mq.dynamicTopic=.*
  18. canal.mq.partition=0
  19. canal.mq.partitionsNum=3
  20. canal.mq.partitionHash=.*\\..*

创建 rocketmq（namesrv、broker）

  1. docker run -it -d --net fixed --ip 172.18.0.4 -p 9876:9876 \
  2. -e JAVA_OPT="-server -Xms256m -Xmx256m -Xmn128m" \
  3. --name namesrv apacherocketmq/rocketmq:4.9.0-alpine sh mqnamesrv
  5. docker run -it -d --net fixed --ip 172.18.0.5 -p 10911:10911 -p 10909:10909 \
  6. -e NAMESRV_ADDR="172.18.0.4:9876" \
  7. -e JAVA_OPT="-server -Xms2g -Xmx2g -Xmn1g" \
  8. -v /usr/rocketmq/test/broker.conf:/home/rocketmq/rocketmq-4.9.0/conf/broker.conf \
  9. --name broker apacherocketmq/rocketmq:4.9.0-alpine \
  10. sh mqbroker autoCreateTopicEnable=true -c /home/rocketmq/rocketmq-4.9.0/conf/broker.conf

创建 canal adapter

  1. docker run -it -d --net fixed --ip 172.18.0.6 -p 8081:8081 \
  2. -v /usr/canal/mq/adapter/conf/application.yml:/opt/canal-adapter/conf/application.yml \
  3. -v /usr/canal/mq/adapter/conf/rdb:/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: rocketMQ #tcp kafka rocketMQ rabbitMQ
  20.   flatMessage: true
  21.   zookeeperHosts:
  22.   syncBatchSize: 1000
  23.   retries: 0
  24.   timeout:
  25.   accessKey:
  26.   secretKey:
  27.   consumerProperties:
  28.     rocketmq.namespace:
  29.     rocketmq.namesrv.addr: 172.18.0.4:9876
  30.     rocketmq.batch.size: 1000
  31.     rocketmq.enable.message.trace: false
  32.     rocketmq.customized.trace.topic:
  33.     rocketmq.access.channel:
  34.     rocketmq.subscribe.filter:
  36.   srcDataSources:
  37.     defaultDS:
  38.       url: jdbc:mysql://172.18.0.2:3306/lihu?useUnicode=true
  39.       username: root
  40.       password: 123456
  41.   canalAdapters:
  42.   - instance: lihu # canal instance Name or mq topic name
  43.     groups:
  44.     - groupId: g1
  45.       outerAdapters:
  46.       - name: rdb
  47.         key: mysql1
  48.         properties:
  49.           jdbc.driverClassName: com.mysql.jdbc.Driver
  50.           jdbc.url: jdbc:mysql://172.18.0.7:3306/lihu?useUnicode=true
  51.           jdbc.username: root
  52.           jdbc.password: 123456
  55. ****************
  56. rdb/mytest_user.yml：适配器配置
  58. dataSourceKey: defaultDS
  59. destination: lihu
  60. groupId: g1
  61. outerAdapterKey: mysql1
  62. concurrent: true
  63. dbMapping:
  64.   mirrorDb: true
  65.   database: lihu

创建 mysql2实例（目标数据库）

  1. docker run -it -d --net fixed --ip 172.18.0.7 -p 3307:3306 --privileged=true \
  2. --name mysql2 -e  MYSQL_ROOT_PASSWORD=123456 mysql
  4. # 创建数据库（不需要创建表）
  5. mysql> create database lihu;
  6. Query OK, 1 row affected (0.00 sec)

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

使用测试

源数据库 mysql：插入数据

  1. mysql> insert into test(id,value) values(1,2);
  2. Query OK, 1 row affected (0.11 sec)
  4. mysql> insert into test(id,value) values(2,2);
  5. Query OK, 1 row affected (0.15 sec)
  7. mysql> insert into test2(id,value) values(1,2);
  8. Query OK, 1 row affected (0.03 sec)
  10. mysql> insert into test2(id,value) values(2,2);
  11. Query OK, 1 row affected (0.11 sec)

canal adapter 日志

  1. 2021-07-14 22:36:09.564 [pool-7-thread-1] DEBUG c.a.o.c.c.adapter.rdb.service.RdbMirrorDbSyncService - DDL: {"data":null,"database":"lihu","destination":"lihu","es":1626273340000,"groupId":"g1","isDdl":true,"old":null,"pkNames":null,"sql":"create table test(id int not null auto_increment primary key, value int not null)","table":"test","ts":1626273345726,"type":"CREATE"}
  2. 2021-07-14 22:42:00.785 [pool-6-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"2","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test","type":"INSERT"}
  3. 2021-07-14 22:44:48.353 [pool-7-thread-1] DEBUG c.a.o.c.c.adapter.rdb.service.RdbMirrorDbSyncService - DDL: {"data":null,"database":"lihu","destination":"lihu","es":1626273887000,"groupId":"g1","isDdl":true,"old":null,"pkNames":null,"sql":"create table test2(id int not null auto_increment primary key, value int not null)","table":"test2","ts":1626273888331,"type":"CREATE"}
  4. 2021-07-14 22:44:54.400 [pool-5-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"1","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test2","type":"INSERT"}
  5. 2021-07-14 22:44:57.226 [pool-6-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"2","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test2","type":"INSERT"}
  6. 2021-07-14 22:57:26.640 [pool-7-thread-1] DEBUG c.a.o.c.c.adapter.rdb.service.RdbMirrorDbSyncService - DDL: {"data":null,"database":"lihu","destination":"lihu","es":1626274646000,"groupId":"g1","isDdl":true,"old":null,"pkNames":null,"sql":"create table test3(id int not null auto_increment primary key, value int not null)","table":"test3","ts":1626274646540,"type":"CREATE"}
  7. 2021-07-14 22:57:34.374 [pool-6-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"2","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test3","type":"INSERT"}
  8. 2021-07-14 22:59:08.462 [pool-5-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"1","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test","type":"DELETE"}
  9. 2021-07-14 22:59:16.007 [pool-5-thread-1] DEBUG c.a.o.canal.client.adapter.rdb.service.RdbSyncService - DML: {"data":{"id":"1","value":"2"},"database":"lihu","destination":"lihu","old":null,"table":"test","type":"INSERT"}

目标数据库 mysql2：查看数据

  1. mysql> select * from test;
  2. +----+-------+
  3. | id | value |
  4. +----+-------+
  5. |  1 |     2 |
  6. |  2 |     2 |
  7. +----+-------+
  8. 2 rows in set (0.00 sec)
  10. mysql> select * from test2;
  11. +----+-------+
  12. | id | value |
  13. +----+-------+
  14. |  1 |     2 |
  15. |  2 |     2 |
  16. +----+-------+

mysql2 可以同步 mysql中的数据

发表评论

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

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

相关阅读

    相关 canal实现mysql数据同步

    前言 canal是实现mysql数据备份，异地灾备，异地数据同步等重要的中间件，在实际的业务场景中有着广泛的使用，本文基于小编所在项目中一个异地数据同步的场景为例，通过案

    水深无声水深无声/ 0 赞/ 88 阅读

    相关 canal mysql 数据同步

    首先canal是什么呢？     canal是阿里巴巴旗下的一款开源项目，纯Java开发。基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了MySQL；简单来说，

    落日映苍穹つ落日映苍穹つ/ 0 赞/ 308 阅读