最近帮一个订单系统接Canal用它监听MySQL的binlog日志把数据准实时同步到Redis和ES整个过程踩了不少坑。整理一下这整套方案的思路、配置细节和排障经验给后面接手类似业务数据库变一下Redis缓存和ES索引马上跟着变需求的朋友做个参考。先说清楚这套东西是什么Canal 是阿里巴巴开源的一个 MySQL binlog 增量订阅组件它把自己伪装成一个 MySQL 从库通过主从复制协议去拉取 binlog然后解析成结构化事件下游无论是写 Redis、ES还是丢进 MQ 里慢慢消费都很顺手。适合的场景很典型——缓存里的热点数据、搜索引擎里的商品或订单数据不能老靠定时任务刷全量也不能在业务代码里双写那就让 Canal 在中间做数据搬运工。这套方案适合谁后端开发、数据同步开发、搞中间件的人都能用上。即使你还没接触过 binlog看完也能照着搭起来知道每一步为什么这么配遇到问题往哪个方向查。1. 业务侧的真实痛点缓存和搜索为何总是慢半拍1.1 缓存失效与搜索索引更新的死结大多数业务系统里MySQL 是数据的最终归宿但 MySQL 扛不住高并发读所以前面挡一层 Redis而复杂查询、模糊搜索、聚合条件MySQL 的 like 和联合查询又很吃力于是又上 Elasticsearch。问题就出在这同样一份数据在 MySQL、Redis、ES 里各存了一份谁来保证它们之间的一致性我见过很多团队的做法是业务代码双写。更新数据库的时候顺手把 Redis 删掉、把 ES 更新一下。听起来没什么实际跑起来全是坑。第一双写导致业务逻辑里到处都是缓存代码和搜索同步代码换数据库或者换缓存的成本极高第二双写不是原子的数据库提交成功、ES 写入失败库存和搜索就永远不一致了第三为了补偿失败还得再加消息队列重试逻辑越来越复杂。另一种常用方案是定时任务扫表。每隔一分钟把变更的数据捞出来同步一次。这个方案在小数据量下能凑合但实时性太差而且每次全表扫描对 MySQL 压力很大扫到刚更新过的数据还好扫不到就会有窗口期。数据量一上来定时任务还没跑完业务数据又变了。Canal 能破这个局是因为它站在数据库外面通过 binlog 感知每一次行级变更把数据库自己产生的日志变成下游存储的更新指令。业务代码不用改一行缓存和搜索的同步完全剥离出去。1.2 binlog 为什么能当数据总线MySQL 本身就有一套日志机制叫 binlog二进制日志专门用来记录所有数据变更。主从复制就是靠它实现的主库把 binlog 发给从库从库拿到后重放就得到了和主库一样的数据。binlog 有三种格式STATEMENT、ROW、MIXED。STATEMENT 记录的是 SQL 原句比如UPDATE user SET name张三 WHERE id1从库拿到后重新执行一遍。ROW 记录的是每一行变更前后的完整字段值对 Canal 来说这才是最有价值的东西——它不需要知道你的 SQL 长什么样只需要拿到id1这行数据更新前是什么、更新后是什么就能直接把结果塞给 Redis 或 ES。所以 Canal 设计上做了一件很妙的事它把自己伪装成一个从库向 MySQL 发送 dump 请求MySQL 就把 binlog 推给它。Canal 拿到日志后解析成结构化消息下游想怎么消费怎么消费。这和我们直接监听主从复制的区别在于主从复制是把 binlog 应用到另一台 MySQL而 Canal 是把它变成事件流喂给异构存储。相当于把 MySQL 的日志能力扩展成了一根数据总线接多少个下游都行下游挂了也不影响 MySQL 本身。2. 整体方案设计先画一条数据流转链路2.1 模式对比Canal Adapter 还是自研 ClientCanal 官方除了 Server 本体还提供了一个 Adapter 组件可以理解成开箱即用的消费者它直接从 Canal Server 拿增量数据再通过配置把数据写到 Redis、ES、RDB 等目标端。使用 Canal Adapter 的好处很直接不用写代码。表结构定了以后写个映射配置文件启动 Adapter数据就开始同步了。尤其适合那种字段不复杂、SCHEMA 比较稳的表。但它的短板也很明显。Adapter 的同步逻辑是内置好的按表映射来做全字段覆盖。如果你想做字段裁剪、数据清洗、调用下游接口、做复杂幂等判断Adapter 就不太够用了。这种时候需要写一个自己的消费端使用 Canal 提供的 Java Client 从 Server 拉取数据自己解析 Entry再按业务规则去更新 Redis 和 ES。两种模式并不互斥。我的建议是快速验证阶段先上 Adapter跑通链路、观察效果如果后面出现复杂场景比如同一个表要同步成两种不同的 Redis 结构或者要把几条 binlog 聚合后一起写 ES就切到自研 Client。2.2 引入 MQ 与否资源如何权衡有一种常见架构是 Canal Server 把 binlog 直接投递到 Kafka 或 RocketMQ下游再各自消费。这样做有几个好处一是 Canal 和 Target 存储解耦Canal 只负责把事件写进 MQ写入 Redis/ES 的消费端可以独立扩缩容二是削峰填谷瞬间大批量更新时MQ 会帮下游把压力摊平避免 Redis 和 ES 被同时打爆三是 MQ 自带位点管理消费端重启后能接着上次的位置消费比起自己维护 offset 省事。但引入 MQ 也意味着多一个中间件要运维链路更长排查问题也更麻烦。数据量不大、对延迟容忍度较高、团队规模小的时候直接用 Adapter 或 Client 直连 Canal Server 完全够用。我在实际操作里一般这样判断单表日均变更量低于几十万直接用 Adapter 或简单 Client变更量很大或者下游消费逻辑很重就上 MQ。前者图省事后者图安稳没有绝对最优只有当前阶段合不合适。2.3 同步粒度与存储策略的划分不是所有表都要同步也不是所有表同步后 Redis 和 ES 都要一份。我习惯先把表按用途分三类。第一类热点查询型。比如商品基本信息、用户资料这些数据高频读、低频写适合同步到 Redis用 hash 或 string 缓存key 设计成表名:id。这类表不需要同步 ES。第二类搜索筛选型。比如订单表、商品 SPU 表查询条件多、需要分词、需要排序这类表必须同步到 ES。同步时可以只同步需要的字段通过 SQL 把关联表拼好再写入一个文档没必要把 MySQL 原表机械地镜像成一个 index。第三类宽表型。比如一张订单宽表既要做详情缓存又要做列表搜索那么 Redis 和 ES 都需要。同步的时候更新 Redis 一份完整 JSON更新 ES 一份文档模型两份数据的字段可以不同只要保证主键一致就行。把表分好类之后Canal 的订阅过滤也更好配。比如在 Canal Server 的 instance 里只订阅某个库的某几张表或者在 Client 里做正则过滤减少无效日志的解析开销。3. 环境准备MySQL 开启 binlog 的完整姿势3.1 检查状态与修改 MySQL 配置网上 MySQL 安装教程很多这里不重复我直接讲 binlog 相关配置。先确认当前实例是否已经开启 binlogSHOW VARIABLES LIKE log_bin;如果返回值是OFF就需要改配置文件。Linux 下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnfWindows 下是my.ini。在[mysqld]段下面加上[mysqld] server-id10 log-binmysql-bin binlog_formatROW binlog_row_imageFULL expire_logs_days7 gtid_modeON enforce_gtid_consistencyON逐行解释一下为什么这么配。server-id在复制协议里是实例的唯一标识Canal 伪装从库时也需要一个不一样的值不能和现存的主从拓扑里任何节点的 server-id 重复否则 MySQL 会踢掉其中一个连接。log-binmysql-bin指定日志文件前缀。binlog_formatROW是最关键的一项上面说过ROW 格式才有行级变更的字段值STATEMENT 让 Canal 没法高效构造目标数据。binlog_row_imageFULL很容易被忽略它表示在 ROW 格式下记录整行所有字段的镜像。如果改成 MINIMAL日志里只包含主键和发生变化的列对某些场景能省日志量但对同步 Redis 或 ES 来说我们往往需要完整的字段值所以才设成 FULL。expire_logs_days7控制日志保留天数避免磁盘被灌满。gtid_modeON和enforce_gtid_consistencyON是给 Canal 一个更稳定的位点管理方式后面配置 instance 时会用到。改完配置需要重启 MySQL。重启后再次查SHOW VARIABLES LIKE log_bin;确认是 ON然后执行SHOW MASTER STATUS;会看到类似mysql-bin.000001 | 154 | ...这样的输出这就是当前 binlog 文件和 position 点位。后面配置 Canal 时如果不想从最新开始消费就可以填写这个点位。注意不要手贱去服务器上直接rmbinlog 文件。如果日志占满磁盘用 MySQL 自带的清理命令比如PURGE BINARY LOGS TO mysql-bin.000100;或者通过expire_logs_days自动过期。3.2 创建最小权限的 Canal 专用账号Canal 需要连上 MySQL 拉取 binlog但它不应该有业务库的写权限。给它建一个专用账号权限收窄到最小集合CREATE USER canal% IDENTIFIED WITH mysql_native_password BY 你的强密码; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;SELECT权限是 Canal Adapter 做全量数据初始化ETL时要读取表数据用的普通增量同步其实只需要后面两个REPLICATION权限。REPLICATION SLAVE让 Canal 可以请求 binlogREPLICATION CLIENT让它可以查看 master 状态和 server 信息。如果你的 MySQL 是 8.0 以上默认认证插件是caching_sha2_password老版本 Canal 可能连不上报Access denied或者Public Key Retrieval is not allowed。上面建用户时我特意写了IDENTIFIED WITH mysql_native_password就是为了兼容。如果你已经是 Canal 1.1.5 以上版本理论上也支持 caching_sha2但为了省事很多生产环境还是沿用 native 密码。注意这是权衡不是绝对要求。验证账号能不能用在另一台机器上执行mysql -h 127.0.0.1 -ucanal -p -P3306需要确认宿主机防火墙放行了 3306且 MySQL 的bind-address没有只绑 127.0.0.1。4. Canal Server 与 Adapter 配置实录4.1 部署 Canal Serverinstance 是关键到 GitHub 的 Release 页面下载canal.deployer-1.1.5.tar.gz解压后的目录结构很清晰bin放启动脚本conf放配置lib放依赖logs放日志。Canal Server 本身有一套全局配置conf/canal.properties但大多数情况下不需要改重点是conf/example/instance.properties这个文件定义了一个名为 example 的 canal instance它决定了要连哪个 MySQL、从哪里开始消费。需要改的核心配置如下# MySQL 主库地址 canal.instance.master.address127.0.0.1:3306 # 刚才创建的 MySQL 账号 canal.instance.dbUsernamecanal canal.instance.dbPassword你的强密码 # 连接编码 canal.instance.connectionCharsetUTF-8 # 如果 MySQL 开了 GTID这里必须打开 canal.instance.gtidontrue # 指定要订阅的库表正则写法 canal.instance.filter.regexmydb\\..*filter.regex的规则是库名\.表名支持正则。如果只想监听 user 和 order 两张表可以写成canal.instance.filter.regexmydb\\.user|mydb\\.order如果你没开 GTIDCanal 默认会从当前位点开始但也可以通过canal.instance.master.journal.name和canal.instance.master.position指定从哪个 binlog 文件的哪个位置开始消费对应我们刚才查到的SHOW MASTER STATUS结果。启动bin/startup.sh看日志tail -f logs/example/example.log如果看到类似dump address ...或者success to connect的字样说明 Canal 已经和 MySQL 正常握手了。接着你去业务表插入一条数据日志里会打印解析到的 CanalEntry这一步通了就说明 binlog 链路没问题。4.2 使用 Adapter 同步到 RedisCanal Adapter 需要单独下载canal.adapter-1.1.5.tar.gz里面同样有conf目录主配置是conf/application.yml。它负责定义数据源、Canal Server 地址、以及多个外部适配器。一个简化可用的配置长这样canal.conf: mode: tcp flatMessage: true consumerProperties: canal.tcp.server.host: 127.0.0.1:11111 canal.tcp.batch.size: 500 srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mydb?useSSLfalseuseUnicodetruecharacterEncodingUTF-8 username: canal password: 你的强密码 canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: logger - name: redis key: redis hosts: 127.0.0.1:6379 mode: single其中logger适配器是打印日志用的方便调试redis适配器会把事件写到 Redis。Redis 的mode有single、cluster、sentinel几种本地测试先 single 就行。真正决定表怎么映射成 Redis key 的是表映射文件放在conf/redis/目录下。假设用户表是user新建一个user.ymldataSourceKey: defaultDS destination: example groupId: g1 outerAdapterKey: redis redisMapping: key: user:{id} value: json targetTable: user这里key模板里的{id}会被替换成行数据里的 id 字段所以插入 id10 的用户后Redis 里会出现一个 key 为user:10的字符串值是整行数据的 JSON。注意targetTable和文件名、destination、groupId 都要对应上否则 Adapter 找不到映射关系。还有一点如果flatMessage: falseAdapter 收到的是 protobuf 格式需要额外配置 proto 反序列化没必要默认 true 就够了。4.3 使用 Adapter 同步到 ESES 适配器同样在主配置的outerAdapters里加一项- name: es hosts: http://127.0.0.1:9200 properties: mode: restES 的 mapping 文件放在conf/es/下比如user_index.ymldataSourceKey: defaultDS destination: example groupId: g1 outerAdapterKey: es esMapping: _index: user _type: _doc _id: id sql: select id, name, age, phone from user etlCondition: where id 0 commitBatch: 1000sql是 Adapter 用来做全量 ETL 和增量关联的查询语句。当 user 表有变化时Adapter 会拿变更行的 id 值带进去查出完整字段再写入 ES 的user索引。这里的_type在 ES 7.x 里统一是_docES 8 里 type 已经被移除了如果你用的是 ES 8建索引时不需要 type映射文件里也不要再写_type。如果表的数据需要关联其他表比如user关联user_ext的 bio 字段你的sql可以写成select u.id, u.name, u.age, e.bio from user u left join user_ext e on u.id e.user_idCanal Adapter 会在每次 user 表变更时拿变化的 id 执行这条 SQL把关联数据取出来拼成一个完整文档后写入 ES。这个能力非常实用相当于把宽表建索引的工作交给了同步层。启动 Adapterbin/startup.sh观察日志如果报连接错误多半是 ES hosts 地址不对或者 Adapter 版本和 ES 版本不兼容。我踩过的一个坑是 ES 7.16 之后强制要求Content-Type老版本 Adapter 用的 restlet 会报406 Not Acceptable升级 Adapter 到 1.1.5 基本能解决。4.4 自定义消费端把控制权拿回自己手里当 Adapter 满足不了需求时我一般会写一个独立的 Java 服务用 Canal Client 直接消费 binlog 事件。这种方式最核心的一点是你亲自掌控 ack 和 rollback。先引入依赖dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.5/version /dependency然后写一个循环拉取逻辑CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(mydb\\..*); connector.rollback(); while (true) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } try { for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! CanalEntry.EntryType.ROWDATA) { continue; } CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType rowChange.getEventType(); String tableName entry.getHeader().getTableName(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { String operation eventType.name(); // INSERT / UPDATE / DELETE ListCanalEntry.Column columns eventType CanalEntry.EventType.DELETE ? rowData.getBeforeColumnsList() : rowData.getAfterColumnsList(); MapString, Object data columns.stream().collect( Collectors.toMap(CanalEntry.Column::getName, c - c.getIsNull() ? null : c.getValue())); syncToRedisAndEs(tableName, operation, data); } } connector.ack(batchId); } catch (Exception e) { connector.rollback(batchId); // 这里要做好本地补偿日志至少记录一条失败消息 } }代码里几个关键点getWithoutAck(100)表示一次最多拉 100 条事件但不确认处理成功后调用ack(batchId)告诉 Canal 可以清掉这批位点处理异常时调用rollback(batchId)Canal 会把这一批重新投递。DELETE事件里afterColumns往往是空的所以要用beforeColumns拿主键否则你连删的是哪条数据都不知道。INSERT和UPDATE用afterColumns也就是变更后的新值。写 Redis 的时候我用 Jackson 把 Map 转成 JSON 后 set 进去。写 ES 时根据操作类型分别调用IndexRequest或DeleteRequest。更稳妥的做法是统一用 upsert更新不存在就插入存在就覆盖。5. 生产级避坑清单与稳定性兜底5.1 缓存更新顺序与数据回源的坑用 Canal 同步 Redis 的时候很多人会问直接把缓存删掉不就行了读的时候回源数据库。这种思路在一致性要求不高时没问题但有一个条件回源读到的是最新数据。如果 Canal 延迟了读请求先到发现缓存没有就去数据库查查到的是旧值因为 Canal 还没消费到最新 binlog然后这个旧值又被写回缓存等 Canal 消费到新值再更新缓存中间就会有一段旧值窗口。我处理这个问题的做法是Canal 同步到 Redis 时直接写新值而不是删 key。Canal 本身能拿到变更后的字段把它序列化后 set 进去读请求在 Canal 异常延迟时最多读到旧 JSON不会触发读了更旧的值回填的问题。另一种常见方案是延迟双删先删缓存间隔几百毫秒再删一次。但双删治标不治本间隔时间长了还是有窗口间隔短了删除顺序又无法保证。既然我们已经用了 binlog不如让缓存始终跟着数据库走。5.2 ES 同步幂等与并发写 ES 时最容易出问题的场景是并发更新。两个线程同时拿到一条数据的 binlog一个写旧版本一个写新版本如果顺序颠倒ES 里的最终值可能就是错的。ES 自带版本号机制可以在请求里带version参数用外部版本号来控制。Canal 解析出来的每条 binlog 不带业务版本但你可以用变更发生的时间戳ts作为版本写入时指定versionTypeexternalES 内部会维护一个_version只有新版本的 ts 更大才能写入成功。如果不引入版本号另一个办法是保证单条数据只被一个消费者处理。用 MQ 的话按主键 Hash 到固定分区这样同一条记录的 binlog 一定会被同一个消费者顺序处理天然规避了并发交错。批量更新时要用 BulkRequest把几百条写操作合成一个请求发过去吞吐量能提升很多。我一般每次攒 1000 条左右或者每隔一秒 flush 一次。攒太多会导致单次请求过大ES 内存压力高攒太少又浪费网络往返。5.3 积压、重复消费与位点管理Canal 通过 ack 机制保证至少一次投递所以消费端必须接受可能重复。第一次写缓存时可能服务刚起来或者 ES 临时不可用rollback 之后同一批数据会重新投递。所以你的消费逻辑必须是幂等的写 Redis 的 set 操作天然幂等写 ES 的 upsert 也是幂等但如果消费逻辑里有累加库存这种计算就不幂等需要额外做去重比如维护一个最近处理过的 eventId 集合。位点管理还有一个容易忽略的问题Canal Server 本身也会挂。如果配置了 zookeeperCanal 会把位点信息放 zk恢复后自动续传。如果没上 zk重启后有可能从当前 binlog 最新位置开始中间一段数据就丢了。我的建议是至少在单机测试时启动tsdbenable也就是让 Canal 自己保存位点表避免每次重启都丢数据。延迟监控方面Canal Server 内置了 metrics可以用 Prometheus 拉取如果上了 MQ看 MQ 堆积数更直观。没有监控系统的话我习惯写一个简单的心跳表业务侧每隔几秒更新一条heartbeat记录的某个字段Canal 消费到这条事件时记录时间戳和当前时间比一下就知道端到端延迟是几秒。5.4 DDL 与敏感字段binlog 里不只包含 DML还会包含 DDL。当某个表加了一列Canal 会向消费者推送一个 DDL 事件。Adapter 对这个事件基本是忽略的它不会自动帮你修改 Redis 结构或 ES mapping。所以如果业务表结构变了需要人工介入重建索引或者改 Adapter 的映射 SQL不能指望同步层自己搞定。还有敏感字段问题。如果你的 binlog 里有手机号、身份证这类数据全量镜像到 ES 会有数据合规风险。两个思路一是做字段裁剪ES 映射 SQL 里只 select 需要索引的字段二是在消费端做脱敏比如手机号只保留前三位后四位再写 ES或者加一层加密。这个必须在同步链路里提前设计好不然后面删数据很痛苦。对于 binlog 日志本身的安全日志文件保留天数要合理且存放目录的权限要收紧。MySQL 账号我们已经做了最小权限不需要再给业务开发开放 binlog 的本地文件读取权限。5.5 问题排查速查表现象可能原因排查命令或手段Canal 连接被拒账号权限不足、认证插件不符、3306未放开SHOW GRANTS FOR canal%;远程mysql -ucanal -p试连binlog 已开但抓不到事件filter 正则写错、表不在匹配范围看 example.log 里订阅的 Filter手动插入测试Adapter 连不上 Redishosts 配置错误、Redis 设置了密码但未配置 passwordapplication.yml 的 redis 节点补passwordES 写入报 406Adapter 版本和 ES 版本兼容问题升级 Adapter 1.1.5或换 ES 7.x同步延迟越来越大消费能力不足、下游慢查询、批量太小看 MQ 堆积开启 bulk增加消费线程UPDATE 拿不到新字段值binlog_row_imageMINIMAL改成 FULL 并重启 MySQL重新生成日志DELETE 事件没有数据错用了 afterColumns用 beforeColumns 拿主键6. 一个收尾的小建议这套方案做到最后我发现自己收获最大的不是把 Canal 配置调通了而是养成了一套验证同步链路的习惯。每次上线一个新的同步任务不管用 Adapter 还是自研 Client我第一步不是对着文档抄配置而是先在测试库手动插一条明显的数据比如把某个字段写成SN_TEST_001然后依次去看三个地方MySQL 的 binlog 里有没有这条变更、Canal 的日志里有没有解析出来、Redis/ES 里有没有出现对应记录。三步串通了再把这个测试数据清掉。这套人工观测链路比任何监控都直白出了问题一眼能看出断在哪个环节。还有一个细节想多说一句如果你也遇到 Redis 里同步出来的数据总是少字段先别怀疑 Canal回去看一眼 binlog_row_image 是不是 FULL。我在这上面浪费了整整一个下午当时只改了 binlog_formatROW以为万事大吉结果 UPDATE 事件里只有主键和修改列ES 和 Redis 拿不到完整字段匹配不上映射关系。这个配置位置在 MySQL 服务器不在 Canal排查方向错了就会一直绕圈。Canal 这套组合拳用顺了之后业务侧几乎无感数据从 MySQL 流向 Redis 和 ES 就像一条安静的水管。希望你也能把它接得稳稳的少踩几个我曾经踩过的坑。
Canal实战：MySQL binlog实时同步到Redis与ES的完整方案
最近帮一个订单系统接Canal用它监听MySQL的binlog日志把数据准实时同步到Redis和ES整个过程踩了不少坑。整理一下这整套方案的思路、配置细节和排障经验给后面接手类似业务数据库变一下Redis缓存和ES索引马上跟着变需求的朋友做个参考。先说清楚这套东西是什么Canal 是阿里巴巴开源的一个 MySQL binlog 增量订阅组件它把自己伪装成一个 MySQL 从库通过主从复制协议去拉取 binlog然后解析成结构化事件下游无论是写 Redis、ES还是丢进 MQ 里慢慢消费都很顺手。适合的场景很典型——缓存里的热点数据、搜索引擎里的商品或订单数据不能老靠定时任务刷全量也不能在业务代码里双写那就让 Canal 在中间做数据搬运工。这套方案适合谁后端开发、数据同步开发、搞中间件的人都能用上。即使你还没接触过 binlog看完也能照着搭起来知道每一步为什么这么配遇到问题往哪个方向查。1. 业务侧的真实痛点缓存和搜索为何总是慢半拍1.1 缓存失效与搜索索引更新的死结大多数业务系统里MySQL 是数据的最终归宿但 MySQL 扛不住高并发读所以前面挡一层 Redis而复杂查询、模糊搜索、聚合条件MySQL 的 like 和联合查询又很吃力于是又上 Elasticsearch。问题就出在这同样一份数据在 MySQL、Redis、ES 里各存了一份谁来保证它们之间的一致性我见过很多团队的做法是业务代码双写。更新数据库的时候顺手把 Redis 删掉、把 ES 更新一下。听起来没什么实际跑起来全是坑。第一双写导致业务逻辑里到处都是缓存代码和搜索同步代码换数据库或者换缓存的成本极高第二双写不是原子的数据库提交成功、ES 写入失败库存和搜索就永远不一致了第三为了补偿失败还得再加消息队列重试逻辑越来越复杂。另一种常用方案是定时任务扫表。每隔一分钟把变更的数据捞出来同步一次。这个方案在小数据量下能凑合但实时性太差而且每次全表扫描对 MySQL 压力很大扫到刚更新过的数据还好扫不到就会有窗口期。数据量一上来定时任务还没跑完业务数据又变了。Canal 能破这个局是因为它站在数据库外面通过 binlog 感知每一次行级变更把数据库自己产生的日志变成下游存储的更新指令。业务代码不用改一行缓存和搜索的同步完全剥离出去。1.2 binlog 为什么能当数据总线MySQL 本身就有一套日志机制叫 binlog二进制日志专门用来记录所有数据变更。主从复制就是靠它实现的主库把 binlog 发给从库从库拿到后重放就得到了和主库一样的数据。binlog 有三种格式STATEMENT、ROW、MIXED。STATEMENT 记录的是 SQL 原句比如UPDATE user SET name张三 WHERE id1从库拿到后重新执行一遍。ROW 记录的是每一行变更前后的完整字段值对 Canal 来说这才是最有价值的东西——它不需要知道你的 SQL 长什么样只需要拿到id1这行数据更新前是什么、更新后是什么就能直接把结果塞给 Redis 或 ES。所以 Canal 设计上做了一件很妙的事它把自己伪装成一个从库向 MySQL 发送 dump 请求MySQL 就把 binlog 推给它。Canal 拿到日志后解析成结构化消息下游想怎么消费怎么消费。这和我们直接监听主从复制的区别在于主从复制是把 binlog 应用到另一台 MySQL而 Canal 是把它变成事件流喂给异构存储。相当于把 MySQL 的日志能力扩展成了一根数据总线接多少个下游都行下游挂了也不影响 MySQL 本身。2. 整体方案设计先画一条数据流转链路2.1 模式对比Canal Adapter 还是自研 ClientCanal 官方除了 Server 本体还提供了一个 Adapter 组件可以理解成开箱即用的消费者它直接从 Canal Server 拿增量数据再通过配置把数据写到 Redis、ES、RDB 等目标端。使用 Canal Adapter 的好处很直接不用写代码。表结构定了以后写个映射配置文件启动 Adapter数据就开始同步了。尤其适合那种字段不复杂、SCHEMA 比较稳的表。但它的短板也很明显。Adapter 的同步逻辑是内置好的按表映射来做全字段覆盖。如果你想做字段裁剪、数据清洗、调用下游接口、做复杂幂等判断Adapter 就不太够用了。这种时候需要写一个自己的消费端使用 Canal 提供的 Java Client 从 Server 拉取数据自己解析 Entry再按业务规则去更新 Redis 和 ES。两种模式并不互斥。我的建议是快速验证阶段先上 Adapter跑通链路、观察效果如果后面出现复杂场景比如同一个表要同步成两种不同的 Redis 结构或者要把几条 binlog 聚合后一起写 ES就切到自研 Client。2.2 引入 MQ 与否资源如何权衡有一种常见架构是 Canal Server 把 binlog 直接投递到 Kafka 或 RocketMQ下游再各自消费。这样做有几个好处一是 Canal 和 Target 存储解耦Canal 只负责把事件写进 MQ写入 Redis/ES 的消费端可以独立扩缩容二是削峰填谷瞬间大批量更新时MQ 会帮下游把压力摊平避免 Redis 和 ES 被同时打爆三是 MQ 自带位点管理消费端重启后能接着上次的位置消费比起自己维护 offset 省事。但引入 MQ 也意味着多一个中间件要运维链路更长排查问题也更麻烦。数据量不大、对延迟容忍度较高、团队规模小的时候直接用 Adapter 或 Client 直连 Canal Server 完全够用。我在实际操作里一般这样判断单表日均变更量低于几十万直接用 Adapter 或简单 Client变更量很大或者下游消费逻辑很重就上 MQ。前者图省事后者图安稳没有绝对最优只有当前阶段合不合适。2.3 同步粒度与存储策略的划分不是所有表都要同步也不是所有表同步后 Redis 和 ES 都要一份。我习惯先把表按用途分三类。第一类热点查询型。比如商品基本信息、用户资料这些数据高频读、低频写适合同步到 Redis用 hash 或 string 缓存key 设计成表名:id。这类表不需要同步 ES。第二类搜索筛选型。比如订单表、商品 SPU 表查询条件多、需要分词、需要排序这类表必须同步到 ES。同步时可以只同步需要的字段通过 SQL 把关联表拼好再写入一个文档没必要把 MySQL 原表机械地镜像成一个 index。第三类宽表型。比如一张订单宽表既要做详情缓存又要做列表搜索那么 Redis 和 ES 都需要。同步的时候更新 Redis 一份完整 JSON更新 ES 一份文档模型两份数据的字段可以不同只要保证主键一致就行。把表分好类之后Canal 的订阅过滤也更好配。比如在 Canal Server 的 instance 里只订阅某个库的某几张表或者在 Client 里做正则过滤减少无效日志的解析开销。3. 环境准备MySQL 开启 binlog 的完整姿势3.1 检查状态与修改 MySQL 配置网上 MySQL 安装教程很多这里不重复我直接讲 binlog 相关配置。先确认当前实例是否已经开启 binlogSHOW VARIABLES LIKE log_bin;如果返回值是OFF就需要改配置文件。Linux 下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnfWindows 下是my.ini。在[mysqld]段下面加上[mysqld] server-id10 log-binmysql-bin binlog_formatROW binlog_row_imageFULL expire_logs_days7 gtid_modeON enforce_gtid_consistencyON逐行解释一下为什么这么配。server-id在复制协议里是实例的唯一标识Canal 伪装从库时也需要一个不一样的值不能和现存的主从拓扑里任何节点的 server-id 重复否则 MySQL 会踢掉其中一个连接。log-binmysql-bin指定日志文件前缀。binlog_formatROW是最关键的一项上面说过ROW 格式才有行级变更的字段值STATEMENT 让 Canal 没法高效构造目标数据。binlog_row_imageFULL很容易被忽略它表示在 ROW 格式下记录整行所有字段的镜像。如果改成 MINIMAL日志里只包含主键和发生变化的列对某些场景能省日志量但对同步 Redis 或 ES 来说我们往往需要完整的字段值所以才设成 FULL。expire_logs_days7控制日志保留天数避免磁盘被灌满。gtid_modeON和enforce_gtid_consistencyON是给 Canal 一个更稳定的位点管理方式后面配置 instance 时会用到。改完配置需要重启 MySQL。重启后再次查SHOW VARIABLES LIKE log_bin;确认是 ON然后执行SHOW MASTER STATUS;会看到类似mysql-bin.000001 | 154 | ...这样的输出这就是当前 binlog 文件和 position 点位。后面配置 Canal 时如果不想从最新开始消费就可以填写这个点位。注意不要手贱去服务器上直接rmbinlog 文件。如果日志占满磁盘用 MySQL 自带的清理命令比如PURGE BINARY LOGS TO mysql-bin.000100;或者通过expire_logs_days自动过期。3.2 创建最小权限的 Canal 专用账号Canal 需要连上 MySQL 拉取 binlog但它不应该有业务库的写权限。给它建一个专用账号权限收窄到最小集合CREATE USER canal% IDENTIFIED WITH mysql_native_password BY 你的强密码; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;SELECT权限是 Canal Adapter 做全量数据初始化ETL时要读取表数据用的普通增量同步其实只需要后面两个REPLICATION权限。REPLICATION SLAVE让 Canal 可以请求 binlogREPLICATION CLIENT让它可以查看 master 状态和 server 信息。如果你的 MySQL 是 8.0 以上默认认证插件是caching_sha2_password老版本 Canal 可能连不上报Access denied或者Public Key Retrieval is not allowed。上面建用户时我特意写了IDENTIFIED WITH mysql_native_password就是为了兼容。如果你已经是 Canal 1.1.5 以上版本理论上也支持 caching_sha2但为了省事很多生产环境还是沿用 native 密码。注意这是权衡不是绝对要求。验证账号能不能用在另一台机器上执行mysql -h 127.0.0.1 -ucanal -p -P3306需要确认宿主机防火墙放行了 3306且 MySQL 的bind-address没有只绑 127.0.0.1。4. Canal Server 与 Adapter 配置实录4.1 部署 Canal Serverinstance 是关键到 GitHub 的 Release 页面下载canal.deployer-1.1.5.tar.gz解压后的目录结构很清晰bin放启动脚本conf放配置lib放依赖logs放日志。Canal Server 本身有一套全局配置conf/canal.properties但大多数情况下不需要改重点是conf/example/instance.properties这个文件定义了一个名为 example 的 canal instance它决定了要连哪个 MySQL、从哪里开始消费。需要改的核心配置如下# MySQL 主库地址 canal.instance.master.address127.0.0.1:3306 # 刚才创建的 MySQL 账号 canal.instance.dbUsernamecanal canal.instance.dbPassword你的强密码 # 连接编码 canal.instance.connectionCharsetUTF-8 # 如果 MySQL 开了 GTID这里必须打开 canal.instance.gtidontrue # 指定要订阅的库表正则写法 canal.instance.filter.regexmydb\\..*filter.regex的规则是库名\.表名支持正则。如果只想监听 user 和 order 两张表可以写成canal.instance.filter.regexmydb\\.user|mydb\\.order如果你没开 GTIDCanal 默认会从当前位点开始但也可以通过canal.instance.master.journal.name和canal.instance.master.position指定从哪个 binlog 文件的哪个位置开始消费对应我们刚才查到的SHOW MASTER STATUS结果。启动bin/startup.sh看日志tail -f logs/example/example.log如果看到类似dump address ...或者success to connect的字样说明 Canal 已经和 MySQL 正常握手了。接着你去业务表插入一条数据日志里会打印解析到的 CanalEntry这一步通了就说明 binlog 链路没问题。4.2 使用 Adapter 同步到 RedisCanal Adapter 需要单独下载canal.adapter-1.1.5.tar.gz里面同样有conf目录主配置是conf/application.yml。它负责定义数据源、Canal Server 地址、以及多个外部适配器。一个简化可用的配置长这样canal.conf: mode: tcp flatMessage: true consumerProperties: canal.tcp.server.host: 127.0.0.1:11111 canal.tcp.batch.size: 500 srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mydb?useSSLfalseuseUnicodetruecharacterEncodingUTF-8 username: canal password: 你的强密码 canalAdapters: - instance: example groups: - groupId: g1 outerAdapters: - name: logger - name: redis key: redis hosts: 127.0.0.1:6379 mode: single其中logger适配器是打印日志用的方便调试redis适配器会把事件写到 Redis。Redis 的mode有single、cluster、sentinel几种本地测试先 single 就行。真正决定表怎么映射成 Redis key 的是表映射文件放在conf/redis/目录下。假设用户表是user新建一个user.ymldataSourceKey: defaultDS destination: example groupId: g1 outerAdapterKey: redis redisMapping: key: user:{id} value: json targetTable: user这里key模板里的{id}会被替换成行数据里的 id 字段所以插入 id10 的用户后Redis 里会出现一个 key 为user:10的字符串值是整行数据的 JSON。注意targetTable和文件名、destination、groupId 都要对应上否则 Adapter 找不到映射关系。还有一点如果flatMessage: falseAdapter 收到的是 protobuf 格式需要额外配置 proto 反序列化没必要默认 true 就够了。4.3 使用 Adapter 同步到 ESES 适配器同样在主配置的outerAdapters里加一项- name: es hosts: http://127.0.0.1:9200 properties: mode: restES 的 mapping 文件放在conf/es/下比如user_index.ymldataSourceKey: defaultDS destination: example groupId: g1 outerAdapterKey: es esMapping: _index: user _type: _doc _id: id sql: select id, name, age, phone from user etlCondition: where id 0 commitBatch: 1000sql是 Adapter 用来做全量 ETL 和增量关联的查询语句。当 user 表有变化时Adapter 会拿变更行的 id 值带进去查出完整字段再写入 ES 的user索引。这里的_type在 ES 7.x 里统一是_docES 8 里 type 已经被移除了如果你用的是 ES 8建索引时不需要 type映射文件里也不要再写_type。如果表的数据需要关联其他表比如user关联user_ext的 bio 字段你的sql可以写成select u.id, u.name, u.age, e.bio from user u left join user_ext e on u.id e.user_idCanal Adapter 会在每次 user 表变更时拿变化的 id 执行这条 SQL把关联数据取出来拼成一个完整文档后写入 ES。这个能力非常实用相当于把宽表建索引的工作交给了同步层。启动 Adapterbin/startup.sh观察日志如果报连接错误多半是 ES hosts 地址不对或者 Adapter 版本和 ES 版本不兼容。我踩过的一个坑是 ES 7.16 之后强制要求Content-Type老版本 Adapter 用的 restlet 会报406 Not Acceptable升级 Adapter 到 1.1.5 基本能解决。4.4 自定义消费端把控制权拿回自己手里当 Adapter 满足不了需求时我一般会写一个独立的 Java 服务用 Canal Client 直接消费 binlog 事件。这种方式最核心的一点是你亲自掌控 ack 和 rollback。先引入依赖dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.5/version /dependency然后写一个循环拉取逻辑CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, , ); connector.connect(); connector.subscribe(mydb\\..*); connector.rollback(); while (true) { Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { Thread.sleep(1000); continue; } try { for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! CanalEntry.EntryType.ROWDATA) { continue; } CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType rowChange.getEventType(); String tableName entry.getHeader().getTableName(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { String operation eventType.name(); // INSERT / UPDATE / DELETE ListCanalEntry.Column columns eventType CanalEntry.EventType.DELETE ? rowData.getBeforeColumnsList() : rowData.getAfterColumnsList(); MapString, Object data columns.stream().collect( Collectors.toMap(CanalEntry.Column::getName, c - c.getIsNull() ? null : c.getValue())); syncToRedisAndEs(tableName, operation, data); } } connector.ack(batchId); } catch (Exception e) { connector.rollback(batchId); // 这里要做好本地补偿日志至少记录一条失败消息 } }代码里几个关键点getWithoutAck(100)表示一次最多拉 100 条事件但不确认处理成功后调用ack(batchId)告诉 Canal 可以清掉这批位点处理异常时调用rollback(batchId)Canal 会把这一批重新投递。DELETE事件里afterColumns往往是空的所以要用beforeColumns拿主键否则你连删的是哪条数据都不知道。INSERT和UPDATE用afterColumns也就是变更后的新值。写 Redis 的时候我用 Jackson 把 Map 转成 JSON 后 set 进去。写 ES 时根据操作类型分别调用IndexRequest或DeleteRequest。更稳妥的做法是统一用 upsert更新不存在就插入存在就覆盖。5. 生产级避坑清单与稳定性兜底5.1 缓存更新顺序与数据回源的坑用 Canal 同步 Redis 的时候很多人会问直接把缓存删掉不就行了读的时候回源数据库。这种思路在一致性要求不高时没问题但有一个条件回源读到的是最新数据。如果 Canal 延迟了读请求先到发现缓存没有就去数据库查查到的是旧值因为 Canal 还没消费到最新 binlog然后这个旧值又被写回缓存等 Canal 消费到新值再更新缓存中间就会有一段旧值窗口。我处理这个问题的做法是Canal 同步到 Redis 时直接写新值而不是删 key。Canal 本身能拿到变更后的字段把它序列化后 set 进去读请求在 Canal 异常延迟时最多读到旧 JSON不会触发读了更旧的值回填的问题。另一种常见方案是延迟双删先删缓存间隔几百毫秒再删一次。但双删治标不治本间隔时间长了还是有窗口间隔短了删除顺序又无法保证。既然我们已经用了 binlog不如让缓存始终跟着数据库走。5.2 ES 同步幂等与并发写 ES 时最容易出问题的场景是并发更新。两个线程同时拿到一条数据的 binlog一个写旧版本一个写新版本如果顺序颠倒ES 里的最终值可能就是错的。ES 自带版本号机制可以在请求里带version参数用外部版本号来控制。Canal 解析出来的每条 binlog 不带业务版本但你可以用变更发生的时间戳ts作为版本写入时指定versionTypeexternalES 内部会维护一个_version只有新版本的 ts 更大才能写入成功。如果不引入版本号另一个办法是保证单条数据只被一个消费者处理。用 MQ 的话按主键 Hash 到固定分区这样同一条记录的 binlog 一定会被同一个消费者顺序处理天然规避了并发交错。批量更新时要用 BulkRequest把几百条写操作合成一个请求发过去吞吐量能提升很多。我一般每次攒 1000 条左右或者每隔一秒 flush 一次。攒太多会导致单次请求过大ES 内存压力高攒太少又浪费网络往返。5.3 积压、重复消费与位点管理Canal 通过 ack 机制保证至少一次投递所以消费端必须接受可能重复。第一次写缓存时可能服务刚起来或者 ES 临时不可用rollback 之后同一批数据会重新投递。所以你的消费逻辑必须是幂等的写 Redis 的 set 操作天然幂等写 ES 的 upsert 也是幂等但如果消费逻辑里有累加库存这种计算就不幂等需要额外做去重比如维护一个最近处理过的 eventId 集合。位点管理还有一个容易忽略的问题Canal Server 本身也会挂。如果配置了 zookeeperCanal 会把位点信息放 zk恢复后自动续传。如果没上 zk重启后有可能从当前 binlog 最新位置开始中间一段数据就丢了。我的建议是至少在单机测试时启动tsdbenable也就是让 Canal 自己保存位点表避免每次重启都丢数据。延迟监控方面Canal Server 内置了 metrics可以用 Prometheus 拉取如果上了 MQ看 MQ 堆积数更直观。没有监控系统的话我习惯写一个简单的心跳表业务侧每隔几秒更新一条heartbeat记录的某个字段Canal 消费到这条事件时记录时间戳和当前时间比一下就知道端到端延迟是几秒。5.4 DDL 与敏感字段binlog 里不只包含 DML还会包含 DDL。当某个表加了一列Canal 会向消费者推送一个 DDL 事件。Adapter 对这个事件基本是忽略的它不会自动帮你修改 Redis 结构或 ES mapping。所以如果业务表结构变了需要人工介入重建索引或者改 Adapter 的映射 SQL不能指望同步层自己搞定。还有敏感字段问题。如果你的 binlog 里有手机号、身份证这类数据全量镜像到 ES 会有数据合规风险。两个思路一是做字段裁剪ES 映射 SQL 里只 select 需要索引的字段二是在消费端做脱敏比如手机号只保留前三位后四位再写 ES或者加一层加密。这个必须在同步链路里提前设计好不然后面删数据很痛苦。对于 binlog 日志本身的安全日志文件保留天数要合理且存放目录的权限要收紧。MySQL 账号我们已经做了最小权限不需要再给业务开发开放 binlog 的本地文件读取权限。5.5 问题排查速查表现象可能原因排查命令或手段Canal 连接被拒账号权限不足、认证插件不符、3306未放开SHOW GRANTS FOR canal%;远程mysql -ucanal -p试连binlog 已开但抓不到事件filter 正则写错、表不在匹配范围看 example.log 里订阅的 Filter手动插入测试Adapter 连不上 Redishosts 配置错误、Redis 设置了密码但未配置 passwordapplication.yml 的 redis 节点补passwordES 写入报 406Adapter 版本和 ES 版本兼容问题升级 Adapter 1.1.5或换 ES 7.x同步延迟越来越大消费能力不足、下游慢查询、批量太小看 MQ 堆积开启 bulk增加消费线程UPDATE 拿不到新字段值binlog_row_imageMINIMAL改成 FULL 并重启 MySQL重新生成日志DELETE 事件没有数据错用了 afterColumns用 beforeColumns 拿主键6. 一个收尾的小建议这套方案做到最后我发现自己收获最大的不是把 Canal 配置调通了而是养成了一套验证同步链路的习惯。每次上线一个新的同步任务不管用 Adapter 还是自研 Client我第一步不是对着文档抄配置而是先在测试库手动插一条明显的数据比如把某个字段写成SN_TEST_001然后依次去看三个地方MySQL 的 binlog 里有没有这条变更、Canal 的日志里有没有解析出来、Redis/ES 里有没有出现对应记录。三步串通了再把这个测试数据清掉。这套人工观测链路比任何监控都直白出了问题一眼能看出断在哪个环节。还有一个细节想多说一句如果你也遇到 Redis 里同步出来的数据总是少字段先别怀疑 Canal回去看一眼 binlog_row_image 是不是 FULL。我在这上面浪费了整整一个下午当时只改了 binlog_formatROW以为万事大吉结果 UPDATE 事件里只有主键和修改列ES 和 Redis 拿不到完整字段匹配不上映射关系。这个配置位置在 MySQL 服务器不在 Canal排查方向错了就会一直绕圈。Canal 这套组合拳用顺了之后业务侧几乎无感数据从 MySQL 流向 Redis 和 ES 就像一条安静的水管。希望你也能把它接得稳稳的少踩几个我曾经踩过的坑。
