MySQL数据库Binlog CDC全链路实现方法全面详解

一、全局视角：从数据变更到业务消费的完整链路

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌───────────────┐    ┌──────────────┐

│  业务代码  │    │  MySQL   │    │ CDC中间件 │    │    Kafka      │    │  应用消费端   │

│           │    │          │    │(Canal等)  │    │               │    │              │

│ INSERT/   │───→│ InnoDB   │───→│伪装Slave  │───→│ binlog topic  │───→│ 框架路由     │

│ UPDATE/   │    │ +Binlog  │    │读取解析   │    │ 持久化存储    │    │ +TableObserver│

│ DELETE    │    │          │    │           │    │               │    │              │

└──────────┘    └──────────┘    └──────────┘    └───────────────┘    └──────────────┘

                                                                              │

                                                                              ▼

                                                                     ┌──────────────┐

                                                                     │  业务逻辑     │

                                                                     │ (推送/同步等) │

                                                                     └──────────────┘

这条链路分为五个阶段，每个阶段有不同的技术实现和设计考量。

二、阶段一：MySQL Binlog 生成

2.1 Binlog 是什么

Binlog（Binary Log）是 MySQL Server 层的日志，以二进制格式记录所有对数据产生变更的操作。它是 MySQL 事务提交的一部分——事务提交时，变更会同时写入 InnoDB 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，因为需要拿到具体哪一行的哪些字段从什么值变成了什么值。

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 核心原理——伪装 Slave

MySQL 的主从复制协议是公开的。CDC 中间件利用这个协议，向 MySQL Master 注册自己为一个 Slave，然后 MySQL 就会主动把 Binlog 推送过来。

