1、部署configmap
vim canal-config.yaml
apiVersion: v1 kind: ConfigMap metadata:name: canal-confignamespace: canal data:canal.properties: |# ========== Canal基础配置 ==========canal.id = 1canal.port = 11111canal.destinations = example# ========== 持久化路径 ==========canal.file.data.dir = /home/admin/canal-server/logs/db# ========== Zookeeper配置 ==========canal.zkServers = z-2.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:2181,z-1.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:2181,z-3.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:2181# ========== Server模式配置 ==========canal.serverMode = kafka# ========== Kafka生产者配置 ==========canal.mq.servers = b-2.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:9092,b-1.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:9092,b-3.subotizstgmsk.ck8tsh.c2.kafka.ap-southeast-1.amazonaws.com:9092canal.mq.topic = subotiz-canal-merchantdbcanal.mq.flatMessage = truecanal.mq.partitionsNum = 1canal.mq.maxRequestSize = 524288# Kafka生产者高级配置canal.mq.kafka.producer.retries = 5canal.mq.kafka.producer.retry.backoff.ms = 1000canal.mq.kafka.producer.retry.backoff.max.ms = 5000canal.mq.kafka.producer.metadata.fetch.timeout.ms = 120000canal.mq.kafka.producer.metadata.max.age.ms = 300000# ========== 动态Topic配置 ==========canal.mq.dynamicTopic = canal-subotiz-test-table:test1_sre\\.test_users canal.mq.dynamicTopicPartitionNum = subotiz-canal-merchantdb:1canal.mq.partitionHash = .*\..*# ========== 自动扫描配置 ==========canal.auto.scan = truecanal.auto.scan.interval = 5# instance.properties 实例配置instance.properties: |# ========== MySQL连接配置 ==========canal.instance.master.address = mypaydb1.cluster-cl20ky4iyqmn.ap-southeast-1.rds.amazonaws.com:3306canal.instance.dbUsername = canalcanal.instance.dbPassword = vGgRTZubA2qsI2mNcanal.instance.connectionCharset = UTF-8# ========== 表过滤配置 ==========# 白名单正则（同步的库表）canal.instance.filter.regex = (test1_sre)\..*# 黑名单正则（不同步的库表）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# Query语句过滤canal.instance.filter.query = truecanal.instance.filter.query.ddl = truecanal.instance.filter.query.dml = truecanal.instance.filter.query.regex = ^(SELECT|SHOW|DESC|EXPLAIN).*$# ========== 心跳检测 ==========canal.instance.filter.heartbeat = true# ========== DDL解析 ==========canal.instance.parse.ddl = true# ========== 表元数据配置 ==========canal.instance.filter.table.meta = truecanal.instance.tableMetaCache.enable = false# ========== TSDB表结构存储 ==========canal.instance.tsdb.enable = true# ========== 默认GTID配置 ==========canal.instance.gtidon = false# ========== 内存配置 ==========canal.instance.memory.buffer.size = 16384canal.instance.memory.buffer.memunit = 1024canal.instance.transaction.size = 1024# ========== 并行解析 ==========canal.instance.parser.parallel = true
kubectl apply -f canal-config.yaml
2、部署statefulset
vim canal-statefulset.yaml
--- apiVersion: v1 kind: Service metadata:name: canal-servicenamespace: canal spec:selector:app: canal-serverports:- name: canalport: 11111targetPort: 11111type: ClusterIP --- apiVersion: apps/v1 kind: StatefulSet metadata:name: canal-servernamespace: canal spec:serviceName: canal-servicereplicas: 1selector:matchLabels:app: canal-servertemplate:metadata:labels:app: canal-serverspec:securityContext:runAsUser: 0runAsGroup: 0fsGroup: 1000containers:- name: canal-serverimage: canal/canal-server:v1.1.8imagePullPolicy: IfNotPresentports:- containerPort: 11111name: canalenv:- name: JAVA_OPTSvalue: "-Xms2g -Xmx4g -XX:+UseG1GC"command: ["/bin/sh", "-c"]args:- |set -e# 配置修改函数（仅改sed分隔符，其余不变）update_config() {local config_file=$1local custom_config=$2while IFS='=' read -r key value; doif [[ -z "$key" || "$key" =~ ^# ]]; thencontinuefilocal clean_key=$(echo "$key" | tr -d ' \t' | xargs)local clean_value=$(echo "$value" | xargs)local standard_line="$clean_key = $clean_value"if grep -q "^[ \t]*$clean_key[ \t]*=" "$config_file"; thensed -i "s@^\([ \t]*$clean_key[ \t]*=[ \t]*\).*@\1$clean_value@" "$config_file"echo "修改配置项：$clean_key = $clean_value"elseecho "$standard_line" >> "$config_file"echo "新增配置项：$clean_key = $clean_value"fidone < "$custom_config"}# 1. 处理全局配置（指���example实例）if [ ! -f /home/admin/canal-server/conf/canal.properties ]; thencp /home/admin/canal-server/conf/canal.properties.example /home/admin/canal-server/conf/canal.propertiesfiecho "===== 处理canal.properties ====="update_config "/home/admin/canal-server/conf/canal.properties" "/tmp/config/canal.properties"# 2. 核心：直接修改example实例的配置（无需创建自定义目录）echo "===== 处理example实例配置 ====="update_config "/home/admin/canal-server/conf/example/instance.properties" "/tmp/config/instance.properties"# 3. 权限配置（仅给example实例赋权）chown -R admin:admin /home/admin/canal-server/conf/example/chmod -R 755 /home/admin/canal-server/conf/mkdir -p /home/admin/canal-server/data/db /home/admin/canal-server/data/logchown -R admin:admin /home/admin/canal-server/data/db /home/admin/canal-server/data/log# 4. 验证配置echo "===== 最终配置验证 ====="grep "canal.destinations" /home/admin/canal-server/conf/canal.propertiesgrep "canal.instance.master.address" /home/admin/canal-server/conf/example/instance.propertiesecho "========================"# 5. 启动Canal（原生方式，自动加载example实例）/home/admin/app.shresources:requests:memory: "1Gi"cpu: "500m"limits:memory: "3Gi"cpu: "2000m"volumeMounts:- name: canal-configmountPath: /tmp/config- name: canal-datamountPath: /home/admin/canal-server/logsvolumes:- name: canal-configconfigMap:name: canal-configvolumeClaimTemplates:- metadata:name: canal-dataspec:accessModes: ["ReadWriteOnce"]storageClassName: "gp2-ap-southeast-1b"resources:requests:storage: 50Gi
kubectl apply -f canal-statefulset.yaml