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

MySQL Master 视角：
"我有3个 Slave 在同步数据"
  Slave 1: 真正的从库（做读写分离）
  Slave 2: 真正的从库（做灾备）
  Slave 3: CDC中间件（伪装的，实际只是读binlog不回放SQL）

// 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();
}

// BinlogObserverFactory 根据配置创建不同类型的 Observer
class BinlogObserverFactory {
    void createBinlogConsumer(properties);         // 普通binlog消费
    void createRdbBinlogObserver(properties);       // 同步到RDB
    void createRedisBinlogObserver(properties);     // 同步到Redis
    void createEventBridgeBinlogObserver(properties); // EventBridge
}

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 内部线程池启动

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

┌──────────────────────────────────────────────────────┐
│ DefaultBinlogObserver                                 │
│                                                       │
│ Kafka Consumer 线程 (Spring Kafka管理)                │
│   → poll() 拉取消息                                   │
│   → 解析 + 路由                                       │
│   → 调用 TableObserver.post()                         │
│                                                       │
├──────────────────────────────────────────────────────┤
│ TableObserver-A 内部线程池                            │
│   → 异步执行 consumer() 方法                          │
│                                                       │
├──────────────────────────────────────────────────────┤
│ TableObserver-B 内部线程池                            │
│   → 异步执行 consumer() 方法                          │
└──────────────────────────────────────────────────────┘

// BinlogWorkQueue 继承 LinkedBlockingQueue
// 重写了 offer 方法
public boolean offer(T item) {
    if (remainingCapacity() == 0) {
        log.warn("Binlog队列已满，消息将被丢弃...");
        // 队列满时打印告警日志
    }
    return super.offer(item);
}

// TableObserver 实现了 DisposableBean
interface DisposableBean {
    void destroy() throws Exception;
}
// destroy() 被调用时：
// 1. isRunning = false （停止接收新任务）
// 2. executor.shutdown() （等待已提交任务执行完）
// 确保优雅停机，不丢正在处理的消息

jsh:
  binlog:
    - tables: xxx.xxx1,xxx.xxx2
// 匹配规则：
// 1. Kafka 消息中的 database.table 必须在 yml 的 tables 列表中
// 2. 且必须有对应的 TableObserver Bean 的 getSubTableName() 返回值与之匹配
// 两个条件同时满足，消息才会被路由到对应的 Observer
// 如果 tables 中有但没有对应 Observer → 消息被忽略（无报错）
// 如果有 Observer 但 tables 中没配 → 该 Observer 永远不会收到消息

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()) + "}";
    }
}

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;
    }
}

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);
    }
}

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;
    }
}

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.
 */
@Configuration
public 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();
        }
    }
}

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 表的变更，执行相应业务逻辑.
 *
 * 示例场景：
 * - 订单状态变更时通知物流系统
 * - 订单金额变更时通知财务系统
 * - 订单取消时通知库存系统释放库存
 */
@Component
public 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";
    }
}

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 表，库存变更时同步到搜索引擎.
 */
@Component
public 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";
    }
}

# 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

关键指标：
├── consumer_lag（消费积压量）= 最新offset - 已消费offset
├── consume_cost（单批消费耗时）
├── queue_size（内部队列当前大小）
└── error_count（消费异常计数）
告警规则：
├── consumer_lag > 10000 持续5分钟 → 告警
├── consume_cost > 5000ms → 告警
└── error_count 持续增长 → 告警