MySQL Master 视角："我有3个 Slave 在同步数据"  Slave 1: 真正的从库（做读写分离）  Slave 2: 真正的从库（做灾备）  Slave 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<Binlog> binlogs) {        // 1. 提交到线程池（框架控制）        // 2. 调用 postToConsumer() （框架控制）        // 3. 最终调用 consumer() （子类实现）    }    // 抽象方法：子类必须实现    public abstract void consumer(List<Binlog> 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<TableObserver>>     │      "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<Binlog>)                │                ├── 提交到内部线程池 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 消费框架示例代码

以下是一个简化版的 Binlog CDC 消费框架实现，展示了完整的设计思路：

6.1 实体定义

package com.example.binlog.entity;import java.io.Serializable;import java.util.List;import java.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<String> pkNames;    /**     * 变更后的行数据（完整行）.     * INSERT: 新插入的数据     * UPDATE: 更新后的完整数据     * DELETE: 被删除的数据     */    private List<Map<String, Object>> data;    /**     * 变更前的数据（仅UPDATE时有值，且只含被修改的字段）.     */    private List<Map<String, Object>> 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<String> getPkNames() { return pkNames; }    public void setPkNames(List<String> pkNames) { this.pkNames = pkNames; }    public List<Map<String, Object>> getData() { return data; }    public void setData(List<Map<String, Object>> data) { this.data = data; }    public List<Map<String, Object>> getOld() { return old; }    public void setOld(List<Map<String, Object>> 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 java.util.List;import java.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<Runnable> 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<BinlogEvent> 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<BinlogEvent> 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 java.util.*;import java.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<String, List<AbstractTableObserver>> 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<String, String> records) {        // 按表名分组，批量投递（减少线程池提交次数）                // 按表名分组，批量投递（减少线程池提交次数）        Map<String, List<BinlogEvent>> groupedEvents = new HashMap<>();        for (ConsumerRecord<String, String> 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<String, List<BinlogEvent>> entry : groupedEvents.entrySet()) {            String tableName = entry.getKey();            List<BinlogEvent> events = entry.getValue();            List<AbstractTableObserver> observers = routeTable.get(tableName);            if (observers != null) {                for (AbstractTableObserver observer : observers) {                    observer.post(events);                }            }        }    }    /**     * 关闭所有Observer.     */    public void shutdown() {        for (List<AbstractTableObserver> observers : routeTable.values()) {            for (AbstractTableObserver observer : observers) {                observer.shutdown();            }        }        log.info("所有BinlogObserver已关闭");    }    /**     * 获取路由表信息（用于监控和调试）.     */    public Map<String, List<String>> getRouteInfo() {        Map<String, List<String>> info = new HashMap<>();        for (Map.Entry<String, List<AbstractTableObserver>> entry : routeTable.entrySet()) {            List<String> 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 javax.annotation.PostConstruct;import javax.annotation.PreDestroy;import java.util.HashMap;import java.util.List;import java.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<String> subscribedTables;    private final ApplicationContext applicationContext;    private ConcurrentMessageListenerContainer<String, String> 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<String, AbstractTableObserver> 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<String, Object> 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<String, String> consumerFactory =            new DefaultKafkaConsumerFactory<>(consumerProps);        // 创建监听容器        ContainerProperties containerProps = new ContainerProperties(binlogTopic);        containerProps.setMessageListener(            (MessageListener<String, String>) record -> {                // 单条消息模式，实际框架中可能使用批量模式                // 这里简化为逐条处理            }        );        // 使用批量监听模式        containerProps.setBatchListener(true);        containerProps.setMessageListener(            (org.springframework.kafka.listener.BatchMessageListener<String, String>)                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 java.util.List;import java.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<BinlogEvent> events) {        for (BinlogEvent event : events) {            try {                processSingleEvent(event);            } catch (Exception e) {                // 单条异常不影响整批处理                log.error("订单binlog处理异常, event: {}", event, e);            }        }    }    private void processSingleEvent(BinlogEvent event) {        Map<String, Object> 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<String, Object> data) {        log.info("订单创建: orderId={}, userId={}", orderId, userId);        // 业务逻辑：发送欢迎通知、初始化物流轨迹等    }    /**     * 处理订单更新.     * 通过对比 old 和 data 判断具体哪些字段变了.     */    private void handleOrderUpdated(String orderId, String newStatus,                                    List<Map<String, Object>> old, Map<String, Object> 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<String, Object> data) {        log.info("订单删除: orderId={}", orderId);        // 业务逻辑：归档、清理关联数据等    }    // ----- 下游通知方法 -----    private void notifyWarehouseToShip(String orderId, Map<String, Object> data) {        log.info("通知仓库发货: orderId={}", orderId);    }    private void notifyStockRelease(String orderId, Map<String, Object> data) {        log.info("通知库存释放: orderId={}", orderId);    }    private void notifyFinanceSystem(String orderId, Object amount) {        log.info("通知财务系统: orderId={}, amount={}", orderId, amount);    }    // ----- 工具方法 -----    private String getStringValue(Map<String, Object> data, String key) {        Object value = data.get(key);        return value != null ? value.toString() : null;    }    private Integer getIntValue(Map<String, Object> 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 java.util.List;import java.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<BinlogEvent> events) {        // 批量处理：收集所有变更的SKU，一次性同步到ES        List<String> changedSkuIds = new java.util.ArrayList<>();        for (BinlogEvent event : events) {            try {                Map<String, Object> 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<String> 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	伪装Slave，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独立线程池，故障不传播
有损降级	队列满时丢消息而非阻塞整个消费链路

以上就是MySQL数据库Binlog CDC全链路实现方法详解的详细内容，更多关于MySQL Binlog CDC全链路实现的资料请关注本站其它相关文章！

您可能感兴趣的文章:

• MySQL误删数据恢复之Binlog回放+全量备份+延迟从库的三种方案实战
• MySQL数据库binlog日志清理过程
• mysql如何查看binlog日志
• MySQL数据库恢复之Binlog格式详解
• MySQL如何修改binlog保存的天数
• MySQL中Binlog的完整格式解析(ROW 模式，默认)
• Mysql 通过binlog日志恢复数据的实现示例