1. 从一次重复消费说起ACK 到底在确认什么Canal 的 ACK 机制简单说就是客户端告诉 Server「这批 binlog 我已经处理完了你可以把位点往前推了」。它解决的是 CDC 链路里最要命的问题消费到哪了、崩了之后从哪继续。适合正在用 Canal 做 MySQL 增量同步、又对数据重复和丢失比较敏感的同学。我见过太多人把getWithoutAck()拿到的数据当成「已经消费成功」业务逻辑还没跑完就急着往下走结果 Job 一重启位点还停在原地同一批数据又灌了一遍下游。ACK 的本质不是「我收到了」而是「我处理完了可以提交位点了」。这个区别决定了你的管道是 At-Least-Once 还是能往 Exactly-Once 靠。Canal 里位点Cursor的格式长这样{journalName:mysql-bin.000003,position:123456,timestamp:1713689000000}它记录的是 binlog 文件名加偏移量。Server 端由 MetaManager 负责把这个位点持久化到 ZooKeeper 或本地文件。客户端每调用一次ack(batchId)Server 就把这个 batch 对应的结束位点写进去。所以 ACK 和位点提交是同一件事的两面ACK 是动作位点更新是结果。下面按「配置骨架 → 客户端 ACK 代码 → 重启验证位点续传」的顺序走一遍每一步都能直接抄。2. TaoToken 前置把模型对话和接入文档放在手边写 Canal 客户端代码时经常要查 API 签名、对照配置项我习惯把文档和模型对话放在一个入口省得来回切。TaoToken 这边可以直接用想快速问「CanalConnector.ack 的参数含义」这类问题走模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite需要拿 Key 调接口做联调去 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite接入细节和参数说明看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果你是要长期跑编码任务、写 Agent 那种Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite注意TaoToken 是模型调用与接入平台不是 Canal 的替代品Canal Server 本身还是要你自己部署和配置。3. 可复制配置canal.properties 与 instance 骨架先让 Server 端能正常跑起来位点持久化方式选对后面 ACK 才有意义。3.1 canal.properties 关键项# canal.properties canal.id 1 canal.ip canal.port 11111 canal.zkServers 127.0.0.1:2181 canal.instance.global.mode spring canal.instance.global.lazy false canal.instance.global.spring.xml classpath:spring/file-instance.xml # 位点持久化默认走 zookeeper也可换 memory/file canal.instance.global.meta.mode zookeepercanal.instance.global.meta.mode决定位点存哪。生产环境用zookeeper本地调试想省事可以用memory但重启就丢位点验证续传时别用它。3.2 instance 配置骨架# conf/example/instance.properties canal.instance.mysql.slaveId 1234 canal.instance.master.address 127.0.0.1:3306 canal.instance.dbUsername canal canal.instance.dbPassword canal canal.instance.defaultDatabaseName test canal.instance.filter.regex test\\..* canal.instance.connectionCharset UTF-8filter.regex用双反斜杠转义写成test\\..*表示只订阅 test 库所有表。这个正则写错会导致订阅不到数据ACK 自然也无从谈起。3.3 启动与位点查看# 启动 canal server sh bin/startup.sh # 查看当前持久化位点 zkCli.sh get /otter/canal/destinations/example/cursor正常会返回类似{journalName:mysql-bin.000003,position:123456,timestamp:1713689000000}的 JSON。记住这个 position后面验证续传要对比。4. 客户端 ACK 代码与验证请求4.1 手动 ACK 的标准写法CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(test\\..*); while (running) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } try { // 1. 业务处理写下游、入库、发 MQ process(message.getEntries()); // 2. 处理成功才 ACK提交位点 connector.ack(batchId); } catch (Exception e) { // 3. 失败不 ACK下次重新拉取同一批 connector.rollback(batchId); } }关键点ack(batchId)必须在业务处理成功之后调用。rollback(batchId)是显式回滚让 Server 重新投递这批数据比什么都不做更清晰。4.2 验证请求确认位点真的推进了处理完一批数据后再查一次 ZooKeeperzkCli.sh get /otter/canal/destinations/example/cursor对比前后的position如果变大了说明 ACK 生效、位点已提交。如果没变检查是不是ack()没被调用或者 batchId 传错了。4.3 重启验证位点续传这是最能说明问题的一步# 1. 记录当前位点 zkCli.sh get /otter/canal/destinations/example/cursor # 2. 停掉客户端往 MySQL 写几条新数据 # 3. 重启客户端观察它从哪个位点开始拉如果客户端重启后从上次 ACK 的位点继续而不是从头再来说明位点续传生效。如果重复消费了已 ACK 的数据多半是 ACK 没提交成功或者位点被别的实例覆盖了。5. 本篇常见错排查5.1 不 ACK 导致内存堆积现象Canal Server 内存持续上涨最终 OOM。原因getWithoutAck()拿到的批次一直没 ACKEventStore 里的数据无法释放Server 还会不断重投同一批。排查看 Server 日志有没有大量重复的 batchId或者监控未 ACK 批次数量。5.2 batchId 传错或重复 ACK// 错误ack 了上一次的 batchId connector.ack(lastBatchId);现象抛batchId:xxx is not exist!。原因batchId 是一次性的ACK 后 Server 就把它从缓存移除了重复 ACK 会报错。排查确保每次getWithoutAck()后用的是当前返回的message.getId()。5.3 位点没持久化现象重启后位点回到旧位置。原因meta.mode配成了memory或者 ZooKeeper 连接失败。排查确认canal.instance.global.meta.mode zookeeper并且zkCli.sh能连上。5.4 与 Flink Checkpoint 没对齐现象Flink Job 重启后重复消费。原因在processElement()里直接 ACK而不是在snapshotState()里跟 Checkpoint 对齐。正确做法是把 ACK 放到 Checkpoint 完成回调里保证「Checkpoint 成功」和「位点提交」是原子的。Override public void snapshotState(FunctionSnapshotContext context) { if (lastProcessedBatchId ! lastAckBatchId) { connector.ack(lastProcessedBatchId); lastAckBatchId lastProcessedBatchId; } }5.5 手动改位点的坑# 仅在数据修复时使用 zkCli.sh set /otter/canal/destinations/example/cursor \ {journalName:mysql-bin.000002,position:99999}改完要重启 Canal Server 才生效。改错位点会导致漏数据或大量重复操作前先备份原值。6. 把 ACK 用对管道才可靠ACK 不是可有可无的收尾动作它是整个 CDC 链路的可靠性开关。TCP Client 手动 ACK 给你最大的控制权代价是你要自己保证「处理成功」和「提交位点」的时序Adapter 自动 ACK 省心但没法跟外部 Checkpoint 对齐Kafka Client 的位点归 Kafka 管配置不当容易丢数据。实操建议就三条业务处理成功再 ACK失败就 rollback位点持久化用 ZooKeeper别用 memory跟 Flink 集成时ACK 必须放进 snapshotState。把这三点做到重启续传、重复消费这些问题基本就绕开了。需要联调接口或查参数时API Keys 和接入文档在这里https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 、https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。长期跑编码和 Agent 任务的话Coding Plan 更省事https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。
