1. 为什么要在数据库和业务系统之间架一条“日志传送带”做后端时间久了你一定会遇到这类需求用户刚在后台改了一条订单状态前端页面要立刻刷新运营在管理端下架了一个商品搜索索引得同步更新风控系统需要盯着每一笔账户余额变动一旦发现异常立刻告警。这些场景的共同点是——业务动作发生在数据库里但后续处理逻辑跑在数据库之外。如果每次都靠业务代码在写库之后手动发一条消息代码会变得又臭又长而且一旦有人漏发、错发数据就对不上了。我这次要聊的这套方案核心就是解决这个“写库之后怎么可靠地把变更通知出去”的问题。具体来说是用Canal去伪装成 MySQL 的从库实时抓取数据库的增量变更日志把每一条 INSERT、UPDATE、DELETE 操作解析成结构化消息投递到RabbitMQ里再由下游的监听程序消费、解析、做业务处理。整条链路对业务代码是零侵入的——业务方只管写库剩下的变更捕获和消息分发全部由中间件层完成。这套东西适合谁看如果你正在做数据同步、缓存刷新、搜索索引更新、异步业务解耦、审计日志采集这类工作或者你所在的项目里已经出现了“写库之后要通知好几个下游”的苗头那这套方案基本可以直接抄作业。哪怕你之前没接触过 Canal只要对 MySQL 和消息队列有基本概念跟着下面的思路走一遍也能把整条链路搭起来。我会把每个环节为什么这么选、参数怎么定、坑在哪里都掰开揉碎讲清楚。2. 整体架构设计与技术选型背后的取舍2.1 整条链路到底长什么样先把全景图在脑子里立起来。MySQL 主库正常接收业务写入它的 binlog 按行模式记录每一次数据变更Canal Server 伪装成一个 MySQL 从库向主库发起 dump 请求拿到 binlog 后解析成 JSON 格式的变更事件Canal Client 作为客户端连上 Canal Server订阅指定库表的变更再把消息转发到 RabbitMQ 的交换机最后业务侧的消费者监听对应队列拿到消息后做解析和后续处理。这条链路里Canal 扮演的是“翻译官”角色——它把 MySQL 底层的二进制日志翻译成人能看懂的结构化数据。RabbitMQ 扮演的是“分发中心”角色——它把变更消息按路由规则投递给一个或多个下游。两者之间用 Canal Client 做桥接桥接层是我们唯一需要写代码的地方其余都是配置。为什么不让业务代码直接发消息因为业务代码发消息存在一个致命问题写库和发消息不是一个原子操作。你写完库、还没发消息服务挂了这条变更就丢了你发了消息、写库失败下游就收到了一条假消息。而基于 binlog 的方案消息来源就是数据库已经提交的变更本身天然和数据库状态一致可靠性高出一个量级。2.2 为什么是 Canal 而不是别的方案捕获数据库变更这件事市面上有几条路可走。我简单对比一下你就明白为什么这个场景下 Canal 是合理选择。方案原理优点缺点业务代码双写写库后手动发消息实现简单、可控非原子、易漏发、侵入业务定时轮询定期查表比对无需额外组件实时性差、数据库压力大触发器写中间表DB 触发器落表再扫实时性尚可触发器难维护、影响主库性能基于 binlog 订阅解析数据库日志零侵入、实时、可靠需要额外组件、有运维成本Canal 走的是最后一条路。它的价值在于对业务完全透明业务方不需要改一行代码甚至不需要知道 Canal 的存在。数据库里发生了什么它就如实捕获什么。而且 binlog 是 MySQL 自身保证顺序和完整性的Canal 只是消费者不参与写入对主库的压力非常小——本质上就是多了一个从库在拉日志。这里有个关键前提必须说清楚MySQL 必须开启 binlog且格式必须是 ROW 模式。ROW 模式记录的是每一行数据变更前后的完整镜像Canal 才能准确还原出“哪个字段从什么值变成了什么值”。如果是 STATEMENT 模式记录的是 SQL 语句本身遇到函数、随机数这类场景就无法确定最终值Canal 解析会出问题。这个配置是整套方案的基石后面实操部分我会给出具体设置。2.3 为什么用 RabbitMQ 做消息中转变更消息捕获出来之后总得有个地方暂存和分发。直接让 Canal Client 调用下游接口行不行行但很脆弱——下游一挂消息就丢了而且没法做多下游广播。引入消息队列之后几个问题一次性解决削峰突发大量变更时队列做缓冲、解耦下游有几个消费者、怎么处理Canal 侧不关心、可靠投递配合确认机制和持久化消息不丢。选 RabbitMQ 而不是其他消息中间件主要看中它的路由灵活性。RabbitMQ 的 Exchange 支持 direct、topic、fanout 等多种类型我们可以按“库名.表名”做 routing key下游想订阅哪张表就绑哪个 key非常自然。比如订单服务只关心order表就用db.order绑定审计服务关心所有表就用db.#通配。这种按需订阅的能力在变更分发场景里特别顺手。另外 RabbitMQ 的消息确认机制成熟稳定消费者手动 ack 能保证“处理成功才删消息”配合持久化队列即使消费者重启也不会丢消息。对于变更同步这种“宁可重复也不能丢”的场景这套机制刚好对口。2.4 消息格式怎么设计才不给自己挖坑Canal 吐出来的原始数据是它自己的 JSON 结构字段比较多直接扔给下游会让消费者很难受。我的做法是在桥接层做一次轻量归一化把消息整理成统一格式再投递。核心字段包括变更类型INSERT/UPDATE/DELETE、库名、表名、变更前后的数据用 Map 存字段名到值的映射、变更时间戳、以及一个用于幂等去重的唯一标识。为什么要做归一化因为下游消费者可能有好几种语言、好几个团队在写如果每个人都去啃 Canal 原始结构重复劳动不说还容易理解不一致。统一格式相当于定了一份“变更消息契约”大家按契约办事后续维护成本低很多。至于幂等标识可以用“库名表名主键binlog位点”组合生成消费者拿它做去重避免消息重复投递导致业务重复执行。3. 环境搭建与核心配置的实操细节3.1 MySQL 侧必须做对的几件事第一步永远是数据库配置。登录 MySQL 执行SHOW VARIABLES LIKE log_bin;如果结果是 OFF说明 binlog 没开后面全白搭。开启方式是在配置文件里加上log-binmysql-bin、binlog-formatROW、server-id1这几项然后重启。server-id不能和 Canal 的从库 id 冲突Canal 默认用 1234所以主库设成 1 就行。接着要给 Canal 建一个专用账号授予它作为从库拉取日志的权限。这里有个细节Canal 需要REPLICATION SLAVE和REPLICATION CLIENT权限前者用于拉 binlog后者用于查询位点信息。账号密码别用 root单独建一个最小权限账号安全上也规范。注意binlog_row_image参数建议设为FULL这样 UPDATE 事件会记录变更前后的完整行数据。如果设成MINIMAL只记录被改的字段Canal 拿到的前镜像就不完整做数据比对或回滚时会缺信息。还有一个容易忽略的点binlog 保留时间。如果 Canal 服务停了一段时间而 binlog 已经被清理Canal 再启动时就找不到位点了只能重新全量。所以要么把expire_logs_days设大一点比如 7 天要么保证 Canal 的高可用别让它长时间掉线。这是运维层面必须提前想好的事。3.2 Canal Server 的部署与 instance 配置Canal Server 我一般用 Docker 部署省去装 JDK、配环境的麻烦。拉镜像、挂载配置目录、映射端口几条命令搞定。真正需要花心思的是instance配置——每个 instance 对应一个数据库实例的订阅核心文件是instance.properties。里面几个关键项canal.instance.master.address填主库地址canal.instance.dbUsername和dbPassword填刚才建的账号canal.instance.filter.regex是过滤规则格式是“库名.表名”的正则比如mydb\\..*表示订阅 mydb 下所有表mydb\\.order,mydb\\.product表示只订阅这两张表。这个正则一定要写对写宽了会收到大量无关变更写窄了会漏表。canal.instance.connectionCharset设成UTF8避免中文乱码。canal.instance.tsdb.enable建议开启它会把位点信息持久化到本地服务重启后能从上次的位置继续不用从头拉。这些配置看着琐碎但每一项都直接影响链路的稳定性和数据准确性配错一个就可能出现“消息重复”或“消息丢失”的诡异现象。3.3 RabbitMQ 的交换机与队列规划RabbitMQ 侧的设计要围绕“下游怎么订阅”来展开。我推荐用topic 类型交换机命名为canal.change.exchangerouting key 用库名.表名的格式。下游消费者各自建队列按需绑定。比如订单服务建队列queue.order绑定 keymydb.order库存服务建队列queue.stock绑定 keymydb.stock。这样一条订单变更只会进订单队列不会打扰库存服务。队列一定要设成持久化交换机也持久化消息投递时设deliveryMode2。这三者配合才能保证 RabbitMQ 重启后消息不丢。另外建议给每个队列配一个死信交换机消费失败多次的消息转到死信队列方便事后排查而不是无限重试把队列堵死。提示如果下游有“全量审计”需求可以再建一个队列绑定#通配所有 key专门做日志归档。topic 交换机的通配能力在这里体现得淋漓尽致一个交换机就能同时服务精准订阅和全量订阅两类消费者。3.4 桥接程序的依赖与初始化桥接程序是整条链路里唯一需要写代码的部分。我用 Java 写依赖主要是canal.client和amqp-client两个包。初始化时要做两件事建立到 Canal Server 的连接建立到 RabbitMQ 的连接。两个连接都要做重连机制网络抖动是常态不能让一次断连把整个程序搞崩。Canal 连接用CanalConnectors.newClusterConnector或newSingleConnector拿到CanalConnector后先connect()再subscribe()订阅目标表然后rollback()把位点回退到当前最新避免启动时重复消费历史消息。RabbitMQ 连接用ConnectionFactory设置 host、port、虚拟主机、账号密码拿到Connection后开Channel声明交换机和队列并绑定。这两步初始化代码不长但每个参数都有讲究后面实操章节我会给出完整示例。4. 从 binlog 到消息队列的完整实现过程4.1 拉取与解析变更事件的核心循环桥接程序的主循环是一个典型的“拉取—解析—投递—确认”四步舞。先调connector.getWithoutAck(batchSize)批量拉取变更batchSize我一般设 1000太小网络往返多太大内存占用高1000 是个经验平衡点。拿到Message后里面是一批Entry每个Entry对应一次数据库事务Entry里又有若干RowChange每个RowChange对应一行数据的变更。解析时要区分事件类型。EventType.INSERT取RowData里的 afterColumnsEventType.UPDATE同时取 beforeColumns 和 afterColumns这样才能知道字段改前改后的值EventType.DELETE取 beforeColumns。每个 Column 有name、value、isNull、updated几个属性updated标记这个字段本次是否真的变了做增量同步时很有用——只处理真正变化的字段能减少下游无谓的计算。这里有个实操心得UPDATE 事件里 beforeColumns 可能为空取决于binlog_row_image配置。如果发现拿不到前镜像先回头检查数据库参数别在代码里死磕。另外isNulltrue时value是空字符串要单独判断否则会把 NULL 误当成空串写进下游造成数据语义错误。4.2 消息体组装与 routing key 生成解析出来的数据不能直接扔要组装成前面说的归一化格式。我定义一个ChangeMessage结构包含type、database、table、before、after、timestamp、id七个字段。before和after用MapString, Object存字段名做 key值做 value。id用“库名:表名:主键值:binlog文件名:位点”拼接后取哈希保证同一条变更重复投递时 id 一致下游据此去重。routing key 直接用database . table和 RabbitMQ 的绑定规则对齐。投递前把ChangeMessage序列化成 JSON用channel.basicPublish发到交换机。发送时设MessageProperties.PERSISTENT_TEXT_PLAIN保证持久化同时带上messageId方便追踪。这一步看着简单但序列化框架的选择有讲究——用 Jackson 要注意空值和日期格式用 Fastjson 要注意版本安全我一般用 Jackson 并显式配置FAIL_ON_UNKNOWN_PROPERTIESfalse避免下游字段增减导致反序列化失败。4.3 位点确认与至少一次投递的保证消息投递成功后必须调connector.ack(batchId)告诉 Canal“这批我处理完了”。这个 ack 是 Canal 位点推进的依据ack 了才会把位点存下来下次从新位点继续。如果投递失败就调connector.rollback(batchId)Canal 会把位点回退下次重新拉这批数据。这套机制保证的是至少一次投递——消息不会丢但可能重复。为什么可能重复因为“投递到 RabbitMQ”和“ack 给 Canal”是两个操作如果投递成功但 ack 前程序崩了重启后这批数据会再拉一遍、再投一遍。这不是 bug是分布式系统的常态。应对办法就是前面说的幂等 id下游消费时先查这个 id 处理过没有处理过就跳过。宁可重复不可丢失这是变更同步场景的基本原则别为了追求“恰好一次”把架构搞得过于复杂。注意批量 ack 的粒度要和批量拉取对齐。如果一批 1000 条里第 500 条投递失败整批 rollback 会导致前 499 条重复投递。想减少重复可以把 batchSize 调小或者做分批投递、分批 ack用一点性能换一点重复率具体看业务对重复的容忍度。4.4 消费者侧的监听与解析处理消费者侧用 RabbitMQ 的basicConsume注册监听器autoAck设为 false手动确认。监听器拿到消息后先反序列化成ChangeMessage然后按type分派处理INSERT 走新增逻辑UPDATE 比对 before 和 after 找出变化字段DELETE 走删除逻辑。处理成功调basicAck处理异常调basicNack并设requeuefalse转入死信队列避免毒消息卡死队列。消费逻辑里最容易出问题的是顺序性。RabbitMQ 单队列单消费者能保证顺序但一旦开多个消费者并发消费同一条记录的先 UPDATE 后 DELETE 可能乱序。如果业务对顺序敏感要么单消费者串行处理要么在消息里带版本号或时间戳消费端做乱序检测。这个取舍要在设计阶段就想清楚别等上线后发现数据错乱再回头改。5. 踩过的坑与排查速查表5.1 消息重复、丢失、乱序的典型成因这套链路跑起来之后我遇到过几类高频问题整理成表方便对照排查。现象可能原因排查方向解决手段消息大量重复ack 前程序重启、批量 rollback看 Canal 日志位点是否回退下游做幂等、调小 batchSize消息丢失队列未持久化、autoAck 为 true检查队列 durable 和消费确认模式队列持久化、手动 ack消费乱序多消费者并发、多队列看是否单队列多消费者单消费者或加版本号校验收不到变更filter 正则写错、binlog 未开查 instance 日志有无订阅记录修正正则、确认 ROW 模式中文乱码字符集配置不一致查 connectionCharset 和连接串统一设 UTF8重复问题最容易被新手当成 bug 去修其实它是设计使然。我见过有人为了消除重复把 ack 提前到投递之前结果消息一丢一大片得不偿失。正确姿势永远是接受重复、做好幂等。5.2 Canal 位点异常的处理经验有一次测试环境 Canal 重启后疯狂重复投递几天前的数据查下来是tsdb目录被误删位点信息丢了Canal 从 binlog 最早位置开始拉。解决办法是重新配置meta.dat里的位点或者干脆用canal.instance.master.position手动指定一个较新的位点启动。这件事给我的教训是位点持久化目录要纳入备份和监控别让它成为单点。还有一种情况是主库做了主从切换Canal 连的地址变了位点对不上。这时候要么改配置指向新主库要么用canal.instance.master.journal.name和position手动对齐。生产环境建议给 Canal 配高可用主库切换时能自动跟随减少人工干预。5.3 消费端幂等与死信处理的落地写法幂等这块我的做法是在消费端维护一张去重表主键就是消息里的id处理前先INSERT IGNORE或查一下存在就跳过。去重表要定期清理比如只保留 7 天的记录否则会无限膨胀。另一种做法是用 Redis 存id做 SETNX带过期时间轻量但要注意 Redis 挂了之后的降级策略。死信处理别偷懒。消息进死信队列不是终点得有告警和人工介入通道。我一般给死信队列配一个监控数量超过阈值就发通知同时提供一个重放工具排查完原因后能把死信消息重新投回原队列。这套兜底机制平时用不上但真出问题时能救命。6. 几个让链路更稳的进阶技巧6.1 按表分流与消费能力匹配当订阅的表越来越多单队列会成为瓶颈。我的做法是按业务域拆队列订单相关表进一个队列商品相关表进另一个各自独立消费、独立扩容。RabbitMQ 的 topic 绑定天然支持这种拆分改的只是绑定关系桥接层代码几乎不用动。拆分之后某个域消费慢不会拖累其他域整体吞吐和稳定性都上一个台阶。6.2 大事务与批量变更的应对遇到批量更新几万行的 SQLbinlog 里会产生一个巨大的事务Canal 解析出来是一大批 RowChange一次性投递可能撑爆内存或超过 RabbitMQ 的消息大小限制。应对办法是在桥接层做分片投递把一个大 Entry 拆成多个小批次发送每批控制在几百条。同时监控单条消息大小超过阈值就告警提醒业务方避免大事务。这个细节不做平时没事一到大促批量操作就出问题。6.3 监控指标该盯哪几个链路稳不稳看几个核心指标就够了Canal 的延迟当前位点和最新 binlog 位点的差距、RabbitMQ 各队列的堆积数、消费端的处理耗时和失败率。延迟持续上涨说明桥接层处理不过来堆积数上涨说明消费端能力不足失败率升高要立刻查死信。这几个指标配上告警阈值基本能覆盖 90% 的异常场景。我在实际运维中发现大部分故障在指标上都有前兆提前几分钟看到曲线异常就能避免一次线上事故。6.4 版本兼容与升级注意事项Canal 和 MySQL 的版本兼容性要留意尤其是 MySQL 8.x 之后 binlog 格式有些变化老版本 Canal 可能解析异常。升级前先在测试环境跑通全链路确认 INSERT、UPDATE、DELETE 三类事件都能正确解析再上生产。RabbitMQ 客户端版本也要和服务器版本匹配协议不兼容时连接会直接失败报错信息还不一定直观容易浪费排查时间。最后分享一个我自己的习惯每次调整 filter 正则或队列绑定都先在测试环境用一张测试表造几条增删改数据确认消息能正确流转到消费端再动生产配置。变更同步这条链路配置错一个字符就可能静默丢数据肉眼很难发现靠测试用例兜底最踏实。这套方案我从零搭过几次每次都会在桥接层加一段“自检日志”把拉到的变更条数、投递成功数、ack 位点打出来出问题时这几行日志比什么都管用。
1. 为什么要在数据库和业务系统之间架一条“日志传送带”做后端时间久了你一定会遇到这类需求用户刚在后台改了一条订单状态前端页面要立刻刷新运营在管理端下架了一个商品搜索索引得同步更新风控系统需要盯着每一笔账户余额变动一旦发现异常立刻告警。这些场景的共同点是——业务动作发生在数据库里但后续处理逻辑跑在数据库之外。如果每次都靠业务代码在写库之后手动发一条消息代码会变得又臭又长而且一旦有人漏发、错发数据就对不上了。我这次要聊的这套方案核心就是解决这个“写库之后怎么可靠地把变更通知出去”的问题。具体来说是用Canal去伪装成 MySQL 的从库实时抓取数据库的增量变更日志把每一条 INSERT、UPDATE、DELETE 操作解析成结构化消息投递到RabbitMQ里再由下游的监听程序消费、解析、做业务处理。整条链路对业务代码是零侵入的——业务方只管写库剩下的变更捕获和消息分发全部由中间件层完成。这套东西适合谁看如果你正在做数据同步、缓存刷新、搜索索引更新、异步业务解耦、审计日志采集这类工作或者你所在的项目里已经出现了“写库之后要通知好几个下游”的苗头那这套方案基本可以直接抄作业。哪怕你之前没接触过 Canal只要对 MySQL 和消息队列有基本概念跟着下面的思路走一遍也能把整条链路搭起来。我会把每个环节为什么这么选、参数怎么定、坑在哪里都掰开揉碎讲清楚。2. 整体架构设计与技术选型背后的取舍2.1 整条链路到底长什么样先把全景图在脑子里立起来。MySQL 主库正常接收业务写入它的 binlog 按行模式记录每一次数据变更Canal Server 伪装成一个 MySQL 从库向主库发起 dump 请求拿到 binlog 后解析成 JSON 格式的变更事件Canal Client 作为客户端连上 Canal Server订阅指定库表的变更再把消息转发到 RabbitMQ 的交换机最后业务侧的消费者监听对应队列拿到消息后做解析和后续处理。这条链路里Canal 扮演的是“翻译官”角色——它把 MySQL 底层的二进制日志翻译成人能看懂的结构化数据。RabbitMQ 扮演的是“分发中心”角色——它把变更消息按路由规则投递给一个或多个下游。两者之间用 Canal Client 做桥接桥接层是我们唯一需要写代码的地方其余都是配置。为什么不让业务代码直接发消息因为业务代码发消息存在一个致命问题写库和发消息不是一个原子操作。你写完库、还没发消息服务挂了这条变更就丢了你发了消息、写库失败下游就收到了一条假消息。而基于 binlog 的方案消息来源就是数据库已经提交的变更本身天然和数据库状态一致可靠性高出一个量级。2.2 为什么是 Canal 而不是别的方案捕获数据库变更这件事市面上有几条路可走。我简单对比一下你就明白为什么这个场景下 Canal 是合理选择。方案原理优点缺点业务代码双写写库后手动发消息实现简单、可控非原子、易漏发、侵入业务定时轮询定期查表比对无需额外组件实时性差、数据库压力大触发器写中间表DB 触发器落表再扫实时性尚可触发器难维护、影响主库性能基于 binlog 订阅解析数据库日志零侵入、实时、可靠需要额外组件、有运维成本Canal 走的是最后一条路。它的价值在于对业务完全透明业务方不需要改一行代码甚至不需要知道 Canal 的存在。数据库里发生了什么它就如实捕获什么。而且 binlog 是 MySQL 自身保证顺序和完整性的Canal 只是消费者不参与写入对主库的压力非常小——本质上就是多了一个从库在拉日志。这里有个关键前提必须说清楚MySQL 必须开启 binlog且格式必须是 ROW 模式。ROW 模式记录的是每一行数据变更前后的完整镜像Canal 才能准确还原出“哪个字段从什么值变成了什么值”。如果是 STATEMENT 模式记录的是 SQL 语句本身遇到函数、随机数这类场景就无法确定最终值Canal 解析会出问题。这个配置是整套方案的基石后面实操部分我会给出具体设置。2.3 为什么用 RabbitMQ 做消息中转变更消息捕获出来之后总得有个地方暂存和分发。直接让 Canal Client 调用下游接口行不行行但很脆弱——下游一挂消息就丢了而且没法做多下游广播。引入消息队列之后几个问题一次性解决削峰突发大量变更时队列做缓冲、解耦下游有几个消费者、怎么处理Canal 侧不关心、可靠投递配合确认机制和持久化消息不丢。选 RabbitMQ 而不是其他消息中间件主要看中它的路由灵活性。RabbitMQ 的 Exchange 支持 direct、topic、fanout 等多种类型我们可以按“库名.表名”做 routing key下游想订阅哪张表就绑哪个 key非常自然。比如订单服务只关心order表就用db.order绑定审计服务关心所有表就用db.#通配。这种按需订阅的能力在变更分发场景里特别顺手。另外 RabbitMQ 的消息确认机制成熟稳定消费者手动 ack 能保证“处理成功才删消息”配合持久化队列即使消费者重启也不会丢消息。对于变更同步这种“宁可重复也不能丢”的场景这套机制刚好对口。2.4 消息格式怎么设计才不给自己挖坑Canal 吐出来的原始数据是它自己的 JSON 结构字段比较多直接扔给下游会让消费者很难受。我的做法是在桥接层做一次轻量归一化把消息整理成统一格式再投递。核心字段包括变更类型INSERT/UPDATE/DELETE、库名、表名、变更前后的数据用 Map 存字段名到值的映射、变更时间戳、以及一个用于幂等去重的唯一标识。为什么要做归一化因为下游消费者可能有好几种语言、好几个团队在写如果每个人都去啃 Canal 原始结构重复劳动不说还容易理解不一致。统一格式相当于定了一份“变更消息契约”大家按契约办事后续维护成本低很多。至于幂等标识可以用“库名表名主键binlog位点”组合生成消费者拿它做去重避免消息重复投递导致业务重复执行。3. 环境搭建与核心配置的实操细节3.1 MySQL 侧必须做对的几件事第一步永远是数据库配置。登录 MySQL 执行SHOW VARIABLES LIKE log_bin;如果结果是 OFF说明 binlog 没开后面全白搭。开启方式是在配置文件里加上log-binmysql-bin、binlog-formatROW、server-id1这几项然后重启。server-id不能和 Canal 的从库 id 冲突Canal 默认用 1234所以主库设成 1 就行。接着要给 Canal 建一个专用账号授予它作为从库拉取日志的权限。这里有个细节Canal 需要REPLICATION SLAVE和REPLICATION CLIENT权限前者用于拉 binlog后者用于查询位点信息。账号密码别用 root单独建一个最小权限账号安全上也规范。注意binlog_row_image参数建议设为FULL这样 UPDATE 事件会记录变更前后的完整行数据。如果设成MINIMAL只记录被改的字段Canal 拿到的前镜像就不完整做数据比对或回滚时会缺信息。还有一个容易忽略的点binlog 保留时间。如果 Canal 服务停了一段时间而 binlog 已经被清理Canal 再启动时就找不到位点了只能重新全量。所以要么把expire_logs_days设大一点比如 7 天要么保证 Canal 的高可用别让它长时间掉线。这是运维层面必须提前想好的事。3.2 Canal Server 的部署与 instance 配置Canal Server 我一般用 Docker 部署省去装 JDK、配环境的麻烦。拉镜像、挂载配置目录、映射端口几条命令搞定。真正需要花心思的是instance配置——每个 instance 对应一个数据库实例的订阅核心文件是instance.properties。里面几个关键项canal.instance.master.address填主库地址canal.instance.dbUsername和dbPassword填刚才建的账号canal.instance.filter.regex是过滤规则格式是“库名.表名”的正则比如mydb\\..*表示订阅 mydb 下所有表mydb\\.order,mydb\\.product表示只订阅这两张表。这个正则一定要写对写宽了会收到大量无关变更写窄了会漏表。canal.instance.connectionCharset设成UTF8避免中文乱码。canal.instance.tsdb.enable建议开启它会把位点信息持久化到本地服务重启后能从上次的位置继续不用从头拉。这些配置看着琐碎但每一项都直接影响链路的稳定性和数据准确性配错一个就可能出现“消息重复”或“消息丢失”的诡异现象。3.3 RabbitMQ 的交换机与队列规划RabbitMQ 侧的设计要围绕“下游怎么订阅”来展开。我推荐用topic 类型交换机命名为canal.change.exchangerouting key 用库名.表名的格式。下游消费者各自建队列按需绑定。比如订单服务建队列queue.order绑定 keymydb.order库存服务建队列queue.stock绑定 keymydb.stock。这样一条订单变更只会进订单队列不会打扰库存服务。队列一定要设成持久化交换机也持久化消息投递时设deliveryMode2。这三者配合才能保证 RabbitMQ 重启后消息不丢。另外建议给每个队列配一个死信交换机消费失败多次的消息转到死信队列方便事后排查而不是无限重试把队列堵死。提示如果下游有“全量审计”需求可以再建一个队列绑定#通配所有 key专门做日志归档。topic 交换机的通配能力在这里体现得淋漓尽致一个交换机就能同时服务精准订阅和全量订阅两类消费者。3.4 桥接程序的依赖与初始化桥接程序是整条链路里唯一需要写代码的部分。我用 Java 写依赖主要是canal.client和amqp-client两个包。初始化时要做两件事建立到 Canal Server 的连接建立到 RabbitMQ 的连接。两个连接都要做重连机制网络抖动是常态不能让一次断连把整个程序搞崩。Canal 连接用CanalConnectors.newClusterConnector或newSingleConnector拿到CanalConnector后先connect()再subscribe()订阅目标表然后rollback()把位点回退到当前最新避免启动时重复消费历史消息。RabbitMQ 连接用ConnectionFactory设置 host、port、虚拟主机、账号密码拿到Connection后开Channel声明交换机和队列并绑定。这两步初始化代码不长但每个参数都有讲究后面实操章节我会给出完整示例。4. 从 binlog 到消息队列的完整实现过程4.1 拉取与解析变更事件的核心循环桥接程序的主循环是一个典型的“拉取—解析—投递—确认”四步舞。先调connector.getWithoutAck(batchSize)批量拉取变更batchSize我一般设 1000太小网络往返多太大内存占用高1000 是个经验平衡点。拿到Message后里面是一批Entry每个Entry对应一次数据库事务Entry里又有若干RowChange每个RowChange对应一行数据的变更。解析时要区分事件类型。EventType.INSERT取RowData里的 afterColumnsEventType.UPDATE同时取 beforeColumns 和 afterColumns这样才能知道字段改前改后的值EventType.DELETE取 beforeColumns。每个 Column 有name、value、isNull、updated几个属性updated标记这个字段本次是否真的变了做增量同步时很有用——只处理真正变化的字段能减少下游无谓的计算。这里有个实操心得UPDATE 事件里 beforeColumns 可能为空取决于binlog_row_image配置。如果发现拿不到前镜像先回头检查数据库参数别在代码里死磕。另外isNulltrue时value是空字符串要单独判断否则会把 NULL 误当成空串写进下游造成数据语义错误。4.2 消息体组装与 routing key 生成解析出来的数据不能直接扔要组装成前面说的归一化格式。我定义一个ChangeMessage结构包含type、database、table、before、after、timestamp、id七个字段。before和after用MapString, Object存字段名做 key值做 value。id用“库名:表名:主键值:binlog文件名:位点”拼接后取哈希保证同一条变更重复投递时 id 一致下游据此去重。routing key 直接用database . table和 RabbitMQ 的绑定规则对齐。投递前把ChangeMessage序列化成 JSON用channel.basicPublish发到交换机。发送时设MessageProperties.PERSISTENT_TEXT_PLAIN保证持久化同时带上messageId方便追踪。这一步看着简单但序列化框架的选择有讲究——用 Jackson 要注意空值和日期格式用 Fastjson 要注意版本安全我一般用 Jackson 并显式配置FAIL_ON_UNKNOWN_PROPERTIESfalse避免下游字段增减导致反序列化失败。4.3 位点确认与至少一次投递的保证消息投递成功后必须调connector.ack(batchId)告诉 Canal“这批我处理完了”。这个 ack 是 Canal 位点推进的依据ack 了才会把位点存下来下次从新位点继续。如果投递失败就调connector.rollback(batchId)Canal 会把位点回退下次重新拉这批数据。这套机制保证的是至少一次投递——消息不会丢但可能重复。为什么可能重复因为“投递到 RabbitMQ”和“ack 给 Canal”是两个操作如果投递成功但 ack 前程序崩了重启后这批数据会再拉一遍、再投一遍。这不是 bug是分布式系统的常态。应对办法就是前面说的幂等 id下游消费时先查这个 id 处理过没有处理过就跳过。宁可重复不可丢失这是变更同步场景的基本原则别为了追求“恰好一次”把架构搞得过于复杂。注意批量 ack 的粒度要和批量拉取对齐。如果一批 1000 条里第 500 条投递失败整批 rollback 会导致前 499 条重复投递。想减少重复可以把 batchSize 调小或者做分批投递、分批 ack用一点性能换一点重复率具体看业务对重复的容忍度。4.4 消费者侧的监听与解析处理消费者侧用 RabbitMQ 的basicConsume注册监听器autoAck设为 false手动确认。监听器拿到消息后先反序列化成ChangeMessage然后按type分派处理INSERT 走新增逻辑UPDATE 比对 before 和 after 找出变化字段DELETE 走删除逻辑。处理成功调basicAck处理异常调basicNack并设requeuefalse转入死信队列避免毒消息卡死队列。消费逻辑里最容易出问题的是顺序性。RabbitMQ 单队列单消费者能保证顺序但一旦开多个消费者并发消费同一条记录的先 UPDATE 后 DELETE 可能乱序。如果业务对顺序敏感要么单消费者串行处理要么在消息里带版本号或时间戳消费端做乱序检测。这个取舍要在设计阶段就想清楚别等上线后发现数据错乱再回头改。5. 踩过的坑与排查速查表5.1 消息重复、丢失、乱序的典型成因这套链路跑起来之后我遇到过几类高频问题整理成表方便对照排查。现象可能原因排查方向解决手段消息大量重复ack 前程序重启、批量 rollback看 Canal 日志位点是否回退下游做幂等、调小 batchSize消息丢失队列未持久化、autoAck 为 true检查队列 durable 和消费确认模式队列持久化、手动 ack消费乱序多消费者并发、多队列看是否单队列多消费者单消费者或加版本号校验收不到变更filter 正则写错、binlog 未开查 instance 日志有无订阅记录修正正则、确认 ROW 模式中文乱码字符集配置不一致查 connectionCharset 和连接串统一设 UTF8重复问题最容易被新手当成 bug 去修其实它是设计使然。我见过有人为了消除重复把 ack 提前到投递之前结果消息一丢一大片得不偿失。正确姿势永远是接受重复、做好幂等。5.2 Canal 位点异常的处理经验有一次测试环境 Canal 重启后疯狂重复投递几天前的数据查下来是tsdb目录被误删位点信息丢了Canal 从 binlog 最早位置开始拉。解决办法是重新配置meta.dat里的位点或者干脆用canal.instance.master.position手动指定一个较新的位点启动。这件事给我的教训是位点持久化目录要纳入备份和监控别让它成为单点。还有一种情况是主库做了主从切换Canal 连的地址变了位点对不上。这时候要么改配置指向新主库要么用canal.instance.master.journal.name和position手动对齐。生产环境建议给 Canal 配高可用主库切换时能自动跟随减少人工干预。5.3 消费端幂等与死信处理的落地写法幂等这块我的做法是在消费端维护一张去重表主键就是消息里的id处理前先INSERT IGNORE或查一下存在就跳过。去重表要定期清理比如只保留 7 天的记录否则会无限膨胀。另一种做法是用 Redis 存id做 SETNX带过期时间轻量但要注意 Redis 挂了之后的降级策略。死信处理别偷懒。消息进死信队列不是终点得有告警和人工介入通道。我一般给死信队列配一个监控数量超过阈值就发通知同时提供一个重放工具排查完原因后能把死信消息重新投回原队列。这套兜底机制平时用不上但真出问题时能救命。6. 几个让链路更稳的进阶技巧6.1 按表分流与消费能力匹配当订阅的表越来越多单队列会成为瓶颈。我的做法是按业务域拆队列订单相关表进一个队列商品相关表进另一个各自独立消费、独立扩容。RabbitMQ 的 topic 绑定天然支持这种拆分改的只是绑定关系桥接层代码几乎不用动。拆分之后某个域消费慢不会拖累其他域整体吞吐和稳定性都上一个台阶。6.2 大事务与批量变更的应对遇到批量更新几万行的 SQLbinlog 里会产生一个巨大的事务Canal 解析出来是一大批 RowChange一次性投递可能撑爆内存或超过 RabbitMQ 的消息大小限制。应对办法是在桥接层做分片投递把一个大 Entry 拆成多个小批次发送每批控制在几百条。同时监控单条消息大小超过阈值就告警提醒业务方避免大事务。这个细节不做平时没事一到大促批量操作就出问题。6.3 监控指标该盯哪几个链路稳不稳看几个核心指标就够了Canal 的延迟当前位点和最新 binlog 位点的差距、RabbitMQ 各队列的堆积数、消费端的处理耗时和失败率。延迟持续上涨说明桥接层处理不过来堆积数上涨说明消费端能力不足失败率升高要立刻查死信。这几个指标配上告警阈值基本能覆盖 90% 的异常场景。我在实际运维中发现大部分故障在指标上都有前兆提前几分钟看到曲线异常就能避免一次线上事故。6.4 版本兼容与升级注意事项Canal 和 MySQL 的版本兼容性要留意尤其是 MySQL 8.x 之后 binlog 格式有些变化老版本 Canal 可能解析异常。升级前先在测试环境跑通全链路确认 INSERT、UPDATE、DELETE 三类事件都能正确解析再上生产。RabbitMQ 客户端版本也要和服务器版本匹配协议不兼容时连接会直接失败报错信息还不一定直观容易浪费排查时间。最后分享一个我自己的习惯每次调整 filter 正则或队列绑定都先在测试环境用一张测试表造几条增删改数据确认消息能正确流转到消费端再动生产配置。变更同步这条链路配置错一个字符就可能静默丢数据肉眼很难发现靠测试用例兜底最踏实。这套方案我从零搭过几次每次都会在桥接层加一段“自检日志”把拉到的变更条数、投递成功数、ack 位点打出来出问题时这几行日志比什么都管用。
