CDCChange Data Capture变更数据捕获是这几年来数据库圈子里绕不开的话题。尤其当业务系统需要实时感知数据库里的增删改查把数据同步到消息队列、数仓、缓存或者另一个数据库时CDC几乎成了标准答案。我最早接触这个场景是被迫的——当时要做一个跨库的实时数据同步既要保证低延迟又不能让业务库承担太多压力试了一圈轮询、触发器方案都不满意最后把目光落在了基于日志的CDC上。这篇文章就把我这几年做CDC实时监听数据库的方案选型、实操踩坑、参数调优一次性讲清楚项目标题很简单但背后能展开的东西非常多。这篇内容适合谁看一类是刚开始接触数据同步、实时数仓的同学想知道CDC到底是什么、跟普通的定时捞数据有什么区别另一类是已经在用CDC工具但被各种细节折磨过的开发比如binlog格式不对导致解析失败、快照阶段锁表、消费积压追不上这篇文章里的排查思路大概率能帮到你。1. 内容整体设计与思路拆解1.1 先搞清楚CDC的本质是什么很多人一听到“实时监听数据库”第一反应是写个定时任务每秒钟查一次表把变化的数据捞出来。这种做法也能叫“监听”但不是真正的实时而且有几个很难绕过去的坑轮询间隔越小对数据库压力越大高频轮询对线上业务库的查询开销不可忽视如果表没有自增ID或更新时间字段你根本没法判断哪些行是新增、哪些是修改物理删除的数据轮询根本发现不了除非走逻辑删除的歪路。CDC的核心思路完全不一样它不主动去问数据库“你有没有变化”而是让数据库自己把变化“说出来”。实现方式有几种但最主流、最可靠的是基于数据库的事务日志比如MySQL的binlog、PostgreSQL的WAL、SQL Server的transaction log。数据库每执行一次增删改操作都会写一条日志记录CDC工具解析这些日志就知道哪张表、哪一行、在什么时间、做了什么操作。数据库本身就用这套日志做主从复制和崩溃恢复CDC相当于是把主从复制的思路挪到了数据集成场景本质上就是“让另一个消费者去读复制日志”。这个思路带来的好处是实实在在的对源库几乎没有额外压力因为读的是日志不是表数据天然支持删除和更新操作不需要业务表任何配合延迟能做到毫秒或秒级取决于下游消费链路。1.2 为什么说这是数据架构里的“稳定支点”我后来把CDC当成一个独立的基础设施层来看待而不是某个工具或一段脚本。意思是说数据库变更数据是业务系统的真相来源CDC负责把这个真相高效地、无侵入地分发出去下游想怎么用都行。数据同步到数仓CDC替代了原先的凌晨ETL批处理数据同步到缓存CDC替代了手工清缓存和双写的一致性难题CDC加上消息队列还能支撑事件驱动架构——比如订单表有变化就产生一个领域事件触发后续的库存扣减、通知推送。我在这几年的实操里发现团队对数据一致性的焦虑大部分都能靠CDC缓解。举个例子很多项目在双写缓存时都会纠结先更新数据库还是先写缓存、失败了怎么办一旦引入CDC监听数据库变更并同步到缓存业务侧只需要操作数据库缓存更新交给CDC链路去处理少了一大堆业务代码里的一致性逻辑。当然这不是万能药CDC链路本身也有延迟和数据丢失的可能但至少比业务侧手工双写要干净得多。1.3 一个完整的CDC链路长什么样为了后面讲实操不绕弯先给一个整体框架。一个典型的CDC实时监听数据库架构通常包含四个环节源数据库、CDC采集端、消息中间件、消费端。源数据库很好理解就是那个产生变更数据的库。CDC采集端负责连接源库、读取binlog以MySQL为例、解析成结构化的变更事件再序列化发送出去。市面上常见的采集端有Debezium、Canal、Flink CDC等每一种的定位和适用场景都有差别我会在下一章详细对比。消息中间件的作用有两个一是缓冲和解耦采集端把数据持续写入消息队列消费端按自己的速度去消费不会因为下游业务处理慢而拖垮采集端二是让CDC链路具备多播能力一套变更数据可以被多个下游同时消费比如既同步到Elasticsearch又同步到数据仓库。消费端就是各类目标系统收到变更事件后执行相应的写入逻辑。我在很多文章里看到有人把Debezium直接往Kafka里塞其实这是Debezium一个非常典型的部署形态Kafka在中间扮演的正是消息缓冲和分发角色。后面我也会基于这个组合给出一份能直接跑的实操配置。2. 方案选型市面上主流的CDC实现方式对比2.1 基于日志解析的CDC工具一劳永逸的主流选择基于日志的方案MySQL这边最常用的是Canal和Debezium两者都伪装成MySQL的从库去拉binlog核心机制相同但生态定位不同。Canal是阿里开源的项目很多早期做大数据同步的公司都靠它支撑社区成熟度很高Canal适配的消息通道包括Kafka、RocketMQ等运维上主要是通过Canal Admin做集群管理。Debezium是Red Hat主导的开源项目基于Kafka Connect框架构建最大好处是天然融入Kafka生态source connector、sink connector统一管理而且支持MySQL、PostgreSQL、SQL Server、Oracle、MongoDB等一大票数据库一个框架走天下。我个人的经验是如果你的基础设施已经围绕Kafka展开团队也熟悉Kafka Connect优先选Debezium因为它有统一的connector生命周期管理、资源配置和日志体系出问题了排查路径也比较清晰。如果就是想快速把MySQL的数据同步到另一个MySQL或者RocketMQ团队也不想引入Kafka Connect这套东西Canal更轻量一些直接部署一个Java进程配置好canal.properties和instance.properties就能跑起来。Flink CDC是后来居上的热门选择它最大卖点是结合Flink的流式计算能力能直接做“数据库入湖入仓”的实时ETL。Flink CDC对无主键表、DDL变更、全量增量自动切换的处理也很成熟而且Flink CDC 2.x支持了增量快照和多并行读取吞吐上限远高于传统单线程单连接的模式。不过它的部署复杂度也更高需要一套Flink集群适合本身就在做Flink实时计算的项目如果只是单纯做数据变更订阅用Debezium或Canal更轻。2.2 基于轮询与触发器的旁路方案能用但尽量别碰考虑到不少读者是新手我还是介绍一下轮询和触发器这两种方案然后说清楚为什么它们只适合小规模场景。轮询方案就是开定时任务周期性执行对更新时间字段的SELECT查询比如WHERE update_time 上次记录的游标值。配置门槛极低任何懂SQL的人上手就能做数据同步延迟取决于轮询间隔最坏情况下会差一个周期。但问题在于物理删除无法感知除非业务层改成逻辑删除高频轮询对数据库主库会产生额外开销线上高并发场景容易受影响而且每次轮询都需要带上一次游标当数据量超过千万级时查询本身也会变慢。触发器方案就是在业务库上建AFTER INSERT/UPDATE/DELETE触发器把变更记录写进一张日志表应用再去日志表里取数据。这个方案确实能捕获所有变更包括物理删除但它属于侵入式方案要给业务库额外建表、建触发器相当于给本就繁忙的业务数据库又加了一层负担而且触发器里写日志通常是同步事务的一部分一旦日志表写入慢业务事务也会被拖慢。最要命的是数据库迁移、DDL变更时触发器极易出问题我已经不止一次见过因为触发器没同步迁移导致数据漏采集的事故。所以我的建议是触发器方案只适合没有二进制日志可用的老旧数据库做临时补救生产环境尽量别长期依赖。2.3 不同方案的选型对照与适用场景方案延迟级别侵入性全量增量支持数据库类型覆盖适用场景基于日志Debezium/Canal/Flink CDC毫秒~秒级几乎无侵入好广MySQL/PG/Oracle/SQL Server等生产环境首选实时数仓、异构同步轮询UPDATE_TIME秒~分钟级需业务表有更新字段弱任意数据库数据量小、对实时性要求低触发器写日志表毫秒~秒级强侵入需建触发器和日志表一般支持的数据库均可无binlog可用时的临时方案数据库原生复制秒级无侵入不支持异构原生同构数据库的主从复制我自己的选型逻辑很简单第一步看源数据库有没有可用的日志机制MySQL就把binlog打开PostgreSQL就调wal_levellogical只要数据库本身支持逻辑复制直接选基于日志的方案第二步看团队已有技术栈如果已经在用Flink就顺着Flink CDC用否则Kafka生态选Debezium轻量场景选Canal第三步才考虑业务表能不能加更新字段、能不能接受逻辑删除这类妥协方案。3. 实操用Debezium Kafka实现MySQL实时监听3.1 环境准备与版本选择实操部分我用的是Docker Compose方式搭建Debezium Kafka环境这是最省事的部署方式适合开发调试也适合快速验证方案可行性。如果你是在生产环境建议把Zookeeper、Kafka、Kafka Connect分别部署成集群但这套搭建思路跟开发环境没有本质区别。我自己测试时使用的版本组合是Debezium2.3.x该版本之前需要Java 11新版本基于Java 17注意JRE版本Kafka Kafka Connect3.4.xMySQL8.0.x开启binlog消息格式JSON或者Confluent Avro开发阶段用JSON最简单生产推荐Avro带Schema RegistryDocker Compose的基础编排文件大致长这样version: 3.8 services: mysql: image: mysql:8.0 container_name: cdc-mysql environment: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: appdb command: [ --server-id223344, --log-binmysql-bin, --binlog-formatROW, --binlog-row-imageFULL, --gtid-modeON, --enforce-gtid-consistencyON ] ports: - 3306:3306 zookeeper: image: confluentinc/cp-zookeeper:latest container_name: cdc-zookeeper environment: ZOOKEEPER_CLIENT_PORT: 2181 kafka: image: confluentinc/cp-kafka:latest container_name: cdc-kafka ports: - 9092:9092 environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 depends_on: - zookeeper connect: image: debezium/connect:2.3 container_name: cdc-connect ports: - 8083:8083 environment: BOOTSTRAP_SERVERS: kafka:9092 GROUP_ID: 1 CONFIG_STORAGE_TOPIC: my_connect_configs OFFSET_STORAGE_TOPIC: my_connect_offsets STATUS_STORAGE_TOPIC: my_connect_statuses depends_on: - kafka - mysql这里有几个细节值得说一下。MySQL的server_id不能跟任何从节点冲突如果不设置可能默认成1在多实例场景下会导致两个节点争抢同一个server_idbinlog拉取会出现异常断开。binlog-format必须设置为ROW因为只有ROW格式的binlog才会记录每行变更前的全量字段值和变更后的值STATEMENT格式只记录SQL语句逻辑复制的解析效率和准确性都差很多。binlog-row-imageFULL是为了保证binlog里保存一行所有字段的镜像这样才能在下游重建出完整的数据行。3.2 注册MySQL Connector让Debezium开始干活环境起来之后需要在Kafka Connect里注册一个MySQL source connector。这一步是通过REST API完成的向Connect服务的/connectors端点提交一个JSON配置。我生产环境里反复调整过这份配置核心字段如下{ name: mysql-appdb-connector, config: { connector.class: io.debezium.connector.mysql.MySqlConnector, database.hostname: mysql, database.port: 3306, database.user: cdc_user, database.password: cdc_password, database.server.id: 223344, database.server.name: appdb-server, database.include.list: appdb, table.include.list: appdb.user,appdb.orders, database.history.kafka.topic: schema-changes.appdb, include.schema.changes: true, snapshot.mode: initial, key.converter: org.apache.kafka.connect.json.JsonConverter, value.converter: org.apache.kafka.connect.json.JsonConverter, key.converter.schemas.enable: false, value.converter.schemas.enable: false } }提交配置的命令很简单一个curl就搞定curl -X POST -H Content-Type: application/json -d mysql-connector.json http://localhost:8083/connectors这个配置里最重要的参数其实有三组。database.hostname和database.user配置的是Debezium用来连接MySQL的账号这个账号必须要有的权限包括SELECT、RELOAD、SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT。RELOAD权限是因为Debezium在进行初始快照时需要FLUSH TABLE WITH READ LOCK来获取一致性位点。REPLICATION相关权限是为了能够开始和停止binlog流。线上用root账号拉binlog是非常不建议的我在生产环境都是单独建一个最小权限的账号给CDC专用方便权限审计和回收。database.server.name是Kafka Topic命名的前缀。默认情况下Debezium监听的每个表都会生成一个topic命名规则是{database.server.name}.{database}.{table}比如上面配置里user表就会写到appdb-server.appdb.user这个topic。如果schema变化了还会写到schema-changes.appdb这个history topic里。这个命名规则很容易被忽略但排查消费端配置时非常关键后续我会在常见问题里再提。snapshot.mode参数控制Debezium启动时如何做初始快照。initial模式会在启动时先对当前全量数据做一次一致性快照然后无缝切换到binlog增量监听schema_only模式只捕获表结构不搬运存量数据适合下游已经有历史数据的场景when_needed模式则是先尝试纯增量发现需要快照时再补这个模式用得相对少never模式则完全不做快照只从binlog当前位点开始读但这样做有个前提就是下游明确知道自己要从哪个点开始接收。3.3 验证链路看数据是否真的实时流过来完整链路搭建完成后验证方式比较简单。用MySQL客户端往user表里插入数据INSERT INTO appdb.user (id, name, email) VALUES (1, 张三, zhangsanexample.com); UPDATE appdb.user SET name 李四 WHERE id 1; DELETE FROM appdb.user WHERE id 1;然后通过kafka-console-consumer消费对应的topickafka-console-consumer --bootstrap-server localhost:9092 --topic appdb-server.appdb.user --from-beginning正常情况下消费端会看到三条消息分别对应INSERT、UPDATE、DELETE三个操作。INSERT消息的payload结构大概是这样的{ before: null, after: { id: 1, name: 张三, email: zhangsanexample.com }, source: { version: 2.3.0.Final, connector: mysql, name: appdb-server, ts_ms: 1714095820000, snapshot: false, db: appdb, table: user, server_id: 223344, binlog_file: mysql-bin.000003, binlog_position: 245, gtid: abc-123-456 }, op: c, ts_ms: 1714095820123 }这里的op字段是操作类型标识c代表createINSERTu代表updated代表delete。Update操作和Delete操作的before字段里会带上被修改前的完整行数据这就是binlog-row-imageFULL的结果。这条链路验证通过意味着MySQL的每一次增删改查准确的说是INSERT/UPDATE/DELETE都会在毫秒级被Debezium解析成结构化JSON写入Kafka对应topic下游系统只需要订阅这些topic就能获得全量实时的数据变更流。3.4 全量快照与增量无缝衔接的机制很多刚接触CDC的人都会有一个疑问如果目标库里是空的我需要把源库已有的几千万条历史数据也同步过来该怎么做Debezium的initial快照模式就是干这个的。具体流程是Connector启动时先开启一个数据库连接获取当前binlog的位点即一个一致性坐标然后在这个位点上建立一致性快照。快照期间源库的增量变更会继续写binlog不会被阻塞。快照读取完所有存量数据后Debezium会从之前记录的binlog位点继续消费增量这样全量数据和期间产生的增量数据在逻辑上是无缝衔接的不会丢也不能重复。这里有个性能细节值得展开Debezium早期版本做全量快照是单线程主键分页读取速度对大表来说确实比较感人。2.x版本引入了增量快照机制Incremental Snapshot可以把快照任务拆分成多段并行执行通过signal topic控制触发停止和恢复实测大表的快照速度提升非常明显。如果你表的数据量在百万级以上建议仔细看看官方文档里关于read.only的连接建议和snapshot.fetch.size参数的调优。snapshot.fetch.size控制每次快照批次的读取行数默认值通常是1024。调大这个值可以减少往返数据库的次数从而提升速度但会增加内存占用和单次事务时间生产上需要根据表宽和网络延迟测试调整。4. 核心细节解析与实操要点4.1 binlog格式与关键参数监听前的第一道坎如果说CDC监听MySQL是座大坝binlog就是坝体的地基。地基没打好后面解析什么都白搭。我在做实战辅导时发现很多人的CDC链路连不通往上排查半天最后都是binlog配置的问题。所以把这块单独拎出来逐条讲讲关键参数的选值逻辑。先说binlog_format这是一个最核心的参数。MySQL支持STATEMENT、ROW、MIXED三种格式。STATEMENT格式记录的是SQL原文占用空间小但无法精确还原每行数据的变化ROW格式记录的是每行变更前后的镜像数据信息量最大MIXED是MySQL自己根据语句类型决定用STATEMENT还是ROW记录。做CDC一定要用ROW这是Debezium和Canal的共同强制要求因为只有ROW格式才能提供足够的信息在异构目标端重建数据行。再说binlog_row_image这个参数会影响ROW格式下记录的字段内容可以设为FULL、MINIMAL、NOBLOB。FULL记录所有列的前后镜像值MINIMAL只记录能够唯一标识这一行的列通常是主键和发生变更的列NOBLOB则是FULL的一个变体大字段不记录。做CDC强烈建议用FULL你自己想像一下假如下游要拿变更数据重建Elasticsearch的文档需要这一行的完整字段MINIMAL模式下某些没变更过的字段根本不在binlog里重建出的JSON就是不完整的。NOBLOB模式更坑一旦表里有TEXT/BLOB类型字段数据在binlog里只出现一个占位符号下游只能拿到一个残破的行。还有一个容易被忽略的参数是expire_logs_days或binlog_expire_logs_seconds它控制binlog文件的保留时长。CDC消费者不是实时在线的假如你停掉connector去维护几天后重启如果binlog文件已经被清理Debezium就会报找不到位点。生产经验是保留至少72小时以上binlog我一般设置binlog_expire_logs_seconds604800即7天。运维上还需要定期检查binlog文件占用的磁盘空间防止过大撑爆存储。4.2 主键与无主键表的处理原则CDC监听过程中有一类场景特别容易让人翻车源表没有主键。Debezium默认情况下要求表必须有主键因为主键有双重作用一是作为快照时的分页游标二是作为Kafka消息key决定同一行数据在topic中的分区策略。对于无主键的表Debezium会在启动快照时报错。解决方案有两条路一是给源表加上主键这对有严格规范的库来说是最好的做法二是利用message.key.columns参数指定一个能唯一标识行的候选列组合比如唯一索引键或业务订单号。但要注意如果基于业务列做消息key而业务列本身后来又发生变化比如订单号修改那么同一行在Kafka里会从分区A跳到分区B下游如果按分区维护状态就会出问题。还有一个我没踩过但要提醒的点无主键且无唯一索引的表如果又没有合适业务列做消息keyDebezium会退化到把所有数据写入同一个分区这会让数据倾斜非常严重生产环境的大表和热点数据场景下Kafka某个分区会被写爆而其他分区闲置。遇到这种情况建议跟业务方坐下来把主键梳理清楚或者由上游应用在写入时生成一个不可变的逻辑ID字段。4.3 Kafka Topic与消息结构解析前面提过Topic命名规则是database.server.name.database.table还有一项容易被忽略的是Debezium还允许你用topic.prefix配置来改变这个前缀命名。比如在Debezium 2.x中官方把database.server.name改名为topic.prefix。我因为升级版本遇到过老配置失效的问题新版本的connector启动直接报错说找不到配置字段这一点在阅读社区文档时一定要注意版本差异。Kafka消息里除了业务数据还有大量的元数据字段比如source块中的binlog_file和binlog_position。这些信息非常有用比如可以精确到某条变更来自哪个binlog文件、哪个位置做故障恢复和位点对比时价值极大。还有op字段c/r/u/d分别代表create、read快照阶段、update、delete。你在写下游消费程序时不要只处理c/u/d快照阶段的r类型是全量数据开始的标记漏处理的话初始同步会丢数据。快照阶段的r消息和增量阶段的c消息从数据结构角度几乎一样唯一的区别是source.snapshot字段值。所以处理下游逻辑时最好统一按after字段的最终值去做幂等upsert对r/c/u用相同逻辑写入目标对d执行删除。这套逻辑是我在实践中摸索出来的通用模式适配大部分同步场景。4.4 数据一致性至少一次与幂等消费的正确理解CDC链路里“至少一次”投递语义是绕不开的话题。Debezium通过记录Kafka的offset和MySQL的binlog位置来保证恢复后不丢数据但故障恢复时可能出现重复消息因为消费者提交offset到Kafka的时机和数据处理完成时机不可能做到完全原子。这意味着下游必须支持幂等重复消息来一遍结果应该跟来一遍一样。比如你同步到MySQL可以用REPLACE INTO或INSERT ... ON DUPLICATE KEY UPDATE同步到Elasticsearch直接对同一id做index操作就是天然的幂等同步到RedisSET操作天然幂等。遇到DELETE操作也建议用幂等方式比如先delete再insert或者干脆发一个tombstone标记消息让下游自行处理。我在真实项目中因为忽略了幂等出现过一次比较难受的事故当时同步到Elasticsearch时没有用upsert而是先查后插结果重复消息导致文件里写入重复数据搜索得分全乱了。从此以后我给自己立了一个规矩下游写入一律默认幂等除非你能证明目标系统天然幂等。这个经验同样可以推广到所有做实时数据同步的团队。5. 常见问题与排查技巧实录5.1 Connector启动就报错binlog配置没到位症状是提交connector后connect日志里不断重试错误信息指向“The database server is not configured to use row-based binary logging”之类。这个问题的原因字面上已经说了binlog_format没有设成ROW。我在Docker环境里看过很多次MySQL官方镜像默认binlog_format其实不是ROW必须通过command参数主动覆盖。排查和修复路径比较直接SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_row_image;如果binlog_format不是ROW需要在MySQL配置文件的[mysqld]段添加server_id223344 log_binmysql-bin binlog_formatROW binlog_row_imageFULL gtid_modeON enforce_gtid_consistencyON修改完重启MySQL生效。注意binlog开启后对写入性能有一点影响但通常不到10%对线上业务基本无感这个代价换来的是所有数据变化的可追溯性值。5.2 快照阶段锁表导致线上业务受影响我见过一些人用Debezium做全量快照时发现线上个别写入事务被阻塞了。原因在于initial快照在获取一致性位点时可能会用FLUSH TABLE WITH READ LOCK这个语句会获取全局读锁如果表数据量很大锁持有时间会很长。解决办法有几个。在MySQL侧如果开启了GTIDDebezium可以避免使用全局读锁而是通过一致的快照读来完成初始一致性位点获取和全量数据读取。此外Debezium还提供一个参数snapshot.locking.mode可选值包括minimal等minimal模式会在极短时间内加锁比默认模式锁时间短很多。如果你的MySQL版本支持GTID且线上不能接受长时间锁定优先开启GTID snapshot.locking.modeminimal。另外还有一个思路是把快照任务放到业务低峰期执行或者直接用schema_only模式跳过存量数据导入让下游通过其他ETL手段先灌历史数据。5.3 数据能查到但增量一直不更新offset落点怎么排查这是实际排查中频率很高的问题注册connector成功全量快照也成功binlog里的新的增删改却迟迟不出现在Kafka topic里。遇到这种情况先别急着重启connect按照优先级来排查。第一步看connector状态curl http://localhost:8083/connectors/mysql-appdb-connector/status。如果state是RUNNING再往下看second。第二步查connect最新日志尤其是有没有task抛出数据库连接超时或者binlog position不存在的异常。第三步检查Kafka topic的新增消息数kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic appdb-server.appdb.user --time -1这个命令能获得该topic的logEndOffset如果offset一直不涨说明connector确实没有读到新数据。常见的隐蔽原因有两类一类是binlog过期清理导致位点失效connector在重启恢复时尝试从已删除的binlog文件读取自然会卡住另一类是connector任务的group.id配置变了或者Kafka Connect的offset topic被误删导致任务状态丢失重新启动时又从旧位点开始或报错。我自己的排查习惯是先在MySQL里手动执行几条INSERT/UPDATE然后立刻看Kafka topic消费位点如果位点不动立刻去connect日志里搜ERROR或WARN关键词。日志里通常会告诉你binlog文件和position再回到MySQL对比SHOW MASTER STATUS两者一对照就知道差距在哪。5.4 DDL变更引发的结构解析异常CDC过程中文件表结构变了怎么办这是个躲不开的话题。MySQL里执行ALTER TABLE ADD COLUMN/RENAME/DROP后Debezium会通过schema history topic感知到schema变更然后更新它维护的表结构信息。但在DDL变更的瞬间如果CDC链路异常中断或者schema history topic数据被清理重启后Debezium可能没有足够的历史信息来解析旧binlog消息导致SourceRecord反序列化失败。经验法则是生产环境变更表结构时尽量选在业务低峰期并且变更完成后立即抽查CDC链路的数据是否正常。如果DDL刚执行完就出现了解析失败可以检查schema-changes.appdb这个history topic确认里面是否记录了最新的表结构变更事件如果history topic被清掉了最省事的方法是从当前时间重新做一次全量快照让Debezium重建schema信息。5.5 消费积压严重从单线程到并行多任务的优化当业务量大增Kafka topic里的消息消费速度低于生产速度积压会越来越严重。这首先说明你的消费端处理能力不足其次要检查connector端的读取并行度。Debezium 2.x对MySQL支持多个task配合增量快照并行读取前提是源表有主键或有效唯一键且connector配置里设置tasks.max大于1。这个特性在阅读级就能明显提升吞吐尤其是全量快照阶段多task并行分片读取的效果立竿见影。不过要注意tasks.max不是越大越好。每个task都会建立独立的binlog连接和快照通道对源库和Kafka Connect节点的资源消耗都会线性增加。我在生产上一般先设置tasks.max4观察CPU、内存和topic lag再决定是否往上调。还有一点binlog的增量读取阶段受制于MySQL单连接解析tasks.max对增量阶段的提升不如快照阶段显著所以大流量的生产环境往往还要配合表拆分成多个connector的方式把热点表单独分发。5.6 常见问题速查表问题现象常见原因排查和解决方案Connector启动失败binlog_format不是ROW、账号权限不足确认binlog_format/snapshot参数复核CDC账号权限快照阶段业务事务被阻塞全局读锁持锁时间过长开启GTID设置snapshot.locking.modeminimal增量数据一直不更新binlog过期清理、offset topic异常检查binlog保留周期检查Kafka Connect状态和offset topicDDL后解析失败schema history topic丢失结构变更记录检查schema history topic必要时重建快照消费积压、吞吐不够单task读取、消费端写目标太慢tasks.max调大快照分片下游优化写入并发下游出现重复数据消费重复消息、未做幂等所有目标写入改为upsert/覆写模式删除用tombstone6. 实操经验总结与个人建议说了这么多方案和代码最后聊两句个人经验。CDC这套东西我经历了从“工具好用”到“链路可靠”的认知变化。最初的关注点全在工具本身比如Canal用哪个版本、Debezium默认配置怎么改后来发现真正决定CDC成败的其实是链路周边的那堆非工具问题binlog保留期、下游幂等、schema变更流程、监控告警、位点管理任何一个环节出问题都会让整个链路的水位崩塌。我建议任何打算把CDC用在生产的朋友开工之前先想清楚三件事第一源库要不要长期开启binlog这涉及DBA运维和磁盘配额第二下游目标系统用什么幂等写入策略千万别觉得重复消息无所谓第三监控告警怎么做至少要对topic lag和connector状态打上监控设置阈值告警。这三件事做好了在线运行阶段会少很多半夜被叫起来的场景。还有一个容易被忽略的小技巧给每个topic的消费端务必开启消费者组并且把消费位点提交做成“先处理后提交”还是“先提交后处理”要明确自己测试过再定。我之前在同步到Elasticsearch时因为采用先提交后处理偶发情况下消息处理失败但位点已经提交导致重启后丢数据。改成先处理后提交之后虽然有一点点重复消费风险但配合幂等写入整体更稳。哪个优先级更高取决于你丢数据敏感还是重复数据敏感。如果你也是刚开始尝试CDC不要一上来就追求把所有表都监听了。我强烈建议先把一张核心业务表打通全流程验证好位点恢复、幂等消费、DDL变更这些关键场景再逐步扩容到全部表。这样每走一步都心里有底出了问题也容易定位。这套方法论我个人用了很多个项目体感一直很稳。
CDC实时监听数据库全解析：从原理到Debezium+Kafka实战
CDCChange Data Capture变更数据捕获是这几年来数据库圈子里绕不开的话题。尤其当业务系统需要实时感知数据库里的增删改查把数据同步到消息队列、数仓、缓存或者另一个数据库时CDC几乎成了标准答案。我最早接触这个场景是被迫的——当时要做一个跨库的实时数据同步既要保证低延迟又不能让业务库承担太多压力试了一圈轮询、触发器方案都不满意最后把目光落在了基于日志的CDC上。这篇文章就把我这几年做CDC实时监听数据库的方案选型、实操踩坑、参数调优一次性讲清楚项目标题很简单但背后能展开的东西非常多。这篇内容适合谁看一类是刚开始接触数据同步、实时数仓的同学想知道CDC到底是什么、跟普通的定时捞数据有什么区别另一类是已经在用CDC工具但被各种细节折磨过的开发比如binlog格式不对导致解析失败、快照阶段锁表、消费积压追不上这篇文章里的排查思路大概率能帮到你。1. 内容整体设计与思路拆解1.1 先搞清楚CDC的本质是什么很多人一听到“实时监听数据库”第一反应是写个定时任务每秒钟查一次表把变化的数据捞出来。这种做法也能叫“监听”但不是真正的实时而且有几个很难绕过去的坑轮询间隔越小对数据库压力越大高频轮询对线上业务库的查询开销不可忽视如果表没有自增ID或更新时间字段你根本没法判断哪些行是新增、哪些是修改物理删除的数据轮询根本发现不了除非走逻辑删除的歪路。CDC的核心思路完全不一样它不主动去问数据库“你有没有变化”而是让数据库自己把变化“说出来”。实现方式有几种但最主流、最可靠的是基于数据库的事务日志比如MySQL的binlog、PostgreSQL的WAL、SQL Server的transaction log。数据库每执行一次增删改操作都会写一条日志记录CDC工具解析这些日志就知道哪张表、哪一行、在什么时间、做了什么操作。数据库本身就用这套日志做主从复制和崩溃恢复CDC相当于是把主从复制的思路挪到了数据集成场景本质上就是“让另一个消费者去读复制日志”。这个思路带来的好处是实实在在的对源库几乎没有额外压力因为读的是日志不是表数据天然支持删除和更新操作不需要业务表任何配合延迟能做到毫秒或秒级取决于下游消费链路。1.2 为什么说这是数据架构里的“稳定支点”我后来把CDC当成一个独立的基础设施层来看待而不是某个工具或一段脚本。意思是说数据库变更数据是业务系统的真相来源CDC负责把这个真相高效地、无侵入地分发出去下游想怎么用都行。数据同步到数仓CDC替代了原先的凌晨ETL批处理数据同步到缓存CDC替代了手工清缓存和双写的一致性难题CDC加上消息队列还能支撑事件驱动架构——比如订单表有变化就产生一个领域事件触发后续的库存扣减、通知推送。我在这几年的实操里发现团队对数据一致性的焦虑大部分都能靠CDC缓解。举个例子很多项目在双写缓存时都会纠结先更新数据库还是先写缓存、失败了怎么办一旦引入CDC监听数据库变更并同步到缓存业务侧只需要操作数据库缓存更新交给CDC链路去处理少了一大堆业务代码里的一致性逻辑。当然这不是万能药CDC链路本身也有延迟和数据丢失的可能但至少比业务侧手工双写要干净得多。1.3 一个完整的CDC链路长什么样为了后面讲实操不绕弯先给一个整体框架。一个典型的CDC实时监听数据库架构通常包含四个环节源数据库、CDC采集端、消息中间件、消费端。源数据库很好理解就是那个产生变更数据的库。CDC采集端负责连接源库、读取binlog以MySQL为例、解析成结构化的变更事件再序列化发送出去。市面上常见的采集端有Debezium、Canal、Flink CDC等每一种的定位和适用场景都有差别我会在下一章详细对比。消息中间件的作用有两个一是缓冲和解耦采集端把数据持续写入消息队列消费端按自己的速度去消费不会因为下游业务处理慢而拖垮采集端二是让CDC链路具备多播能力一套变更数据可以被多个下游同时消费比如既同步到Elasticsearch又同步到数据仓库。消费端就是各类目标系统收到变更事件后执行相应的写入逻辑。我在很多文章里看到有人把Debezium直接往Kafka里塞其实这是Debezium一个非常典型的部署形态Kafka在中间扮演的正是消息缓冲和分发角色。后面我也会基于这个组合给出一份能直接跑的实操配置。2. 方案选型市面上主流的CDC实现方式对比2.1 基于日志解析的CDC工具一劳永逸的主流选择基于日志的方案MySQL这边最常用的是Canal和Debezium两者都伪装成MySQL的从库去拉binlog核心机制相同但生态定位不同。Canal是阿里开源的项目很多早期做大数据同步的公司都靠它支撑社区成熟度很高Canal适配的消息通道包括Kafka、RocketMQ等运维上主要是通过Canal Admin做集群管理。Debezium是Red Hat主导的开源项目基于Kafka Connect框架构建最大好处是天然融入Kafka生态source connector、sink connector统一管理而且支持MySQL、PostgreSQL、SQL Server、Oracle、MongoDB等一大票数据库一个框架走天下。我个人的经验是如果你的基础设施已经围绕Kafka展开团队也熟悉Kafka Connect优先选Debezium因为它有统一的connector生命周期管理、资源配置和日志体系出问题了排查路径也比较清晰。如果就是想快速把MySQL的数据同步到另一个MySQL或者RocketMQ团队也不想引入Kafka Connect这套东西Canal更轻量一些直接部署一个Java进程配置好canal.properties和instance.properties就能跑起来。Flink CDC是后来居上的热门选择它最大卖点是结合Flink的流式计算能力能直接做“数据库入湖入仓”的实时ETL。Flink CDC对无主键表、DDL变更、全量增量自动切换的处理也很成熟而且Flink CDC 2.x支持了增量快照和多并行读取吞吐上限远高于传统单线程单连接的模式。不过它的部署复杂度也更高需要一套Flink集群适合本身就在做Flink实时计算的项目如果只是单纯做数据变更订阅用Debezium或Canal更轻。2.2 基于轮询与触发器的旁路方案能用但尽量别碰考虑到不少读者是新手我还是介绍一下轮询和触发器这两种方案然后说清楚为什么它们只适合小规模场景。轮询方案就是开定时任务周期性执行对更新时间字段的SELECT查询比如WHERE update_time 上次记录的游标值。配置门槛极低任何懂SQL的人上手就能做数据同步延迟取决于轮询间隔最坏情况下会差一个周期。但问题在于物理删除无法感知除非业务层改成逻辑删除高频轮询对数据库主库会产生额外开销线上高并发场景容易受影响而且每次轮询都需要带上一次游标当数据量超过千万级时查询本身也会变慢。触发器方案就是在业务库上建AFTER INSERT/UPDATE/DELETE触发器把变更记录写进一张日志表应用再去日志表里取数据。这个方案确实能捕获所有变更包括物理删除但它属于侵入式方案要给业务库额外建表、建触发器相当于给本就繁忙的业务数据库又加了一层负担而且触发器里写日志通常是同步事务的一部分一旦日志表写入慢业务事务也会被拖慢。最要命的是数据库迁移、DDL变更时触发器极易出问题我已经不止一次见过因为触发器没同步迁移导致数据漏采集的事故。所以我的建议是触发器方案只适合没有二进制日志可用的老旧数据库做临时补救生产环境尽量别长期依赖。2.3 不同方案的选型对照与适用场景方案延迟级别侵入性全量增量支持数据库类型覆盖适用场景基于日志Debezium/Canal/Flink CDC毫秒~秒级几乎无侵入好广MySQL/PG/Oracle/SQL Server等生产环境首选实时数仓、异构同步轮询UPDATE_TIME秒~分钟级需业务表有更新字段弱任意数据库数据量小、对实时性要求低触发器写日志表毫秒~秒级强侵入需建触发器和日志表一般支持的数据库均可无binlog可用时的临时方案数据库原生复制秒级无侵入不支持异构原生同构数据库的主从复制我自己的选型逻辑很简单第一步看源数据库有没有可用的日志机制MySQL就把binlog打开PostgreSQL就调wal_levellogical只要数据库本身支持逻辑复制直接选基于日志的方案第二步看团队已有技术栈如果已经在用Flink就顺着Flink CDC用否则Kafka生态选Debezium轻量场景选Canal第三步才考虑业务表能不能加更新字段、能不能接受逻辑删除这类妥协方案。3. 实操用Debezium Kafka实现MySQL实时监听3.1 环境准备与版本选择实操部分我用的是Docker Compose方式搭建Debezium Kafka环境这是最省事的部署方式适合开发调试也适合快速验证方案可行性。如果你是在生产环境建议把Zookeeper、Kafka、Kafka Connect分别部署成集群但这套搭建思路跟开发环境没有本质区别。我自己测试时使用的版本组合是Debezium2.3.x该版本之前需要Java 11新版本基于Java 17注意JRE版本Kafka Kafka Connect3.4.xMySQL8.0.x开启binlog消息格式JSON或者Confluent Avro开发阶段用JSON最简单生产推荐Avro带Schema RegistryDocker Compose的基础编排文件大致长这样version: 3.8 services: mysql: image: mysql:8.0 container_name: cdc-mysql environment: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: appdb command: [ --server-id223344, --log-binmysql-bin, --binlog-formatROW, --binlog-row-imageFULL, --gtid-modeON, --enforce-gtid-consistencyON ] ports: - 3306:3306 zookeeper: image: confluentinc/cp-zookeeper:latest container_name: cdc-zookeeper environment: ZOOKEEPER_CLIENT_PORT: 2181 kafka: image: confluentinc/cp-kafka:latest container_name: cdc-kafka ports: - 9092:9092 environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 depends_on: - zookeeper connect: image: debezium/connect:2.3 container_name: cdc-connect ports: - 8083:8083 environment: BOOTSTRAP_SERVERS: kafka:9092 GROUP_ID: 1 CONFIG_STORAGE_TOPIC: my_connect_configs OFFSET_STORAGE_TOPIC: my_connect_offsets STATUS_STORAGE_TOPIC: my_connect_statuses depends_on: - kafka - mysql这里有几个细节值得说一下。MySQL的server_id不能跟任何从节点冲突如果不设置可能默认成1在多实例场景下会导致两个节点争抢同一个server_idbinlog拉取会出现异常断开。binlog-format必须设置为ROW因为只有ROW格式的binlog才会记录每行变更前的全量字段值和变更后的值STATEMENT格式只记录SQL语句逻辑复制的解析效率和准确性都差很多。binlog-row-imageFULL是为了保证binlog里保存一行所有字段的镜像这样才能在下游重建出完整的数据行。3.2 注册MySQL Connector让Debezium开始干活环境起来之后需要在Kafka Connect里注册一个MySQL source connector。这一步是通过REST API完成的向Connect服务的/connectors端点提交一个JSON配置。我生产环境里反复调整过这份配置核心字段如下{ name: mysql-appdb-connector, config: { connector.class: io.debezium.connector.mysql.MySqlConnector, database.hostname: mysql, database.port: 3306, database.user: cdc_user, database.password: cdc_password, database.server.id: 223344, database.server.name: appdb-server, database.include.list: appdb, table.include.list: appdb.user,appdb.orders, database.history.kafka.topic: schema-changes.appdb, include.schema.changes: true, snapshot.mode: initial, key.converter: org.apache.kafka.connect.json.JsonConverter, value.converter: org.apache.kafka.connect.json.JsonConverter, key.converter.schemas.enable: false, value.converter.schemas.enable: false } }提交配置的命令很简单一个curl就搞定curl -X POST -H Content-Type: application/json -d mysql-connector.json http://localhost:8083/connectors这个配置里最重要的参数其实有三组。database.hostname和database.user配置的是Debezium用来连接MySQL的账号这个账号必须要有的权限包括SELECT、RELOAD、SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT。RELOAD权限是因为Debezium在进行初始快照时需要FLUSH TABLE WITH READ LOCK来获取一致性位点。REPLICATION相关权限是为了能够开始和停止binlog流。线上用root账号拉binlog是非常不建议的我在生产环境都是单独建一个最小权限的账号给CDC专用方便权限审计和回收。database.server.name是Kafka Topic命名的前缀。默认情况下Debezium监听的每个表都会生成一个topic命名规则是{database.server.name}.{database}.{table}比如上面配置里user表就会写到appdb-server.appdb.user这个topic。如果schema变化了还会写到schema-changes.appdb这个history topic里。这个命名规则很容易被忽略但排查消费端配置时非常关键后续我会在常见问题里再提。snapshot.mode参数控制Debezium启动时如何做初始快照。initial模式会在启动时先对当前全量数据做一次一致性快照然后无缝切换到binlog增量监听schema_only模式只捕获表结构不搬运存量数据适合下游已经有历史数据的场景when_needed模式则是先尝试纯增量发现需要快照时再补这个模式用得相对少never模式则完全不做快照只从binlog当前位点开始读但这样做有个前提就是下游明确知道自己要从哪个点开始接收。3.3 验证链路看数据是否真的实时流过来完整链路搭建完成后验证方式比较简单。用MySQL客户端往user表里插入数据INSERT INTO appdb.user (id, name, email) VALUES (1, 张三, zhangsanexample.com); UPDATE appdb.user SET name 李四 WHERE id 1; DELETE FROM appdb.user WHERE id 1;然后通过kafka-console-consumer消费对应的topickafka-console-consumer --bootstrap-server localhost:9092 --topic appdb-server.appdb.user --from-beginning正常情况下消费端会看到三条消息分别对应INSERT、UPDATE、DELETE三个操作。INSERT消息的payload结构大概是这样的{ before: null, after: { id: 1, name: 张三, email: zhangsanexample.com }, source: { version: 2.3.0.Final, connector: mysql, name: appdb-server, ts_ms: 1714095820000, snapshot: false, db: appdb, table: user, server_id: 223344, binlog_file: mysql-bin.000003, binlog_position: 245, gtid: abc-123-456 }, op: c, ts_ms: 1714095820123 }这里的op字段是操作类型标识c代表createINSERTu代表updated代表delete。Update操作和Delete操作的before字段里会带上被修改前的完整行数据这就是binlog-row-imageFULL的结果。这条链路验证通过意味着MySQL的每一次增删改查准确的说是INSERT/UPDATE/DELETE都会在毫秒级被Debezium解析成结构化JSON写入Kafka对应topic下游系统只需要订阅这些topic就能获得全量实时的数据变更流。3.4 全量快照与增量无缝衔接的机制很多刚接触CDC的人都会有一个疑问如果目标库里是空的我需要把源库已有的几千万条历史数据也同步过来该怎么做Debezium的initial快照模式就是干这个的。具体流程是Connector启动时先开启一个数据库连接获取当前binlog的位点即一个一致性坐标然后在这个位点上建立一致性快照。快照期间源库的增量变更会继续写binlog不会被阻塞。快照读取完所有存量数据后Debezium会从之前记录的binlog位点继续消费增量这样全量数据和期间产生的增量数据在逻辑上是无缝衔接的不会丢也不能重复。这里有个性能细节值得展开Debezium早期版本做全量快照是单线程主键分页读取速度对大表来说确实比较感人。2.x版本引入了增量快照机制Incremental Snapshot可以把快照任务拆分成多段并行执行通过signal topic控制触发停止和恢复实测大表的快照速度提升非常明显。如果你表的数据量在百万级以上建议仔细看看官方文档里关于read.only的连接建议和snapshot.fetch.size参数的调优。snapshot.fetch.size控制每次快照批次的读取行数默认值通常是1024。调大这个值可以减少往返数据库的次数从而提升速度但会增加内存占用和单次事务时间生产上需要根据表宽和网络延迟测试调整。4. 核心细节解析与实操要点4.1 binlog格式与关键参数监听前的第一道坎如果说CDC监听MySQL是座大坝binlog就是坝体的地基。地基没打好后面解析什么都白搭。我在做实战辅导时发现很多人的CDC链路连不通往上排查半天最后都是binlog配置的问题。所以把这块单独拎出来逐条讲讲关键参数的选值逻辑。先说binlog_format这是一个最核心的参数。MySQL支持STATEMENT、ROW、MIXED三种格式。STATEMENT格式记录的是SQL原文占用空间小但无法精确还原每行数据的变化ROW格式记录的是每行变更前后的镜像数据信息量最大MIXED是MySQL自己根据语句类型决定用STATEMENT还是ROW记录。做CDC一定要用ROW这是Debezium和Canal的共同强制要求因为只有ROW格式才能提供足够的信息在异构目标端重建数据行。再说binlog_row_image这个参数会影响ROW格式下记录的字段内容可以设为FULL、MINIMAL、NOBLOB。FULL记录所有列的前后镜像值MINIMAL只记录能够唯一标识这一行的列通常是主键和发生变更的列NOBLOB则是FULL的一个变体大字段不记录。做CDC强烈建议用FULL你自己想像一下假如下游要拿变更数据重建Elasticsearch的文档需要这一行的完整字段MINIMAL模式下某些没变更过的字段根本不在binlog里重建出的JSON就是不完整的。NOBLOB模式更坑一旦表里有TEXT/BLOB类型字段数据在binlog里只出现一个占位符号下游只能拿到一个残破的行。还有一个容易被忽略的参数是expire_logs_days或binlog_expire_logs_seconds它控制binlog文件的保留时长。CDC消费者不是实时在线的假如你停掉connector去维护几天后重启如果binlog文件已经被清理Debezium就会报找不到位点。生产经验是保留至少72小时以上binlog我一般设置binlog_expire_logs_seconds604800即7天。运维上还需要定期检查binlog文件占用的磁盘空间防止过大撑爆存储。4.2 主键与无主键表的处理原则CDC监听过程中有一类场景特别容易让人翻车源表没有主键。Debezium默认情况下要求表必须有主键因为主键有双重作用一是作为快照时的分页游标二是作为Kafka消息key决定同一行数据在topic中的分区策略。对于无主键的表Debezium会在启动快照时报错。解决方案有两条路一是给源表加上主键这对有严格规范的库来说是最好的做法二是利用message.key.columns参数指定一个能唯一标识行的候选列组合比如唯一索引键或业务订单号。但要注意如果基于业务列做消息key而业务列本身后来又发生变化比如订单号修改那么同一行在Kafka里会从分区A跳到分区B下游如果按分区维护状态就会出问题。还有一个我没踩过但要提醒的点无主键且无唯一索引的表如果又没有合适业务列做消息keyDebezium会退化到把所有数据写入同一个分区这会让数据倾斜非常严重生产环境的大表和热点数据场景下Kafka某个分区会被写爆而其他分区闲置。遇到这种情况建议跟业务方坐下来把主键梳理清楚或者由上游应用在写入时生成一个不可变的逻辑ID字段。4.3 Kafka Topic与消息结构解析前面提过Topic命名规则是database.server.name.database.table还有一项容易被忽略的是Debezium还允许你用topic.prefix配置来改变这个前缀命名。比如在Debezium 2.x中官方把database.server.name改名为topic.prefix。我因为升级版本遇到过老配置失效的问题新版本的connector启动直接报错说找不到配置字段这一点在阅读社区文档时一定要注意版本差异。Kafka消息里除了业务数据还有大量的元数据字段比如source块中的binlog_file和binlog_position。这些信息非常有用比如可以精确到某条变更来自哪个binlog文件、哪个位置做故障恢复和位点对比时价值极大。还有op字段c/r/u/d分别代表create、read快照阶段、update、delete。你在写下游消费程序时不要只处理c/u/d快照阶段的r类型是全量数据开始的标记漏处理的话初始同步会丢数据。快照阶段的r消息和增量阶段的c消息从数据结构角度几乎一样唯一的区别是source.snapshot字段值。所以处理下游逻辑时最好统一按after字段的最终值去做幂等upsert对r/c/u用相同逻辑写入目标对d执行删除。这套逻辑是我在实践中摸索出来的通用模式适配大部分同步场景。4.4 数据一致性至少一次与幂等消费的正确理解CDC链路里“至少一次”投递语义是绕不开的话题。Debezium通过记录Kafka的offset和MySQL的binlog位置来保证恢复后不丢数据但故障恢复时可能出现重复消息因为消费者提交offset到Kafka的时机和数据处理完成时机不可能做到完全原子。这意味着下游必须支持幂等重复消息来一遍结果应该跟来一遍一样。比如你同步到MySQL可以用REPLACE INTO或INSERT ... ON DUPLICATE KEY UPDATE同步到Elasticsearch直接对同一id做index操作就是天然的幂等同步到RedisSET操作天然幂等。遇到DELETE操作也建议用幂等方式比如先delete再insert或者干脆发一个tombstone标记消息让下游自行处理。我在真实项目中因为忽略了幂等出现过一次比较难受的事故当时同步到Elasticsearch时没有用upsert而是先查后插结果重复消息导致文件里写入重复数据搜索得分全乱了。从此以后我给自己立了一个规矩下游写入一律默认幂等除非你能证明目标系统天然幂等。这个经验同样可以推广到所有做实时数据同步的团队。5. 常见问题与排查技巧实录5.1 Connector启动就报错binlog配置没到位症状是提交connector后connect日志里不断重试错误信息指向“The database server is not configured to use row-based binary logging”之类。这个问题的原因字面上已经说了binlog_format没有设成ROW。我在Docker环境里看过很多次MySQL官方镜像默认binlog_format其实不是ROW必须通过command参数主动覆盖。排查和修复路径比较直接SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_row_image;如果binlog_format不是ROW需要在MySQL配置文件的[mysqld]段添加server_id223344 log_binmysql-bin binlog_formatROW binlog_row_imageFULL gtid_modeON enforce_gtid_consistencyON修改完重启MySQL生效。注意binlog开启后对写入性能有一点影响但通常不到10%对线上业务基本无感这个代价换来的是所有数据变化的可追溯性值。5.2 快照阶段锁表导致线上业务受影响我见过一些人用Debezium做全量快照时发现线上个别写入事务被阻塞了。原因在于initial快照在获取一致性位点时可能会用FLUSH TABLE WITH READ LOCK这个语句会获取全局读锁如果表数据量很大锁持有时间会很长。解决办法有几个。在MySQL侧如果开启了GTIDDebezium可以避免使用全局读锁而是通过一致的快照读来完成初始一致性位点获取和全量数据读取。此外Debezium还提供一个参数snapshot.locking.mode可选值包括minimal等minimal模式会在极短时间内加锁比默认模式锁时间短很多。如果你的MySQL版本支持GTID且线上不能接受长时间锁定优先开启GTID snapshot.locking.modeminimal。另外还有一个思路是把快照任务放到业务低峰期执行或者直接用schema_only模式跳过存量数据导入让下游通过其他ETL手段先灌历史数据。5.3 数据能查到但增量一直不更新offset落点怎么排查这是实际排查中频率很高的问题注册connector成功全量快照也成功binlog里的新的增删改却迟迟不出现在Kafka topic里。遇到这种情况先别急着重启connect按照优先级来排查。第一步看connector状态curl http://localhost:8083/connectors/mysql-appdb-connector/status。如果state是RUNNING再往下看second。第二步查connect最新日志尤其是有没有task抛出数据库连接超时或者binlog position不存在的异常。第三步检查Kafka topic的新增消息数kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic appdb-server.appdb.user --time -1这个命令能获得该topic的logEndOffset如果offset一直不涨说明connector确实没有读到新数据。常见的隐蔽原因有两类一类是binlog过期清理导致位点失效connector在重启恢复时尝试从已删除的binlog文件读取自然会卡住另一类是connector任务的group.id配置变了或者Kafka Connect的offset topic被误删导致任务状态丢失重新启动时又从旧位点开始或报错。我自己的排查习惯是先在MySQL里手动执行几条INSERT/UPDATE然后立刻看Kafka topic消费位点如果位点不动立刻去connect日志里搜ERROR或WARN关键词。日志里通常会告诉你binlog文件和position再回到MySQL对比SHOW MASTER STATUS两者一对照就知道差距在哪。5.4 DDL变更引发的结构解析异常CDC过程中文件表结构变了怎么办这是个躲不开的话题。MySQL里执行ALTER TABLE ADD COLUMN/RENAME/DROP后Debezium会通过schema history topic感知到schema变更然后更新它维护的表结构信息。但在DDL变更的瞬间如果CDC链路异常中断或者schema history topic数据被清理重启后Debezium可能没有足够的历史信息来解析旧binlog消息导致SourceRecord反序列化失败。经验法则是生产环境变更表结构时尽量选在业务低峰期并且变更完成后立即抽查CDC链路的数据是否正常。如果DDL刚执行完就出现了解析失败可以检查schema-changes.appdb这个history topic确认里面是否记录了最新的表结构变更事件如果history topic被清掉了最省事的方法是从当前时间重新做一次全量快照让Debezium重建schema信息。5.5 消费积压严重从单线程到并行多任务的优化当业务量大增Kafka topic里的消息消费速度低于生产速度积压会越来越严重。这首先说明你的消费端处理能力不足其次要检查connector端的读取并行度。Debezium 2.x对MySQL支持多个task配合增量快照并行读取前提是源表有主键或有效唯一键且connector配置里设置tasks.max大于1。这个特性在阅读级就能明显提升吞吐尤其是全量快照阶段多task并行分片读取的效果立竿见影。不过要注意tasks.max不是越大越好。每个task都会建立独立的binlog连接和快照通道对源库和Kafka Connect节点的资源消耗都会线性增加。我在生产上一般先设置tasks.max4观察CPU、内存和topic lag再决定是否往上调。还有一点binlog的增量读取阶段受制于MySQL单连接解析tasks.max对增量阶段的提升不如快照阶段显著所以大流量的生产环境往往还要配合表拆分成多个connector的方式把热点表单独分发。5.6 常见问题速查表问题现象常见原因排查和解决方案Connector启动失败binlog_format不是ROW、账号权限不足确认binlog_format/snapshot参数复核CDC账号权限快照阶段业务事务被阻塞全局读锁持锁时间过长开启GTID设置snapshot.locking.modeminimal增量数据一直不更新binlog过期清理、offset topic异常检查binlog保留周期检查Kafka Connect状态和offset topicDDL后解析失败schema history topic丢失结构变更记录检查schema history topic必要时重建快照消费积压、吞吐不够单task读取、消费端写目标太慢tasks.max调大快照分片下游优化写入并发下游出现重复数据消费重复消息、未做幂等所有目标写入改为upsert/覆写模式删除用tombstone6. 实操经验总结与个人建议说了这么多方案和代码最后聊两句个人经验。CDC这套东西我经历了从“工具好用”到“链路可靠”的认知变化。最初的关注点全在工具本身比如Canal用哪个版本、Debezium默认配置怎么改后来发现真正决定CDC成败的其实是链路周边的那堆非工具问题binlog保留期、下游幂等、schema变更流程、监控告警、位点管理任何一个环节出问题都会让整个链路的水位崩塌。我建议任何打算把CDC用在生产的朋友开工之前先想清楚三件事第一源库要不要长期开启binlog这涉及DBA运维和磁盘配额第二下游目标系统用什么幂等写入策略千万别觉得重复消息无所谓第三监控告警怎么做至少要对topic lag和connector状态打上监控设置阈值告警。这三件事做好了在线运行阶段会少很多半夜被叫起来的场景。还有一个容易被忽略的小技巧给每个topic的消费端务必开启消费者组并且把消费位点提交做成“先处理后提交”还是“先提交后处理”要明确自己测试过再定。我之前在同步到Elasticsearch时因为采用先提交后处理偶发情况下消息处理失败但位点已经提交导致重启后丢数据。改成先处理后提交之后虽然有一点点重复消费风险但配合幂等写入整体更稳。哪个优先级更高取决于你丢数据敏感还是重复数据敏感。如果你也是刚开始尝试CDC不要一上来就追求把所有表都监听了。我强烈建议先把一张核心业务表打通全流程验证好位点恢复、幂等消费、DDL变更这些关键场景再逐步扩容到全部表。这样每走一步都心里有底出了问题也容易定位。这套方法论我个人用了很多个项目体感一直很稳。
