★ FEATURED ARTICLE 做数据同步这行的都清楚业务库的每一次增删改背后往往牵着一串下游动作缓存要失效、搜索索引要重建、报表要刷新、风控要记账。早期大家习惯在业务代码里埋点写库之后顺手发个消息出去简单直接但埋着埋着就发现代码里到处是消息发送逻辑业务和同步彻底耦合在一起改一个字段要动三四个地方谁都不敢轻易碰。后来 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必须是 ONbinlog_format必须是 ROW。如果没开得改配置文件my.cnf 或 my.ini里的log-bin、binlog-formatROW然后重启 MySQL。这里有个坑binlog_format在部分版本里是可以在线改的SET GLOBAL binlog_formatROW但只对新连接生效而且重启后会丢所以务必写进配置文件。server_id这个参数特别容易被忽略。Canal 伪装成从库从库必须有唯一的 server_id如果 Canal 配置的 slaveId 和 MySQL 集群里其他从库冲突复制会直接报错。一般给 Canal 分配一个明显区别于业务从库的号段比如业务从库用 1-100Canal 就用 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.address127.0.0.1:3306 canal.instance.dbUsernamecanal_user canal.instance.dbPasswordyour_password canal.instance.filter.regexmyapp\\.order,myapp\\.user canal.instance.filter.black.regexfilter.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 exchangerouting key 设计成库名.表名.操作类型这样的三段式比如myapp.order.insert、myapp.user.update。下游要订阅订单表所有操作绑定myapp.order.*只订阅插入绑定myapp.order.insert订阅整个库绑定myapp.#。这种设计下新增下游只需要加一个队列和绑定完全不用动生产端扩展性非常好。队列声明时要注意持久化exchange 和 queue 都设 durabletrue消息投递时设 deliveryMode2持久化这样 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 的配置化路线与它的能力边界如果需求比较标准就是想把变更按库表投递到 MQcanal-adapter 是最省事的。它的 RabbitMQ 投递配置大概长这样放在conf/rabbitmq/mq.ymldataSourceKey: 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解析出来拿到EventTypeINSERT/UPDATE/DELETE和每行的Column列表组装成消息体。3.3 消息体设计保留前后镜像还是只发变更字段消息体设计是个容易被低估的决策点。Canal 的 RowData 里UPDATE 事件同时包含 beforeColumns 和 afterColumnsINSERT 只有 afterColumnsDELETE 只有 beforeColumns。全量发过去消息体会比较大但下游能拿到完整信息做审计、做对比都方便。只发变更字段afterColumns 里updatedtrue的列消息体小但下游拿不到完整行做全量覆盖更新时还得回查数据库反而增加负担。我的建议是默认发完整 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 按配置的时区解析成 LocalDateTimetinyint(1) 按布尔处理MySQL 里它常被当布尔用。这份元数据可以在启动时从数据库information_schema.columns查出来缓存也可以在 Canal 消息里带Canal 的 flatMessage 里有 sqlType 信息。用 sqlType 更可靠因为它反映的是 binlog 里的真实类型。还有一个细节UPDATE 事件的 afterColumns 里只有真正变更的列updatedtrue没变的列也在列表里但updatedfalse。如果下游要做全量覆盖得把所有列都取出来如果只关心变更就过滤updatedtrue的。这个判断别搞反了我见过有人把updatedfalse的当变更字段处理结果每次更新都触发全字段重算白白浪费性能。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 且 requeuefalse消息进入死信 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 自建投递端做精细控制中间踩过的坑基本都在这了。最深的体会是配置和代码只是表象真正决定这套东西能不能在生产扛住的是对至少一次投递最终一致幂等这几个基本概念的敬畏以及在每个环节都留好退路和观测手段的习惯。链路越复杂越要在简单的地方下功夫。
阅读完成 · 觉得有帮助？
