canal 1.1.5同步mysql到kafka

參考：https://blog.csdn.net/ashic/article/details/104722975

參考：https://blog.csdn.net/nmsoftklb/article/details/149118188

canal的github地址：https://github.com/alibaba/canal

canal安裝和canal和kafka結合使用：https://blog.csdn.net/weixin_42191498/article/details/129153667

提示：canal版本差距較大，1.1.5使用的是舊架構。新版本架構是MySQL -> Canal Server -> Canal Adapter -> Kafka，舊架構是MySQL -> Canal Server -> Kafka

目前我的canal部署在k8s上，所以配置使用的是configmap

1、需求：merchantdb庫的所有表同步到kafka的subotiz-canal-merchantdb的topic，paymentdb庫所有表同步到kafka的subotiz-canal-paymentdb的topic，tradingdb庫所有表同步到kafka的subotiz-canal-tradingdb的topic

vim canal-cm-databases-dynamic.yaml

apiVersion: v1
data:
  canal_destinations: canal-merchantdb-dynamic
  canal_instance_connectionCharset: UTF-8
  canal_instance_dbPassword: vGgRTZubA2qsI2mN
  canal_instance_dbUsername: canal
  canal_instance_filter_regex: (merchantdb|paymentdb|tradingdb)\\..*
  #canal_instance_filter_regex: merchantdb\\..*
  canal_instance_filter_black_regex: (mysql|dts|mqbone|oauth2_production|account|act|crab|crawfish|datafountain|email_api_production|exec|fortunecat|kirin|notify_api_production|pangu_db|leopard)\\\\..*,tiger\\\\._order_transactions_new,main\\\\.menus,main\\\\.menu_items
  canal_instance_master_address: pc-rj9y7id894cs4ld05.rwlb.rds.aliyuncs.com:3306
  #canal_mq_dynamicTopic: merchantdb\\.*:subotiz-canal-merchantdb,paymentdb\\.*:subotiz-canal-paymentdb,tradingdb\\.*:subotiz-canal-tradingdb
  canal_mq_dynamicTopic: subotiz-canal-merchantdb:merchantdb,subotiz-canal-paymentdb:paymentdb,subotiz-canal-tradingdb:tradingdb
  canal_mq_dynamicTopicPartitionNum: subotiz-canal-merchantdb:3,subotiz-canal-paymentdb:3,subotiz-canal-tradingdb:3
  #canal_mq_dynamicTopicPartitionNum: subotiz-canal-merchantdb:3
  canal_mq_flatMessage: "true"
  canal_mq_maxRequestSize: "52428800"
  canal_mq_partitionHash: .*\..*:$pk$
  canal_mq_partitionsNum: "3"
  canal_mq_servers: alikafka-post-cn-qzm4d4obv002-3-vpc.alikafka.aliyuncs.com:9092
  canal_mq_topic: subotiz-canal-merchantdb
  #canal_mq_topic: subotiz-canal-default
  canal_serverMode: kafka
  canal_zkServers: mse-f01be7312-zk.mse.aliyuncs.com:2181
  kafka_bootstrap_servers: alikafka-post-cn-qzm4d4obv002-3-vpc.alikafka.aliyuncs.com:9092
  kafka_max_request_size: "92428800"
kind: ConfigMap
metadata:
  name: canal-cm-merchantdb
  namespace: staging

2、需求：指定某個數據庫的某個表同步到kafka的某個topic，如tradingdb庫的trade表同步到Topic: subotiz-canal-tradingdb-trade、tradingdb庫的refund_record表同步到Topic: subotiz-canal-tradingdb-refund-record、 tradingdb庫的checkout_session表同步到Topic: subotiz-canal-tradingdb-checkout-session、merchantdb庫的customer表同步到Topic: subotiz-canal-merchantdb-customer

vim canal-cm-tables-dynamic.yaml

apiVersion: v1
data:
  canal_destinations: canal-tables-dynamic
  canal_instance_connectionCharset: UTF-8
  canal_instance_dbPassword: vGgRTZubA2qsI2mN
  canal_instance_dbUsername: canal
  canal_instance_filter_regex: tradingdb\\.trade|tradingdb\\.refund_record|tradingdb\\.checkout_session|merchantdb\\.customer
  canal_instance_filter_black_regex: (mysql|dts|mqbone|oauth2_production|account|act|crab|crawfish|datafountain|email_api_production|exec|fortunecat|kirin|notify_api_production|pangu_db|leopard)\\\\..*,tiger\\\\._order_transactions_new,main\\\\.menus,main\\\\.menu_items
  canal_instance_master_address: pc-rj9y7id894cs4ld05.rwlb.rds.aliyuncs.com:3306
  canal_mq_dynamicTopic: subotiz-canal-tradingdb-trade:tradingdb\\.trade,subotiz-canal-tradingdb-refund-record:tradingdb\\.refund_record,subotiz-canal-tradingdb-checkout-session:tradingdb\\.checkout_session,subotiz-canal-merchantdb-customer:merchantdb\\.customer
  canal_mq_dynamicTopicPartitionNum: subotiz-canal-tradingdb-trade:5,subotiz-canal-tradingdb-refund-record:5,subotiz-canal-tradingdb-checkout-session:5
  #canal_mq_dynamicTopicPartitionNum: subotiz-canal-merchantdb:3
  canal_mq_flatMessage: "true"
  canal_mq_maxRequestSize: "52428800"
  canal_mq_partitionHash: .*\..*:$pk$
  canal_mq_partitionsNum: "5"
  canal_mq_servers: alikafka-post-cn-qzm4d4obv002-3-vpc.alikafka.aliyuncs.com:9092
  canal_mq_topic: subotiz-canal-merchantdb
  #canal_mq_topic: subotiz-canal-default
  canal_serverMode: kafka
  canal_zkServers: mse-f01be7312-zk.mse.aliyuncs.com:2181
  kafka_bootstrap_servers: alikafka-post-cn-qzm4d4obv002-3-vpc.alikafka.aliyuncs.com:9092
  kafka_max_request_size: "92428800"
kind: ConfigMap
metadata:
  name: canal-cm-tables-dynamic
  namespace: staging

posted @ 2025-09-09 10:40 苦逼yw 閱讀(16) 評論(0) 收藏舉報

刷新頁面返回頂部

lfxx

canal 1.1.5同步mysql到kafka