做数据同步这行的都清楚，业务库的每一次增删改，背后往往牵着一串下游动作：缓存要失效、搜索索引要重建、报表要刷新、风控要记账。早期大家习惯在业务代码里埋点，写库之后顺手发个消息出去，简单直接，但埋着埋着就发现代码里到处是消息发送逻辑，业务和同步彻底耦合在一起，改一个字段要动三四个地方，谁都不敢轻易碰。后来 binlog 订阅这套思路成熟起来，把"变更捕获"这件事从业务代码里彻底剥离，交给独立的中间件去做，业务只管写库，下游爱怎么消费怎么消费。Canal 就是这类工具里落地最广的一个，它伪装成 MySQL 的从库，订阅 binlog，把行级变更解析成结构化数据，再推给下游。这篇要聊的，就是把 Canal 抓到的实时操作日志，稳稳当当地送进 RabbitMQ，再在消费端做监听和解析处理的完整链路。适合正在做异构数据同步、缓存一致性、审计日志、实时数仓入门的同学参考，不管你是刚接触 Canal 的新手，还是已经跑起来但被各种坑折腾过的老手，下面这些内容应该都能对上你的某些场景。
1. 先想清楚这条链路到底在解决什么问题
很多人一上来就问 Canal 怎么装、RabbitMQ 怎么配，配置抄一堆，跑起来发现数据对不上，回头排查半天。问题往往不在配置本身，而在于没想明白这条链路的设计意图。我见过不少项目，明明只是想让缓存跟着数据库变，结果搭了一套全量 binlog 订阅，把所有库所有表的所有变更都往消息队列里灌，下游消费端天天被无关消息淹没，最后不得不加一堆过滤逻辑，反而更乱。所以在动手之前，把"为什么是 Canal + RabbitMQ 这个组合"想透，比抄配置重要得多。
1.1 业务代码埋点、定时扫表、binlog 订阅三条路的取舍
做数据同步，主流就三条路，各有各的脾气。业务代码埋点是最好理解的，写库成功后直接调消息发送，实时性最高，逻辑也最直白。但它的致命伤是耦合：同步逻辑散落在各个 service 里，一旦要加个下游或者改个消息格式，得把相关业务代码翻个底朝天。更麻烦的是事务问题，写库和发消息不在一个事务里，写库成功、发消息失败，数据就丢了；反过来发消息成功、写库回滚，下游就收到一条根本不存在的变更。要处理好得引入本地消息表或者事务消息，复杂度陡增。
定时扫表是另一条路，拿个时间戳字段或者自增 ID 做游标，每隔几秒扫一批变更出来。实现简单，对业务零侵入，但实时性差，扫表频率高了给数据库压力大，频率低了延迟明显。而且它只能捕获"最终状态"，中间过程全丢了，比如一个字段从 A 改到 B 再改到 C，扫表只能看到 C，做审计或者需要还原变更序列的场景直接歇菜。删除操作也麻烦，物理删除扫表根本发现不了。
binlog 订阅则是从数据库的变更日志入手，MySQL 主库把每一次行级变更都写进 binlog，订阅方伪装成从库去拉取，拿到的是最原始、最完整的变更流，增删改一个不漏，还能拿到变更前后的完整行数据。它对业务代码零侵入，实时性也好，主库写完 binlog 基本就能推出来。代价是需要开启 binlog 并设置成 ROW 格式，还要给订阅账号授复制权限，运维上多一层要求。三条路的对比如下：
| 方案 | 实时性 | 业务侵入 | 变更完整性 | 运维成本 | 适用场景 |
|---|---|---|---|---|---|
| 业务埋点 | 最高 | 强耦合 | 完整但易漏 | 低 | 少量核心表、强一致要求 |
| 定时扫表 | 秒级到分钟级 | 零侵入 | 仅最终状态 | 低 | 对实时性要求不高的报表 |
| binlog 订阅 | 准实时 | 零侵入 | 完整含前后镜像 | 中 | 缓存同步、审计、实时数仓 |
选 binlog 订阅这条路，本质是用一点运维成本换来了业务解耦和变更完整性，对于中大型系统来说这笔账很划算。
1.2 为什么中间要加一层 RabbitMQ，而不是 Canal 直连消费端
Canal 本身支持直接推给客户端，客户端连上去拉数据就行，那为什么还要在中间塞一个 RabbitMQ？这个问题我被问过很多次。直接连 Canal 的问题在于，Canal 的客户端连接是长连接拉取模式，一个 instance 能挂的客户端数量有限，而且客户端一旦挂了或者处理慢了，会直接影响 Canal 的投递节奏，严重的还会导致 Canal 端积压甚至阻塞。换句话说，消费端的稳定性会反向传导到采集端，这在生产环境里是很危险的耦合。
加上 RabbitMQ 之后，Canal 只负责把变更投递到队列，投完就算完事，消费端有几个、处理快慢、挂没挂，Canal 都不关心。队列本身起到了削峰填谷和缓冲的作用，下游某个消费者临时故障，消息在队列里堆着，恢复后继续消费，不影响采集链路。更关键的是，RabbitMQ 的 exchange、routing key 机制天然适合做消息分发，不同下游订阅不同的 routing key，各取所需，不用在消费端做大量过滤。当然代价是多了一个中间件的运维，消息也可能重复投递，消费端必须做幂等，这个后面会专门讲。
1.3 这条链路适合什么样的数据规模和一致性要求
说句实在话，Canal + RabbitMQ 这套组合不是万能的。它适合的是"准实时、最终一致"的场景，比如缓存失效、搜索索引更新、审计日志落库、轻量级实时数仓。如果你的业务要求强一致，比如扣款后余额必须立刻准确，那还是老老实实在同一个事务里处理，别指望异步链路。数据规模上，单实例 Canal 处理每秒几千到上万条变更问题不大，再往上要考虑分库分表后的多实例部署。消息体大小也要注意，大字段（比如长文本、blob）会让 binlog 和消息都膨胀，必要时在 Canal 端就过滤掉不需要的列，只同步关键字段，这个配置后面会讲。
2. 环境准备里那些容易翻车的细节
环境准备看着是最没技术含量的部分，但实际项目里，十次联调有六七次卡在这一步。MySQL 的 binlog 没开对、账号权限给少了、Canal 的 instance 配置和实际库对不上，这些问题报错信息往往很含糊，排查起来费时费力。我把自己踩过的和帮别人排查过的坑整理一下，按顺序过一遍，能省掉大量来回折腾的时间。
2.1 MySQL 侧：binlog 格式、server_id 与账号权限三件套
Canal 能工作的前提是 MySQL 开启了 binlog 且格式为 ROW。ROW 格式记录的是每一行数据变更前后的完整镜像，Canal 解析起来才有意义；STATEMENT 格式记录的是 SQL 语句，MIXED 是两者混合，这两种 Canal 都没法可靠地还原行级变更。检查当前配置：
SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'server_id';log_bin必须是 ON，binlog_format必须是 ROW。如果没开，得改配置文件（my.cnf 或 my.ini）里的log-bin、binlog-format=ROW，然后重启 MySQL。这里有个坑：binlog_format在部分版本里是可以在线改的（SET GLOBAL binlog_format='ROW'），但只对新连接生效，而且重启后会丢，所以务必写进配置文件。
server_id这个参数特别容易被忽略。Canal 伪装成从库，从库必须有唯一的 server_id，如果 Canal 配置的 slaveId 和 MySQL 集群里其他从库冲突，复制会直接报错。一般给 Canal 分配一个明显区别于业务从库的号段，比如业务从库用 1-100，Canal 就用 1000 以上。
账号权限方面，Canal 需要复制权限和读取权限。建账号的语句大概是这样：
CREATE USER 'canal_user'@'%' IDENTIFIED BY 'your_password'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal_user'@'%'; FLUSH PRIVILEGES;REPLICATION SLAVE是拉 binlog 必需的，REPLICATION CLIENT用于查看主库状态，SELECT用于初次启动时获取表结构。少给任何一个都会在启动时报权限相关的错，而且报错信息不一定直白，所以一次给全最省事。
2.2 Canal 服务端：instance 配置与过滤规则怎么写才不误伤
Canal 服务端（canal-server）的核心是 instance 配置，一个 instance 对应一个 MySQL 实例的订阅。配置文件在conf/example/instance.properties，几个关键项：
canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=canal_user canal.instance.dbPassword=your_password canal.instance.filter.regex=myapp\\.order,myapp\\.user canal.instance.filter.black.regex=filter.regex是过滤规则，用正则匹配库名.表名，多个用逗号分隔。这里最常见的坑是转义：在 properties 文件里，反斜杠本身是转义符，所以点号要写成\\.，实际传给 Canal 的是\.。如果写成单个\.，properties 解析后变成.，正则里.匹配任意字符，结果就是匹配了所有表，过滤形同虚设。我见过有人图省事直接写.*\\..*，那确实是全库全表，但下游就得自己扛过滤压力了。
另一个容易忽略的是canal.instance.filter.black.regex，黑名单优先级高于白名单，用来排除某些表特别方便，比如排除掉日志表、临时表。建议白名单精确到表，黑名单兜底排除明显不需要的，两层过滤下来，投递到 MQ 的消息量能降不少。
2.3 RabbitMQ 侧：exchange 类型选择与队列绑定策略
RabbitMQ 这边，exchange 类型的选择直接影响后续的扩展性。常见的有 direct、topic、fanout 三种。direct 按 routing key 精确匹配，简单但扩展性差，加一个下游就得改绑定；fanout 广播给所有绑定队列，适合所有下游都要全量消息的场景；topic 支持通配符匹配，*匹配一个单词，#匹配零个或多个单词，最适合按库表维度分发。
我的建议是用 topic exchange，routing key 设计成库名.表名.操作类型这样的三段式，比如myapp.order.insert、myapp.user.update。下游要订阅订单表所有操作，绑定myapp.order.*；只订阅插入，绑定myapp.order.insert；订阅整个库，绑定myapp.#。这种设计下，新增下游只需要加一个队列和绑定，完全不用动生产端，扩展性非常好。
队列声明时要注意持久化：exchange 和 queue 都设 durable=true，消息投递时设 deliveryMode=2（持久化），这样 RabbitMQ 重启后消息不丢。但要注意，持久化会带来一定的性能开销，如果业务能接受少量丢失，可以只持久化队列不持久化消息，这个取舍看具体场景。
3. Canal 到 RabbitMQ 的桥接：自己写还是用现成的
Canal 官方提供了几种投递方式，最常用的是 canal-adapter 和 canal-client 自己写。canal-adapter 是官方封装的适配器，支持把数据同步到 ES、HBase、RDB 等目标，也支持投递到 RabbitMQ、RocketMQ、Kafka。它的好处是配置化，不用写代码，改改 yml 就能跑；坏处是灵活性受限，消息格式、routing key 规则都是固定的，遇到特殊需求不好改。canal-client 则是拿 Java 客户端自己写消费逻辑，灵活度最高，但要自己处理连接管理、异常重试、消息格式转换。
3.1 canal-adapter 的配置化路线与它的能力边界
如果需求比较标准，就是想把变更按库表投递到 MQ，canal-adapter 是最省事的。它的 RabbitMQ 投递配置大概长这样（放在conf/rabbitmq/mq.yml）：
dataSourceKey: defaultDS destination: example groupId: g1 rabbitmq: host: 127.0.0.1 port: 5672 virtualHost: / username: guest password: guest exchange: canal.exchange exchangeType: topic routingKey: ${database}.${table}.${type}routingKey支持占位符，${database}、${table}、${type}会被替换成实际的库名、表名、操作类型（INSERT/UPDATE/DELETE）。这个设计刚好对上前面说的三段式 routing key，很顺手。消息体是 Canal 的 flatMessage 格式，包含变更前后的字段值、主键、执行时间等。
但 canal-adapter 的边界也很明显：消息体格式固定，想加自定义字段、想做字段脱敏、想按业务规则改写 routing key，都得改源码或者加一层转换。而且它对异常的处理比较粗放，投递失败的重试策略不够灵活。所以我的经验是，需求简单、追求快速上线，用 adapter；需求复杂、要精细控制，用 client 自己写。
3.2 用 canal-client 自建投递端的核心代码骨架
自建投递端的核心是继承AbstractCanalClient或者直接用SimpleCanalConnector，拿到 Message 后转成需要的格式再发 MQ。骨架大概是这样：
public class CanalToMqClient { public static void main(String[] args) throws Exception { CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "", ""); connector.connect(); connector.subscribe("myapp\\..*"); connector.rollback(); while (true) { Message message = connector.getWithoutAck(1000); long batchId = message.getId(); if (batchId == -1 || message.getEntries().isEmpty()) { Thread.sleep(100); continue; } try { for (Entry entry : message.getEntries()) { if (entry.getEntryType() == EntryType.ROWDATA) { handleRowData(entry); } } connector.ack(batchId); } catch (Exception e) { connector.rollback(batchId); } } } }这里有几个关键点。getWithoutAck是拉取但不确认，处理成功后再ack，失败则rollback，这样能保证至少一次投递。subscribe的过滤规则和 Canal 服务端的 filter 是叠加的，两边都配能减少无效传输。handleRowData里要把RowChange解析出来，拿到EventType（INSERT/UPDATE/DELETE）和每行的Column列表，组装成消息体。
3.3 消息体设计：保留前后镜像还是只发变更字段
消息体设计是个容易被低估的决策点。Canal 的 RowData 里，UPDATE 事件同时包含 beforeColumns 和 afterColumns，INSERT 只有 afterColumns，DELETE 只有 beforeColumns。全量发过去，消息体会比较大，但下游能拿到完整信息，做审计、做对比都方便。只发变更字段（afterColumns 里updated=true的列），消息体小，但下游拿不到完整行，做全量覆盖更新时还得回查数据库，反而增加负担。
我的建议是：默认发完整 afterColumns，同时把主键单独拎出来放在消息头或者固定字段里，方便下游做幂等和路由。对于大字段（text、blob），在 Canal 端就通过 filter 或者自定义逻辑裁掉，别让它们进消息体。如果确实需要 before 镜像（比如做变更对比、审计），再额外带上，但要评估消息体积。一个折中的做法是消息体里带 afterColumns 全量 + 变更字段列表（只记字段名），下游需要对比时按字段名去取，兼顾了体积和信息量。
4. 消费端的监听、解析与幂等处理
消息进了 RabbitMQ，消费端的工作才真正开始。监听、反序列化、按操作类型分发、幂等、异常重试，每一环都有讲究。这一块写得好不好，直接决定了整条链路在生产环境能不能扛住。
4.1 监听容器的并发、预取与手动 ack 配置
Spring AMQP 的@RabbitListener用起来很顺手，但默认配置在生产环境往往不够用。几个关键参数：concurrency控制并发消费者数量，prefetch控制每个消费者预取的消息数，ackMode控制确认模式。默认 prefetch 是 250，这个值在消息处理较慢时会出问题——一个消费者预取了 250 条，处理到第 10 条卡住了，剩下 240 条都在它手里，其他消费者闲着也拿不到，整体吞吐反而下降。一般建议 prefetch 设小一点，比如 10 到 50，让消息在消费者之间更均匀地分配。
ackMode 建议用手动确认（MANUAL），处理成功再 ack，失败 nack 并决定是否重新入队。自动确认（AUTO）在消费者处理到一半崩溃时会丢消息，不适合数据同步这种场景。配置示例：
spring: rabbitmq: listener: simple: concurrency: 5 max-concurrency: 20 prefetch: 20 acknowledge-mode: manualconcurrency是初始消费者数，max-concurrency是弹性上限，RabbitMQ 会根据积压情况自动扩容到这个上限。这个弹性机制在流量突增时很有用，但要注意消费逻辑必须是线程安全的。
4.2 从消息体还原 RowChange：字段类型映射的坑
消息体反序列化回对象后，要按操作类型解析字段。这里最大的坑是字段类型映射。Canal 传过来的字段值都是字符串形式，但数据库里可能是 int、bigint、decimal、datetime、tinyint(1) 等各种类型。直接当字符串用，做数值比较、时间计算时会出错；盲目转成数字，遇到 null 或者空串又会抛异常。
我的做法是维护一份字段元数据，记录每个表的每个字段的实际类型，解析时按类型转换。比如 decimal 用 BigDecimal 接，避免精度丢失；datetime 按配置的时区解析成 LocalDateTime；tinyint(1) 按布尔处理（MySQL 里它常被当布尔用）。这份元数据可以在启动时从数据库information_schema.columns查出来缓存，也可以在 Canal 消息里带（Canal 的 flatMessage 里有 sqlType 信息）。用 sqlType 更可靠，因为它反映的是 binlog 里的真实类型。
还有一个细节：UPDATE 事件的 afterColumns 里，只有真正变更的列updated=true，没变的列也在列表里但updated=false。如果下游要做全量覆盖，得把所有列都取出来；如果只关心变更，就过滤updated=true的。这个判断别搞反了，我见过有人把updated=false的当变更字段处理，结果每次更新都触发全字段重算，白白浪费性能。
4.3 幂等键怎么选：主键、业务唯一键还是消息 ID
消息队列的"至少一次"投递语义决定了重复消息不可避免：网络抖动导致 ack 丢失、消费者处理完还没 ack 就重启、Canal 端 rollback 重发，都会产生重复。消费端必须做幂等，否则缓存被重复失效还算轻的，账户余额被重复扣减就是事故了。
幂等键的选择要看业务。最直接的是用主键 + 操作类型 + 版本号（或者 binlog 位点）组合。binlog 位点（binlog filename + position）是全局唯一的，用它做幂等键最严格，但需要把位点信息透传到消息里，Canal 的 Entry 里有entry.getHeader().getLogfileName()和getLogfileOffset()，可以带上。如果嫌位点太重，用"表名 + 主键 + 操作类型 + 关键字段值哈希"也行，但要注意同一行短时间内多次更新，哈希可能碰撞，得加时间戳或者版本号区分。
实现上，可以用 Redis 的 SETNX 做短期去重（比如 5 分钟内相同幂等键只处理一次），也可以用数据库唯一索引做长期去重。前者轻量但有窗口期，后者严格但要落库。我的经验是，缓存失效这类操作天然幂等（失效多次结果一样），可以不做严格去重；涉及金额、库存的，必须用数据库唯一约束或者乐观锁兜底，别只靠 Redis。
4.4 消费失败的重试、死信与人工介入
消费失败不可怕，可怕的是失败后无限重试把队列堵死。RabbitMQ 的重试机制要配合死信队列（DLX）用。基本思路是：消费失败 nack 且 requeue=false，消息进入死信 exchange，再路由到死信队列，同时可以配置延迟重试（用 TTL + DLX 实现退避）。配置示例：
spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0 max-interval: 10000这样失败后按 1s、2s、4s 退避重试，三次都失败就进死信队列。死信队列要有监控和告警，定期人工排查。这里有个坑：如果异常是消息格式错误这种"确定性失败"，重试多少次都没用，只会白白消耗资源。所以消费逻辑里要区分"可重试异常"（如数据库连接超时）和"不可重试异常"（如反序列化失败、字段缺失），后者直接进死信，别浪费重试次数。
5. 上线后才会暴露的几个真实问题
配置跑通、联调通过，不代表生产就稳了。下面这几个问题，基本都是上线跑了一段时间才冒出来的，提前知道能少走弯路。
5.1 消息积压时的排查顺序与应急手段
消息积压是数据同步链路最常见的告警。看到积压，别急着加消费者，先按顺序排查：第一，看消费端日志有没有大量异常，异常导致的反复重试是积压的头号原因；第二，看消费逻辑是不是变慢了，比如某次发版加了个远程调用，单条处理从 5ms 变成 50ms，吞吐直接掉一个数量级；第三，看是不是上游突然放量，比如批量导入或者大促，这种是正常峰值，加消费者能扛过去；第四，看 RabbitMQ 本身有没有问题，比如磁盘告警触发了流控，生产者被阻塞。
应急手段上，临时扩容消费者是最快的，但前提是消费逻辑支持并发且幂等。如果积压是因为某类消息处理失败反复重试，先把这类消息路由到死信队列，让正常消息先流通，再单独处理问题消息。千万别直接清空队列，数据同步链路丢消息意味着下游数据永久不一致，恢复起来非常麻烦。
5.2 大事务与批量操作导致的消息风暴
一个容易被忽略的场景是大事务。业务方一个批量更新语句改了 10 万行，binlog 里就是 10 万条行变更，Canal 会解析成 10 万条消息瞬间灌进 RabbitMQ，消费端直接被冲垮。这种情况在数据修复、批量导入时特别常见。
应对办法有几个层次。Canal 端可以配置canal.instance.transaction.size控制单次事务拉取的最大条数，避免一次拉太多。消费端可以做批量消费，一次拉一批一起处理，减少单条开销。更根本的是和业务方约定，大批量操作走专门的通道（比如离线任务），别直接在主库上跑大事务。如果实在避免不了，消费端要有削峰能力，比如把消息先落到本地队列或者临时表，再慢慢处理。
5.3 表结构变更后消费端解析失败的连锁反应
加字段、改字段类型、删字段，这些 DDL 操作在业务迭代中很常见，但对消费端是灾难。消费端如果按固定字段列表解析，表加了字段可能不影响（多余字段忽略），但删字段或者改类型就会直接抛异常，导致该类消息全部进死信，积压随之而来。
预防办法是消费端解析要"宽容"：按字段名取值，取不到给默认值或者跳过，别用位置索引；类型转换用 try-catch 包住，转换失败记录日志但不中断整批处理。同时建立 DDL 变更的通知机制，表结构变更前通知下游，消费端提前适配。Canal 本身也能捕获 DDL 事件（EntryType 为 TRANSACTIONBEGIN 之外的 DDL），可以在投递端识别出来，发一条特殊的"结构变更"消息，让消费端刷新元数据缓存。
5.4 监控指标该盯哪几个：延迟、积压、失败率
监控不用面面俱到，盯住几个核心指标就够。延迟方面，看消息从产生到消费完成的时间差，这个最能反映链路健康度，延迟持续上涨说明消费能力跟不上。积压方面，看队列的 ready 消息数，配合延迟一起看。失败率方面，看死信队列的增长速度和消费异常日志的频率。还有一个容易忽略的指标是 Canal 端的位点延迟，即 Canal 解析到的 binlog 位点和 MySQL 当前位点的差距，这个反映的是采集端是否跟得上主库。
这几个指标配上告警阈值，基本能覆盖大部分问题。延迟超过 1 分钟告警、积压超过 1 万条告警、死信队列 5 分钟内增长超过 100 条告警，具体阈值按业务量调整。告警要带上下文，比如当前消费者数量、最近的异常摘要，方便快速定位。
6. 几个能明显提升稳定性的工程习惯
链路跑通只是起点，要让它长期稳定，靠的是一些工程习惯。这些习惯不复杂，但坚持做和不做，差距在半年后会见分晓。
6.1 消息体加版本号，给未来留后路
消息体格式一定会变，加字段、改结构、换序列化方式，这些在项目生命周期里几乎必然发生。如果消息体没有版本号，新旧消费者混跑时就会解析失败。做法很简单，消息头或者消息体里加一个version字段，消费端按版本走不同的解析逻辑，新版本上线时老消费者继续用老逻辑，平滑过渡。这个成本极低，但省下的排查时间不可估量。
6.2 消费逻辑与业务逻辑分离，别把消费端写成大泥球
消费端最容易犯的错是把所有处理逻辑堆在一个 listener 方法里，缓存失效、索引更新、审计落库全塞一起，改一处影响一片。正确的做法是消费端只做"解析和分发"，把消息解析成统一的事件对象，然后按事件类型分发给不同的处理器，每个处理器只管一件事。这样加下游、改逻辑都是增量的，不会牵一发动全身。事件对象的设计要稳定，字段只增不减，语义清晰。
6.3 灰度与回滚：新消费者上线先影子消费
新消费者或者消费逻辑大改上线时，别直接切流量。可以先让新消费者以"影子模式"运行，即订阅同样的消息但处理结果不生效（只打日志或者写影子表），对比新旧处理结果是否一致，确认无误后再切换。这个做法在数据同步这种"错了很难回滚"的场景里尤其重要。回滚方案也要提前想好，比如消费端保留开关，出问题能一键停掉某个处理器，而不是整个服务下线。
6.4 定期做全量对账，别等出事才发现不一致
异步链路再稳，也可能因为各种边角情况出现不一致：消息丢了、消费逻辑有 bug、DDL 导致漏处理。定期对账是最后的兜底。做法是每天低峰期跑一次全量或者抽样对账，比对源库和目标端的关键字段，发现不一致的记录打出来人工或者自动修复。对账任务本身要轻量，别给主库太大压力，可以走从库或者离线数仓。对账发现的问题要复盘根因，是链路问题就修链路，是逻辑问题就修逻辑，别只是把数据修了了事。
这套链路我从早期用 adapter 快速搭原型，到后来用 client 自建投递端做精细控制，中间踩过的坑基本都在这了。最深的体会是，配置和代码只是表象，真正决定这套东西能不能在生产扛住的，是对"至少一次投递""最终一致""幂等"这几个基本概念的敬畏，以及在每个环节都留好退路和观测手段的习惯。链路越复杂，越要在简单的地方下功夫。
