一、全局视角：从数据变更到业务消费的完整链路
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌───────────────┐ ┌──────────────┐
│ 业务代码 │ │ MySQL │ │ CDC中间件 │ │ Kafka │ │ 应用消费端 │
│ │ │ │ │(Canal等) │ │ │ │ │
│ INSERT/ │───→│ InnoDB │───→│伪装Sla ve │───→│ binlog topic │───→│ 框架路由 │
│ UPDATE/ │ │ +Binlog │ │读取解析 │ │ 持久化存储 │ │ +TableObserver│
│ DELETE │ │ │ │ │ │ │ │ │
└──────────┘ └──────────┘ └──────────┘ └───────────────┘ └──────────────┘
│
▼
┌──────────────┐
│ 业务逻辑 │
│ (推送/同步等) │
└──────────────┘
将这条链路进行拆分，可以清晰地划分为五个关键阶段。每个阶段的技术选型与设计思路都各有不同，接下来我们将逐一进行深入剖析。
二、阶段一：MySQL Binlog 生成
2.1 Binlog 究竟是什么
Binlog，全称为Binary Log，是MySQL Server层特有的一种二进制日志文件，专门用于记录所有对数据库数据产生变更的操作。该日志与事务提交紧密绑定——当事务提交时，变更数据会同时被写入InnoDB的redo log和Binlog。这两类日志各有分工：redo log主要负责数据恢复，而Binlog则专注于记录操作历史，两者协同工作，保障数据安全。
2.2 事务提交与 Binlog 写入的关联机制（两阶段提交）
事务执行过程：
┌─────────────────────────────────────────────────┐
│ BEGIN │
│ │
│ INSERT INTO xxx (...) │
│ → InnoDB 写入 redo log (prepare 状态) │
│ │
│ COMMIT │
│ → 写入 Binlog │ ← binlog 在这里产生
│ → InnoDB redo log 改为 commit 状态 │
│ → 事务完成 │
└─────────────────────────────────────────────────┘
这里有一个关键点需要牢记：只有事务成功提交后，binlog 才会被写入。这意味着CDC（数据变更捕获）所捕获到的全都是已提交的、具备确定性的变更，绝不会捕获到那些被回滚的数据。这一点，为数据消费的可靠性提供了坚实保障。
2.3 Binlog 的三种记录格式
| 格式 | 记录内容 | CDC可用性 |
|---|---|---|
| STATEMENT | SQL 原文 | ❌ 不可用（NOW()等函数不确定） |
| ROW | 每行数据变更前后的值 | ✅ 必须使用 |
| MIXED | MySQL 自动选择 | ⚠️ 不推荐 |
在CDC场景下，必须将 binlog_format 设置为 ROW。原因很简单，业务系统需要明确知道具体是哪一行、哪个字段，从什么值变成了什么值。STATEMENT格式记录的是SQL原文，其中可能包含 NOW() 这类不确定函数，导致数据回放结果不一致，因此无法用于CDC场景。
2.4 ROW 格式下一条 Binlog 所包含的详细信息
┌─────────────────────────────────────────┐
│ Binlog Event (ROW格式) │
│ │
│ ├── database: "xxx" │ 哪个数据库
│ ├── table: "xxx" │ 哪张表
│ ├── type: "INSERT" │ 什么操作
│ ├── ts: 1707552600000 │ 操作时间戳
│ ├── pkNames: ["id"] │ 主键列名
│ ├── isDdl: false │ 是否DDL
│ ├── data: [ │ 变更后的数据（完整行）
│ │ { │
│ │ "id": 12345, │
│ │ "member_id": 213681, │
│ │ "warehouse_id": 100, │
│ │ "order_code": "CK20250210001", │
│ │ "trans_qty": 5 │
│ │ } │
│ │ ] │
│ └── old: null │ 变更前的数据（INSERT时为null）
└─────────────────────────────────────────┘
三、阶段二：CDC 中间件捕获 Binlog
3.1 核心运作原理——伪装成 Sla ve
MySQL的主从复制协议是公开的，这为CDC中间件提供了可乘之机。它利用该协议，向MySQL Master注册自身为一个Sla ve节点，随后MySQL便会主动将Binlog推送过来。在MySQL Master的视角来看，它仅仅是多了一个Sla ve在同步数据而已。
MySQL Master 视角："我有3个 Sla ve 在同步数据" Sla ve 1: 真正的从库（做读写分离） Sla ve 2: 真正的从库（做灾备） Sla ve 3: CDC中间件（伪装的，实际只是读binlog不回放SQL）
3.2 CDC 中间件的详细工作流程
┌──────────────────────────────────────────────────────────────┐
│ CDC 中间件内部流程 │
│ │
│ 1. 启动时： │
│ → 连接 MySQL Master │
│ → 发送 DUMP 命令（携带上次读取的 binlog 位置） │
│ → MySQL 开始推送 binlog 事件流 │
│ │
│ 2. 运行时（持续循环）： │
│ → 接收二进制 binlog 事件 │
│ → 解析二进制协议为结构化数据 │
│ → 按 database.table 过滤（只处理配置的表） │
│ → 序列化为 JSON │
│ → 发送到 Kafka topic │
│ → 记录消费位点（下次重启从这里继续） │
│ │
│ 3. 位点管理： │
│ → 记录 (binlog文件名, offset) 或 GTID │
│ → 确保不丢不重（至少不丢） │
└──────────────────────────────────────────────────────────────┘
3.3 为何选择 Kafka 而非直接回调应用
| 方案 | 优点 | 缺点 |
|---|---|---|
| CDC → 直接调用应用 | 链路简单，延迟低 | 应用重启时丢消息；无法回溯；无法多消费者 |
| CDC → Kafka → 应用 | 消息持久化；支持多消费者；支持回溯重放；削峰 | 链路多一环，延迟略高(通常仍在秒级内) |
生产环境中几乎毫无例外地选择 Kafka 作为中间层。别看多了一环，但它带来的消息持久化、回溯重放以及多消费者能力，是直连方案所无法比拟的。
四、阶段三：应用端 Binlog 消费框架设计
这套框架的核心诉求其实非常明确：让开发者只需关注业务逻辑，无需关心Kafka消费、消息解析、路由分发等基础设施代码。简而言之，就是避免在业务代码中混杂过多的非业务性内容。
4.1 框架整体设计（分层架构）
┌─────────────────────────────────────────────────────────────┐
│ 开发者层（你写的代码） │
│ │
│ BinlogConsumerConfig (注册Bean) │
│ StockTransactionBinlogConsumer (实现业务逻辑) │
├──────────────────────────────────────────────────────────────┤
│ 框架路由层 │
│ │
│ DefaultBinlogObserver (Kafka消费 + 表名路由) │
│ TableObserver (基类，提供线程池和生命周期管理) │
├──────────────────────────────────────────────────────────────┤
│ 框架启动层 │
│ │
│ GlobalConfig / DefaultBinlogConfig (自动配置 + Bean收集) │
│ BinlogObserverFactory (工厂模式，创建Observer实例) │
├──────────────────────────────────────────────────────────────┤
│ 基础设施层 │
│ │
│ Spring Kafka (ConcurrentMessageListenerContainer) │
│ xxxBinlogProperties (配置映射) │
│ BinlogWorkQueue (内部工作队列) │
└─────────────────────────────────────────────────────────────┘
4.2 核心设计模式应用
模式一：观察者模式（Observer Pattern）
- Subject（被观察者）: DefaultBinlogObserver，负责监听Kafka消息
- Observer（观察者）: TableObserver 的子类们，负责处理特定表的变更事件
DefaultBinlogObserver
(监听binlog Kafka topic)
│
┌──────────────┼──────────────┐
│ │ │
▼ ▼ ▼
xxxBarcode xxxTransaction xxxStock
Consumer BinlogConsumer OcoLogConsumer
模式二：模板方法模式（Template Method Pattern）
// TableObserver 定义了处理骨架abstract class TableObserver { // 模板方法：框架调用 public void post(List binlogs) { // 1. 提交到线程池（框架控制） // 2. 调用 postToConsumer() （框架控制） // 3. 最终调用 consumer() （子类实现） } // 抽象方法：子类必须实现 public abstract void consumer(List binlogs); public abstract String getSubTableName();} 开发者只需实现 consumer() 和 getSubTableName() 这两个方法，框架将负责整个调度骨架。这无疑是模板方法模式最典型的应用场景——将不变的部分封装起来，让可变的部分留待开发者实现。
模式三：工厂模式（Factory Pattern）
// BinlogObserverFactory 根据配置创建不同类型的 Observerclass BinlogObserverFactory { void createBinlogConsumer(properties); // 普通binlog消费 void createRdbBinlogObserver(properties); // 同步到RDB void createRedisBinlogObserver(properties); // 同步到Redis void createEventBridgeBinlogObserver(properties); // EventBridge}模式四：策略模式（Strategy Pattern）
不同的 TableObserver 子类代表了不同的处理策略。框架通过 getSubTableName() 方法的返回值来决定使用哪个具体的策略。
4.3 启动初始化时序
Spring Boot 启动 │ ▼┌─── spring.factories ──→ 注册 GlobalConfig Bean│├─── 扫描 @Configuration ──→ BinlogConsumerConfig│ ││ ├── @Bean xxxConsumer()│ ├── @Bean sendxxxOcoLogConsumer()│ ├── @Bean xxxTransactionBinlogConsumer() │ └── ...│├─── Spring 容器初始化完成│▼ContextRefreshedEvent 事件发布 │ ▼DefaultBinlogConfig.onApplicationEvent() 触发 │ ├── 1. 读取配置：xxx.binlog[0].tables = "xxx.xxx,xxx.xxx" │ ├── 2. 从 Spring 容器获取所有 TableObserver Bean │ → xxxConsumer.getSubTableName() = "xxx.xxx" │ → xxxTransactionBinlogConsumer.getSubTableName() = "xxx.xxx" │ → ... │ ├── 3. 构建路由表 Map<表名, List> │ "xxx.xxx" → [xxxBarcodeConsumer] │ ├── 4. BinlogObserverFactory.createBinlogConsumer() │ → new DefaultBinlogObserver(properties) │ → 将路由表注入到 DefaultBinlogObserver.subObserver │ → 创建 Kafka ConcurrentMessageListenerContainer │ (监听 topic: binlog-xxxx, groupId: xxxx) │ └── 5. start() → Kafka Consumer 开始 poll 循环 → 各 TableObserver 内部线程池启动
4.4 运行时消息消费时序
Kafka topic: binlog-xxx 收到一条消息 │ │ 消息内容 (JSON): │ { │ "database": "xxx", │ "table": "xxx", │ "type": "INSERT", │ "data": [{"member_id": 213681, "warehouse_id": 100, ...}], │ "old": null │ } │ ▼DefaultBinlogObserver.consumer(ConsumerRecords) │ ├── 1. 遍历 records，取 record.value() │ ├── 2. JSON 反序列化 → Binlog 对象 │ binlog.getDatabase() = "xxx" │ binlog.getTable() = "xxx" │ binlog.getType() = "INSERT" │ binlog.getData() = [{...}] │ ├── 3. 拼接路由 key = "xxx" + "." + "xxx" │ = "xxx.xxx" │ ├── 4. 查询路由表 │ subObserver.get("xxx.xxx") │ → [xxxTransactionBinlogConsumer] │ └── 5. 调用 observer.post(binlogList) │ ▼ TableObserver.post(List) │ ├── 提交到内部线程池 executor.submit(...) │ ▼ (线程池中异步执行) postToConsumer(binlogList, database, table) │ ├── 记录 metric（消费耗时、吞吐量等） │ ▼ consumer(binlogList) │ ├── 从 binlog.getData().get(0) 取行数据 ├── 提取 member_id, warehouse_id 五、关键设计点深度解析
5.1 线程模型设计
┌──────────────────────────────────────────────────────┐│ DefaultBinlogObserver ││ ││ Kafka Consumer 线程 (Spring Kafka管理) ││ → poll() 拉取消息 ││ → 解析 + 路由 ││ → 调用 TableObserver.post() ││ │├──────────────────────────────────────────────────────┤│ TableObserver-A 内部线程池 ││ → 异步执行 consumer() 方法 ││ │├──────────────────────────────────────────────────────┤│ TableObserver-B 内部线程池 ││ → 异步执行 consumer() 方法 │└──────────────────────────────────────────────────────┘
这种设计的好处非常明显：Kafka Consumer 线程不会被业务逻辑阻塞，各个 TableObserver 之间实现了线程隔离，互不影响。一个 Observer 处理变慢，不会拖累其他 Observer——这正是此设计的关键所在。
5.2 内部队列背压机制（BinlogWorkQueue）
// BinlogWorkQueue 继承 LinkedBlockingQueue// 重写了 offer 方法public boolean offer(T item) { if (remainingCapacity() == 0) { log.warn("Binlog队列已满，消息将被丢弃..."); // 队列满时打印告警日志 } return super.offer(item);}这是一种有损降级策略。当消费速度跟不上生产速度时，队列满后便丢弃新消息并打印告警，而不是阻塞 Kafka Consumer 线程，从而避免触发消费组 Rebalance。这个取舍，在线上环境中显得尤为重要。
5.3 生命周期管理
// TableObserver 实现了 DisposableBeaninterface DisposableBean { void destroy() throws Exception;}// destroy() 被调用时：// 1. isRunning = false （停止接收新任务）// 2. executor.shutdown() （等待已提交任务执行完）// 确保优雅停机，不丢正在处理的消息5.4 配置与路由的匹配规则
jsh: binlog: - tables: xxx.xxx1,xxx.xxx2// 匹配规则：// 1. Kafka 消息中的 database.table 必须在 yml 的 tables 列表中// 2. 且必须有对应的 TableObserver Bean 的 getSubTableName() 返回值与之匹配// 两个条件同时满足，消息才会被路由到对应的 Observer// 如果 tables 中有但没有对应 Observer → 消息被忽略（无报错）// 如果有 Observer 但 tables 中没配 → 该 Observer 永远不会收到消息
六、通用 CDC 消费框架示例代码
下面来看一个实际可用的简化版实现，以便完整展示框架的设计思路。
6.1 实体定义
package com.example.binlog.entity;import ja va.io.Serializable;import ja va.util.List;import ja va.util.Map;/** * Binlog事件实体. * 对应CDC中间件解析后的一条binlog记录. */public class BinlogEvent implements Serializable { /** * 数据库名. */ private String database; /** * 表名. */ private String table; /** * 操作类型：INSERT / UPDATE / DELETE. */ private String type; /** * 事件时间戳（毫秒）. */ private Long timestamp; /** * 主键列名列表. */ private List pkNames; /** * 变更后的行数据（完整行）. * INSERT: 新插入的数据 * UPDATE: 更新后的完整数据 * DELETE: 被删除的数据 */ private List> data; /** * 变更前的数据（仅UPDATE时有值，且只含被修改的字段）. */ private List> old; // getter/setter public String getDatabase() { return database; } public void setDatabase(String database) { this.database = database; } public String getTable() { return table; } public void setTable(String table) { this.table = table; } public String getType() { return type; } public void setType(String type) { this.type = type; } public Long getTimestamp() { return timestamp; } public void setTimestamp(Long timestamp) { this.timestamp = timestamp; } public List getPkNames() { return pkNames; } public void setPkNames(List pkNames) { this.pkNames = pkNames; } public List> getData() { return data; } public void setData(List> data) { this.data = data; } public List> getOld() { return old; } public void setOld(List> old) { this.old = old; } /** * 获取路由键（database.table）. */ public String getRouteKey() { return database + "." + table; } @Override public String toString() { return "BinlogEvent{database='" + database + "', table='" + table + "', type='" + type + "', dataSize=" + (data == null ? 0 : data.size()) + "}"; }} 6.2 操作类型枚举
package com.example.binlog.enums;/** * Binlog操作类型枚举. */public enum BinlogOperationType { INSERT("INSERT", "插入"), UPDATE("UPDATE", "更新"), DELETE("DELETE", "删除"); private final String code; private final String desc; BinlogOperationType(String code, String desc) { this.code = code; this.desc = desc; } public String getCode() { return code; } public String getDesc() { return desc; } public static BinlogOperationType fromCode(String code) { for (BinlogOperationType type : values()) { if (type.code.equalsIgnoreCase(code)) { return type; } } return null; }}6.3 表观察者抽象基类（模板方法模式）
package com.example.binlog.observer;import com.example.binlog.entity.BinlogEvent;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import ja va.util.List;import ja va.util.concurrent.*;/** * 表观察者抽象基类. * 使用模板方法模式：框架控制处理骨架，子类实现具体业务逻辑. * * 设计要点： * 1. 每个Observer有独立线程池，互不影响 * 2. 内部使用有界队列，防止OOM * 3. 支持优雅停机 */public abstract class AbstractTableObserver { private static final Logger log = LoggerFactory.getLogger(AbstractTableObserver.class); private final ExecutorService executor; private volatile boolean isRunning = true; private final String observerName; /** * 构造函数. * * @param queueSize 内部工作队列大小，控制背压。null时使用默认值1024. */ public AbstractTableObserver(Integer queueSize) { int capacity = (queueSize != null && queueSize > 0) ? queueSize : 1024; this.observerName = this.getClass().getSimpleName(); // 使用有界队列的线程池： // 核心线程=1（单线程顺序消费，保证同一表的处理顺序） // 最大线程=1 // 队列满时使用 CallerRunsPolicy（由调用线程执行，形成背压） BlockingQueue workQueue = new LinkedBlockingQueue<>(capacity); this.executor = new ThreadPoolExecutor( 1, 1, 60L, TimeUnit.SECONDS, workQueue, r -> new Thread(r, "binlog-" + observerName), new ThreadPoolExecutor.CallerRunsPolicy() ); } /** * 框架调用入口（模板方法）. * 将binlog事件提交到内部线程池异步处理. */ public final void post(List events) { if (!isRunning) { log.warn("[{}] Observer已停止，丢弃消息, size: {}", observerName, events.size()); return; } executor.submit(() -> { try { long startTime = System.currentTimeMillis(); // 调用子类实现的业务方法 consume(events); long cost = System.currentTimeMillis() - startTime; log.debug("[{}] 消费完成, size: {}, cost: {}ms", observerName, events.size(), cost); } catch (Exception e) { log.error("[{}] 消费异常", observerName, e); } }); } /** * 业务处理方法（子类必须实现）. * 框架保证：同一个Observer内的consume调用是串行的（单线程池）. * * @param events 本批次的binlog事件列表 */ public abstract void consume(List events); /** * 返回监听的表名（database.table格式）. * 框架根据此值做路由匹配. * * @return 例如 "my_db.order_table" */ public abstract String getSubscribeTable(); /** * 优雅停机. */ public void shutdown() { isRunning = false; executor.shutdown(); try { // 等待已提交任务完成 if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { executor.shutdownNow(); log.warn("[{}] 线程池强制关闭", observerName); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } log.info("[{}] Observer已停止", observerName); }} 6.4 Binlog 路由分发器（核心调度）
package com.example.binlog.dispatcher;import com.alibaba.fastjson.JSON;import com.example.binlog.entity.BinlogEvent;import com.example.binlog.observer.AbstractTableObserver;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import ja va.util.*;import ja va.util.concurrent.ConcurrentHashMap;/** * Binlog消息路由分发器. * 职责：消费Kafka消息 → 反序列化 → 按表名路由到对应Observer. * * 设计模式： * - 观察者模式：一个分发器对应多个Observer * - 策略模式：不同表使用不同的处理策略（Observer） */public class BinlogDispatcher { private static final Logger log = LoggerFactory.getLogger(BinlogDispatcher.class); /** * 路由表：表名 → Observer列表. * 一张表可以有多个Observer（广播模式）. */ private final Map> routeTable = new ConcurrentHashMap<>(); /** * 注册Observer到路由表. */ public void registerObserver(AbstractTableObserver observer) { String tableName = observer.getSubscribeTable(); routeTable.computeIfAbsent(tableName, k -> new ArrayList<>()).add(observer); log.info("注册BinlogObserver: {} → {}", tableName, observer.getClass().getSimpleName()); } /** * 消费Kafka消息并路由（由Spring Kafka Listener调用）. */ public void onMessage(ConsumerRecords records) { // 按表名分组，批量投递（减少线程池提交次数） // 按表名分组，批量投递（减少线程池提交次数） Map> groupedEvents = new HashMap<>(); for (ConsumerRecord record : records) { try { BinlogEvent event = JSON.parseObject(record.value(), BinlogEvent.class); if (event == null || event.getData() == null || event.getData().isEmpty()) { continue; } String routeKey = event.getRouteKey(); // 只处理路由表中有注册的表 if (routeTable.containsKey(routeKey)) { groupedEvents.computeIfAbsent(routeKey, k -> new ArrayList<>()).add(event); } } catch (Exception e) { log.warn("Binlog消息解析失败, offset: {}, value: {}", record.offset(), record.value(), e); } } // 批量分发到各Observer for (Map.Entry> entry : groupedEvents.entrySet()) { String tableName = entry.getKey(); List events = entry.getValue(); List observers = routeTable.get(tableName); if (observers != null) { for (AbstractTableObserver observer : observers) { observer.post(events); } } } } /** * 关闭所有Observer. */ public void shutdown() { for (List observers : routeTable.values()) { for (AbstractTableObserver observer : observers) { observer.shutdown(); } } log.info("所有BinlogObserver已关闭"); } /** * 获取路由表信息（用于监控和调试）. */ public Map> getRouteInfo() { Map> info = new HashMap<>(); for (Map.Entry> entry : routeTable.entrySet()) { List observerNames = new ArrayList<>(); for (AbstractTableObserver observer : entry.getValue()) { observerNames.add(observer.getClass().getSimpleName()); } info.put(entry.getKey(), observerNames); } return info; }} 6.5 自动配置类（Spring Boot Starter 模式）
package com.example.binlog.config;import com.example.binlog.dispatcher.BinlogDispatcher;import com.example.binlog.observer.AbstractTableObserver;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.ApplicationContext;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.core.ConsumerFactory;import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import org.springframework.kafka.listener.ContainerProperties;import org.springframework.kafka.listener.MessageListener;import ja vax.annotation.PostConstruct;import ja vax.annotation.PreDestroy;import ja va.util.HashMap;import ja va.util.List;import ja va.util.Map;/** * Binlog CDC 自动配置. * 负责：收集所有TableObserver → 注册到Dispatcher → 启动Kafka Consumer. */@Configurationpublic class BinlogAutoConfiguration { private static final Logger log = LoggerFactory.getLogger(BinlogAutoConfiguration.class); @Value("${binlog.kafka.bootstrap-servers}") private String bootstrapServers; @Value("${binlog.kafka.topic}") private String binlogTopic; @Value("${binlog.kafka.group-id}") private String groupId; @Value("${binlog.kafka.max-poll-records:500}") private Integer maxPollRecords; @Value("#{'${binlog.tables}'.split(',')}") private List subscribedTables; private final ApplicationContext applicationContext; private ConcurrentMessageListenerContainer container; private BinlogDispatcher dispatcher; public BinlogAutoConfiguration(ApplicationContext applicationContext) { this.applicationContext = applicationContext; } @Bean public BinlogDispatcher binlogDispatcher() { return new BinlogDispatcher(); } @PostConstruct public void init() { dispatcher = binlogDispatcher(); // 1. 收集所有 AbstractTableObserver Bean Map observerBeans = applicationContext.getBeansOfType(AbstractTableObserver.class); log.info("发现 {} 个BinlogObserver Bean", observerBeans.size()); // 2. 注册到Dispatcher（只注册在tables配置中声明的表） for (AbstractTableObserver observer : observerBeans.values()) { String tableName = observer.getSubscribeTable(); if (subscribedTables.contains(tableName)) { dispatcher.registerObserver(observer); } else { log.warn("Observer [{}] 的表 [{}] 未在binlog.tables配置中声明，跳过注册", observer.getClass().getSimpleName(), tableName); } } // 3. 创建并启动 Kafka Consumer startKafkaConsumer(); log.info("Binlog CDC 启动完成, topic: {}, tables: {}", binlogTopic, subscribedTables); } private void startKafkaConsumer() { // 构建消费者配置 Map consumerProps = new HashMap<>(); consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps); // 创建监听容器 ContainerProperties containerProps = new ContainerProperties(binlogTopic); containerProps.setMessageListener( (MessageListener) record -> { // 单条消息模式，实际框架中可能使用批量模式 // 这里简化为逐条处理 } ); // 使用批量监听模式 containerProps.setBatchListener(true); containerProps.setMessageListener( (org.springframework.kafka.listener.BatchMessageListener) records -> dispatcher.onMessage(records) ); container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps); container.setConcurrency(1); // 单线程消费，保证分区内顺序 container.start(); log.info("Kafka Consumer 已启动, topic: {}, groupId: {}", binlogTopic, groupId); } @PreDestroy public void destroy() { if (container != null) { container.stop(); log.info("Kafka Consumer 已停止"); } if (dispatcher != null) { dispatcher.shutdown(); } }} 6.6 使用示例：订单表变更监听
package com.example.binlog.demo;import com.example.binlog.entity.BinlogEvent;import com.example.binlog.enums.BinlogOperationType;import com.example.binlog.observer.AbstractTableObserver;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import ja va.util.List;import ja va.util.Map;/** * 订单表Binlog消费者. * 监听 order_db.t_order 表的变更，执行相应业务逻辑. * * 示例场景： * - 订单状态变更时通知物流系统 * - 订单金额变更时通知财务系统 * - 订单取消时通知库存系统释放库存 */@Componentpublic class OrderBinlogObserver extends AbstractTableObserver { private static final Logger log = LoggerFactory.getLogger(OrderBinlogObserver.class); public OrderBinlogObserver() { // 队列大小512，超过后由调用线程执行（背压） super(512); } @Override public void consume(List events) { for (BinlogEvent event : events) { try { processSingleEvent(event); } catch (Exception e) { // 单条异常不影响整批处理 log.error("订单binlog处理异常, event: {}", event, e); } } } private void processSingleEvent(BinlogEvent event) { Map data = event.getData().get(0); BinlogOperationType opType = BinlogOperationType.fromCode(event.getType()); String orderId = getStringValue(data, "order_id"); String status = getStringValue(data, "status"); Integer userId = getIntValue(data, "user_id"); if (orderId == null) { log.warn("订单binlog跳过：order_id为空, data: {}", data); return; } switch (opType) { case INSERT: handleOrderCreated(orderId, userId, data); break; case UPDATE: handleOrderUpdated(orderId, status, event.getOld(), data); break; case DELETE: handleOrderDeleted(orderId, data); break; default: log.warn("未知操作类型: {}", event.getType()); } } /** * 处理订单创建. */ private void handleOrderCreated(String orderId, Integer userId, Map data) { log.info("订单创建: orderId={}, userId={}", orderId, userId); // 业务逻辑：发送欢迎通知、初始化物流轨迹等 } /** * 处理订单更新. * 通过对比 old 和 data 判断具体哪些字段变了. */ private void handleOrderUpdated(String orderId, String newStatus, List> old, Map data) { // 判断状态是否变更 if (old != null && !old.isEmpty() && old.get(0).containsKey("status")) { String oldStatus = old.get(0).get("status").toString(); log.info("订单状态变更: orderId={}, {} → {}", orderId, oldStatus, newStatus); // 状态变为已支付：通知仓库发货 if ("PAID".equals(newStatus)) { notifyWarehouseToShip(orderId, data); } // 状态变为已取消：通知库存释放 if ("CANCELLED".equals(newStatus)) { notifyStockRelease(orderId, data); } } // 判断金额是否变更 if (old != null && !old.isEmpty() && old.get(0).containsKey("total_amount")) { Object oldAmount = old.get(0).get("total_amount"); Object newAmount = data.get("total_amount"); log.info("订单金额变更: orderId={}, {} → {}", orderId, oldAmount, newAmount); notifyFinanceSystem(orderId, newAmount); } } /** * 处理订单删除. */ private void handleOrderDeleted(String orderId, Map data) { log.info("订单删除: orderId={}", orderId); // 业务逻辑：归档、清理关联数据等 } // ----- 下游通知方法 ----- private void notifyWarehouseToShip(String orderId, Map data) { log.info("通知仓库发货: orderId={}", orderId); } private void notifyStockRelease(String orderId, Map data) { log.info("通知库存释放: orderId={}", orderId); } private void notifyFinanceSystem(String orderId, Object amount) { log.info("通知财务系统: orderId={}, amount={}", orderId, amount); } // ----- 工具方法 ----- private String getStringValue(Map data, String key) { Object value = data.get(key); return value != null ? value.toString() : null; } private Integer getIntValue(Map data, String key) { Object value = data.get(key); if (value == null) return null; return Integer.valueOf(value.toString()); } @Override public String getSubscribeTable() { return "order_db.t_order"; }} 6.7 使用示例：库存表变更监听（多表多 Observer）
package com.example.binlog.demo;import com.example.binlog.entity.BinlogEvent;import com.example.binlog.observer.AbstractTableObserver;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import ja va.util.List;import ja va.util.Map;/** * 库存表Binlog消费者. * 监听 stock_db.t_stock 表，库存变更时同步到搜索引擎. */@Componentpublic class StockSyncToEsObserver extends AbstractTableObserver { private static final Logger log = LoggerFactory.getLogger(StockSyncToEsObserver.class); public StockSyncToEsObserver() { super(2048); // 库存变更频繁，队列大一些 } @Override public void consume(List events) { // 批量处理：收集所有变更的SKU，一次性同步到ES List changedSkuIds = new ja va.util.ArrayList<>(); for (BinlogEvent event : events) { try { Map data = event.getData().get(0); Object skuId = data.get("sku_id"); if (skuId != null && !changedSkuIds.contains(skuId.toString())) { changedSkuIds.add(skuId.toString()); } } catch (Exception e) { log.warn("库存binlog解析异常", e); } } if (!changedSkuIds.isEmpty()) { batchSyncToElasticsearch(changedSkuIds); } } private void batchSyncToElasticsearch(List skuIds) { log.info("批量同步库存到ES, skuIds: {}, count: {}", skuIds, skuIds.size()); // 实际：查询DB最新库存 → 批量写入ES } @Override public String getSubscribeTable() { return "stock_db.t_stock"; }} 6.8 配置文件
# Binlog CDC 配置binlog: kafka: bootstrap-servers: localhost:9092 topic: binlog-business-db group-id: ${spring.application.name} max-poll-records: 500 tables: order_db.t_order,stock_db.t_stock七、生产环境注意事项
7.1 消费延迟监控
关键指标：├── consumer_lag（消费积压量）= 最新offset - 已消费offset├── consume_cost（单批消费耗时）├── queue_size（内部队列当前大小）└── error_count（消费异常计数）告警规则：├── consumer_lag > 10000 持续5分钟 → 告警├── consume_cost > 5000ms → 告警└── error_count 持续增长 → 告警
7.2 常见故障场景
| 故障 | 现象 | 应对 |
|---|---|---|
| CDC 中间件宕机 | Kafka 无新消息写入 | 中间件高可用部署；重启后自动续读 |
| 消费者处理慢 | consumer_lag 持续增长 | 优化业务逻辑；增加分区和消费者 |
| 消费者 OOM | 服务重启 | 减小 max-poll-records；增大 JVM 内存 |
| 消息格式异常 | 解析报错 | try-catch 单条跳过；记录到死信表 |
| 下游服务不可用 | 业务逻辑执行失败 | 重试+降级；消息不提交offset等待恢复 |
谈到生产环境，有几个常见的坑是必须要注意的。
7.3 幂等设计
为什么需要幂等？原因主要有两个：Kafka的At Least Once语义意味着消息可能会被重复消费；应用重启时，那些尚未提交offset的消息会被重新消费。因此，消费端必须做好幂等处理。
常见的解决方案有三种：
方案1：唯一键约束
INSERT IGNORE INTO event_log (event_id) VALUES (?) → 成功则处理，失败则跳过
方案2：乐观锁
UPDATE order SET status = 'PAID' WHERE id = ? AND version = ? → 影响行数=0则跳过
方案3：Redis去重
SET event:{eventId} EX 3600 NX → 设置成功则处理，失败则跳过
八、总结
完整链路一览
| 阶段 | 组件 | 职责 | 关键技术 |
|---|---|---|---|
| 1 | MySQL | 生成Binlog | ROW格式，两阶段提交 |
| 2 | CDC中间件 | 读取推送Binlog | 伪装Sla ve，Dump协议 |
| 3 | Kafka | 消息持久化传输 | Topic/Partition/Offset |
| 4 | 框架启动层 | Bean收集+路由表构建 | Spring ApplicationListener |
| 5 | 框架路由层 | 消费Kafka+按表名分发 | 观察者模式，批量处理 |
| 6 | 业务Observer | 处理特定表的变更 | 模板方法，线程隔离 |
涉及的设计模式
| 模式 | 在CDC链路中的体现 |
|---|---|
| 观察者模式 | Dispatcher（被观察者）通知多个 TableObserver（观察者） |
| 模板方法模式 | TableObserver 定义处理骨架，子类实现 consume() |
| 工厂模式 | BinlogObserverFactory 根据配置创建不同类型的 Observer |
| 策略模式 | 不同 Observer 代表不同表的不同处理策略 |
| 单例模式 | BinlogObserverFactory 全局唯一实例 |
| 适配器模式 | BinlogHandlerAdapter 适配 Spring Kafka 与框架内部接口 |
核心设计原则
| 原则 | 体现 |
|---|---|
| 单一职责 | 每个Observer只处理一张表的逻辑 |
| 开闭原则 | 新增表监听只需新建Observer，不修改框架代码 |
| 依赖倒置 | 框架依赖抽象TableObserver，不依赖具体实现 |
| 线程隔离 | 每个Observer独立线程池，故障不传播 |
| 有损降级 | 队列满时丢消息而非阻塞整个消费链路 |
