场景痛点：你的电商平台遭遇了“超卖危机”。MySQL 处理库存扣减，Redis 缓存商品详情页库存。当流量洪峰来临，因缓存库存未及时更新，10 台 iPhone 被抢成了 100 台 —— 数据不一致引发资损！

解决方案核心：MySQL Binlog + Canal + Redis，实现数据变更的毫秒级同步。

一、技术架构拆解（核心流程）

MySQL Server → 开启Binlog → Canal Server 监听 → Canal Client (Java) → Redis 执行更新

为何选 Binlog？

• MySQL 原生二进制日志，记录所有数据变更
• 高可靠性：主从复制的基石
• 低侵入：对业务代码零改造

二、实战五步走，代码级详解

1. 启用 MySQL Binlog（必须ROW模式）

# my.cnf 关键配置
[mysqld]
server_id        = 1
log_bin          = /var/lib/mysql/mysql-bin
binlog_format    = ROW        # 必须为ROW模式！
expire_logs_days = 3

2. 部署 Canal Server（1.1.7版本）

# 下载解压
wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz
tar -zxvf canal.deployer-1.1.7.tar.gz

# 配置 canal.properties
canal.instance.master.address = 127.0.0.1:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = StrongPassword!

3. Java 编写 Canal 客户端（监听变更）

// Spring Boot 示例 (依赖：com.alibaba.otter:canal.client)
CanalConnector connector = CanalConnectors.newClusterConnector(
  "127.0.0.1:11111", "example", "", "");

connector.connect();
connector.subscribe("db_order.t_product"); // 监控订单库商品表

while (running) {
  Message message = connector.getWithoutAck(100); // 批量获取
  for (CanalEntry.Entry entry : message.getEntries()) {
    if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
      processRowChange(entry.getStoreValue()); // 关键：处理数据变更
    }
  }
  connector.ack(message.getId()); // 确认消费
}

4. 解析 Binlog 并更新 Redis

// 解析产品表库存变更
private void processRowChange(ByteString data) throws Exception {
  CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(data);
  
  for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
    if (rowChange.getEventType() == CanalEntry.EventType.UPDATE) {
      Map<String, String> beforeMap = parseColumnsToMap(rowData.getBeforeColumnsList());
      Map<String, String> afterMap = parseColumnsToMap(rowData.getAfterColumnsList());
      
      // 核心：当库存(stock)变更时，更新Redis
      if (!beforeMap.get("stock").equals(afterMap.get("stock"))) {
        String redisKey = "product:" + afterMap.get("id") + ":stock";
        jedis.set(redisKey, afterMap.get("stock"));
        System.out.println("更新Redis库存: " + redisKey + "=" + afterMap.get("stock"));
      }
    }
  }
}

5. Redis 执行原子更新（Lua 保证一致性）

-- 库存扣减Lua脚本（避免超卖）
local key = KEYS[1]
local change = tonumber(ARGV[1])
local current = tonumber(redis.call('GET', key))

if current >= change then
  return redis.call('INCRBY', key, -change) -- 原子扣减
else
  return -1 -- 库存不足
end

三、性能压测结果（百万级数据验证）

四、避坑指南（血泪经验）

1. 幂等性设计：网络抖动可能导致消息重发，更新操作需支持重复执行

• 使用 Redis 的 SET 而非 INCR
• Lua 脚本内判断版本号

1. 断点续传：Canal 需记录消费位点 (position)，避免重启后数据丢失

connector.ack(batchId); // 消费成功后确认 connector.rollback(batchId); // 失败时回滚

1. 监控三件套：

• Canal 堆积告警 (canal.getWithoutAck.exception)
• Redis 内存/网络监控
• MySQL Binlog 延迟检测 (SHOW SLAVE STATUS)

五、进阶优化策略

• 并行消费：分库分表场景下，按 instance 分区并行处理
• 压缩传输：开启 Canal 的 zstd 压缩，降低网络开销
• 批量提交：合并多个更新事件，减少 Redis 请求次数

// 每100条更新批量提交一次 
Pipeline pipeline = jedis.pipelined(); 
for (int i = 0; i < 100; i++) { 
  pipeline.set(key[i], value[i]); 
} 
pipeline.sync();

结语：技术选型对比

技术本质：Binlog 是 MySQL 的数据脉搏，Canal 是实时听诊器，Redis 则是系统的缓存心脏。精准同步，方能打造无懈可击的高并发系统。

立刻行动：

1. 在测试环境部署 Canal（30分钟）
2. 用文中的 Java 代码监听商品表变更
3. 压测验证 Redis 库存与 MySQL 的差异

技术没有银弹，但 Binlog 同步是当前 MySQL→Redis 实时同步的最优解。