2026/04/26/10 分钟/8 阅读/0 评论

CDC 消费者基类与数据同步模板：Binlog → Kafka → ES/ClickHouse

ERP 系统核心业务数据（商品、库存、订单）存储在 MySQL 中，查询需求包括： MySQL 擅长事务处理，但**模糊查询 + 多条件组合 + 历史版本**是 MySQL 的弱项——这些场景天然适合 Elasticsearch。新上线的

面试 #基础架构

1. 背景与痛点

业务背景

ERP 系统核心业务数据（商品、库存、订单）存储在 MySQL 中，查询需求包括：

商品域：模糊搜索（平台 SKU / ASIN / 标题）、组合条件过滤（品牌 + 品类 + 店铺 + 销售状态）
库存域：实时库存快照历史追踪（谁在什么时间把哪个仓库的哪个 SKU 改成了什么）
订单域：订单全链路状态变更日志

MySQL 擅长事务处理，但模糊查询 + 多条件组合 + 历史版本是 MySQL 的弱项——这些场景天然适合 Elasticsearch。

技术痛点

痛点一：同步时机与一致性问题

方案A：定时全量同步  → 延迟大，数据陈旧
方案B：业务代码双写  → 代码耦合，改一处动全身，漏写=数据不一致
方案C：CDC 监听 Binlog → 实时，无侵入，但实现复杂

痛点二：历史数据回填

新上线的 CDC 链路，历史数据怎么迁移？全量导入期间不能停服。

痛点三：数据过滤与字段筛选

Binlog 记录的是全字段变更，但 ES 只存储业务关心的字段。哪些字段变更需要同步？哪些不需要？

痛点四：幂等消费

Kafka 消费可能重复（Rebalance、重试），同一条 Binlog 被处理两次要能兜住。

2. 技术方案

整体架构

┌──────────────────────────────────────────────────────────────────────────────┐
│                           MySQL 主库集群                                     │
│                                                                              │
│   ┌────────────┐  ┌────────────┐  ┌────────────┐                           │
│   │mall_goods │  │ inv_stock  │  │pms_combo  │  ← 监听的表               │
│   └─────┬──────┘  └─────┬──────┘  └─────┬──────┘                           │
│         │               │               │                                  │
└─────────┼───────────────┼───────────────┼──────────────────────────────────┘
          │ Binlog        │ Binlog        │ Binlog
          ▼               ▼               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                      creeks-maxwell（定制 Maxwell）                           │
│                                                                              │
│   ┌──────────────────────────────────────────────────────────────────┐     │
│   │  BinlogConnectorReplicator  →  解析 Binlog 事件                   │     │
│   │  Filter                    →  按表/操作类型过滤                    │     │
│   │  MaxwellJson               →  JSON 序列化（含 old + new 数据）      │     │
│   │  MaxwellKafkaProducer       →  发布到 Kafka                        │     │
│   └──────────────────────────────────────────────────────────────────┘     │
│                                                                              │
│   Bootstrap 模式（历史数据回填）：                                           │
│   ┌──────────────────────────────────────────────────────────────────┐     │
│   │  BootstrapController + SynchronousBootstrapper                    │     │
│   │  → 按主键范围分批读取历史数据 → 同样发到 Kafka → Consumer 视为普通INSERT │  │
│   └──────────────────────────────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │ Kafka Topic
                                    ▼
┌──────────────────────────────────────────────────────────────────────────────┐
│                      creeks-cdc（CDC 消费者服务）                             │
│                                                                              │
│   ┌──────────────────────────────────────────────────────────────────┐     │
│   │  KafkaBatchListenerContainerFactory                                │     │
│   │  → 批量消费（max.poll.records=5000）                               │     │
│   │  → MANUAL_IMMEDIATE 手动提交                                       │     │
│   │  → SeekToCurrentErrorHandler → 重试3次 → DLT                       │     │
│   └──────────────────────────────────────────────────────────────────┘     │
│                                                                              │
│   消费者链路：                                                             │
│   ┌──────────┐   ┌───────────────┐   ┌────────────────┐                  │
│   │DataConvert│──▶│ BaseLogConsumer │──▶│ ES Bulk写入   │ (基础数据变更日志) │
│   │ (反序列化) │   │  过滤+字段筛选 │   │ base:log:index │                  │
│   └──────────┘   └───────────────┘   └────────────────┘                  │
│                                                                              │
│   ┌──────────────────────┐   ┌───────────────┐   ┌──────────────┐       │
│   │ GoodsSyncConsumer    │──▶│ Feign 调用    │──▶│ ES Bulk写入  │ (商品同步)  │
│   │ (按平台分 topic)      │   │ GoodsSyncService│  │ goods:index │          │
│   └──────────────────────┘   └───────────────┘   └──────────────┘       │
│                                                                              │
│   ┌────────────────────────┐   ┌────────────────┐   ┌────────────────┐   │
│   │GoodsSnapshotConsumer   │──▶│ ES Bulk写入    │──▶│ClickHouse写入 │ (商品快照)  │
│   │ (增量变更+字段对比)    │   │ mall-goods-snapshot │  │ 历史版本追溯  │       │
│   └────────────────────────┘   └────────────────┘   └────────────────┘   │
│                                                                              │
│   ┌──────────────────────┐   ┌──────────────────────┐                     │
│   │ SkuSyncConsumer      │──▶│ Kafka 转发 (SKU状态) │ (采购联动)        │
│   └──────────────────────┘   └──────────────────────┘                     │
│                                                                              │
│   ┌────────────────────────┐   ┌────────────────┐                       │
│   │StockSnapshotConsumer   │──▶│ ClickHouse写入  │ (库存快照)           │
│   └────────────────────────┘   └────────────────┘                       │
└─────────────────────────────────────────────────────────────────────────────┘

Maxwell Kafka Topic 路由

表 mall_goods 变更 → kafka.topic.cdc_api_mall_goods
表 inv_stock 变更   → kafka.topic.cdc_api_stock
表 pms_combo      → kafka.topic.cdc_api_combo
...

CdcData 统一数据模型

{
  "database": "creeks_mall",
  "table": "mall_goods",
  "type": "update",
  "data": { "id": 123, "title": "新标题", "status": 1 },
  "old": { "title": "旧标题", "status": 0 },
  "xid": 9876543,
  "commit": true,
  "ts": 1713763200
}

3. 实现细节

模块一：creeks-maxwell（Binlog 监听 + Kafka 发布）

核心定制点：

BinlogConnectorReplicator：连接 MySQL Binlog，解析 WRITE / UPDATE / DELETE / ERD 事件
Filter：按 include_dbs / exclude_tables 配置只监听需要的表
MaxwellKafkaProducer：将变更数据发布到 Kafka，支持自定义 ProducerFactory

// creeks-maxwell 自定义 ProducerFactory 示例
public class CustomProducerFactory implements ProducerFactory {
    @Override
    public AbstractProducer createProducer(MaxwellConfig config) {
        return new MaxwellKafkaProducer(config); // 发布到 ERP 的 Kafka 集群
    }
}

模块二：Bootstrap 历史数据回填

问题：CDC 链路上线前，MySQL 中已有上亿条历史数据，这些数据怎么同步到 ES？

方案：Maxwell Bootstrap 机制——按主键范围分批扫描历史表，每批数据包装成 bootstrap-insert 类型的消息发到 Kafka。Consumer 将其视为普通 INSERT 写入 ES。

// Maxwell 配置（maxwell.properties）
bootstrap.nodes=10                    // 分 10 个节点并行回填
bootstrap.fetch_size=50000            // 每批 5 万条
bootstrap.method=synchronous          // 同步扫描模式（vs async 异步）

Consumer 中过滤 Bootstrap 消息：

// GoodsSyncConsumer.processGoodsSyncToEs()
Set<String> filterTypeSet = CollUtil.newHashSet(
    CdcType.BOOTSTRAP_START.getType(),  // 忽略开始标记
    CdcType.BOOTSTRAP_COMPLETE.getType() // 忽略完成标记
);
List<CdcData> filtered = cdcDataList.stream()
    .filter(i -> !filterTypeSet.contains(i.getType()))
    .collect(Collectors.toList());
// filtered 中包含 BOOTSTRAP_INSERT，视为普通 INSERT 处理

模块三：BaseLogConsumer（字段级过滤引擎）

这是 CDC 消费者中最复杂的部分——根据不同表、不同操作类型、不同字段组合，决定"是否记录"和"记录哪些字段"。

// 核心判断逻辑（BaseLogConsumer.recordBaseLog）

// 1. DELETE 操作：只关注 pms_combo_sku_relation 表
if (isDeleteAndPmsComboSkuRelation(cdcData)) { ... }

// 2. INSERT 操作：只关注 pms_combo_sku_relation 和 mall_goods_inventory
if (isInsertAndSpecialTable(cdcData)) { ... }

// 3. UPDATE 操作：所有表都需要（记录变更前后的差异）
if (isUpdate(cdcData)) { ... }

// 4. 字段级过滤：调用 cleanCdcData() 比对 old vs new，只保留真正变化的字段
if (cleanCdcData(cdcDataCopy)) continue; // 变化字段不在关注列表中，跳过

cleanCdcData 核心逻辑：

public boolean cleanCdcData(CdcData cdcData) {
    Map<String, Object> currentRes = new HashMap<>();
    Map<String, Object> oldRes = new HashMap<>();

    // 必须记录的字段
    if (isNeedLog(key)) { ... }

    // 商品表：只记录特定字段（platform_sku, asin, status 等 20+ 个）
    if (isMallGoodsTable(table) && isGoodsNeedRecordColumn(key)) { ... }

    // 库存相关表：inv_stock / inv_winit_stock / inv_winit_declare_stock 各有不同字段集
    if (invStockRelatedTable && isInvStockOrInvWinitStockNeedRecordColumn(key, table)) { ... }
}

模块四：IndexHelper（ES 索引生命周期管理）

与 DeadLetterService 中的逻辑类似，CDC 服务也需要创建和管理 ES 索引：

public class IndexHelper {
    // 本地缓存已创建的索引，避免每次都查 ES
    private static final Set<String> createdIndices = new HashSet<>();

    public void ensureIndex(String indexName, String lockKey, Map<String, Object> indexMapping) {
        if (createdIndices.contains(indexName)) {
            return;
        }
        // 分布式锁：多实例同时启动时，只有一个实例创建索引
        lockService.lockAndExecute(lockKey, 30, -1, TimeUnit.SECONDS, () -> {
            if (createdIndices.contains(indexName)) return;
            createOrUpdateIndex(indexName, indexMapping);
            createdIndices.add(indexName);
        });
    }

    // 按月分索引：base:log:index-202404
    public String indexNameOfDate(String originIndexName) {
        return originIndexName + "-" + monthFormatter.format(TimeUtils.today());
    }
}

模块五：多 Consumer 分工

Consumer	监听 Topic	目标存储	消费组	批量大小
`BaseLogConsumer`	`base-log.topics`（多表合并）	ES	`cdc-base-log`	100
`GoodsSyncConsumer`	`cdc_api_mall_goods`	ES（分平台）	`goods-sync-to-es`	5000
`GoodsSnapshotConsumer`	`goods-snapshot.topic`	ES + ClickHouse	`goods-snapshot-log`	2000
`SkuSyncConsumer`	`cdc_api_pms_sku`	Kafka（转发）	`sku-snapshot-log`	100
`StockSnapshotConsumer`	`stock-snapshot.topic`	ClickHouse	`stock-snapshot-log`	100

平台分流设计：GoodsSyncConsumer 监听商品变更后，按 platform_name 分流：

eBay 平台 → ebay_item Topic → ebay-goods-sync-to-es 消费组
其他平台 → other_item Topic → other-goods-sync-to-es 消费组

这样拆分的原因是：eBay 商品同步逻辑与其他平台有差异（字段映射不同），分流后各自独立消费，互不干扰。

模块六：商品快照变更对比（GoodsSnapshotConsumer）

商品变更不仅记录到 ES 搜索索引，还记录到 ClickHouse 用于历史版本追溯：

// compareGoodsChangeField：对比变更前后，标记哪些类型的字段发生了变化
private Integer[] compareGoodsChangeField(CdcData cdcData) {
    List<Integer> changeFieldTypeList = new ArrayList<>();
    for (GoodsFieldChangeTypeEnum changeTypeEnum : GoodsFieldChangeTypeEnum.values()) {
        boolean haveFieldChanged = cdcData.getData().containsKey(enum.getField())
            && cdcData.getOld().containsKey(enum.getField());
        boolean haveValueChanged = !ObjectUtil.equals(
            cdcData.getData().get(enum.getField()),
            cdcData.getOld().get(enum.getField()));
        if (haveFieldChanged && haveValueChanged) {
            changeFieldTypeList.add(changeTypeEnum.getType());
        }
    }
    return ArrayUtil.toArray(changeFieldTypeList, Integer.class);
}

// ES 存储：new_data + old_data + 变更字段类型数组
Map<String, Object> source = buildGoodsSnapshotSource(cdcData);
// → { "new_data": "{...}", "old_data": "{...}", "type": [1, 3, 7] }

// ClickHouse 存储：完整快照（每条变更一条记录，含时间戳）
GoodsSnapshot gs = getGoodsSnapshotFromData(cdcData);
gs.setId(generateId()); // 雪花算法生成 ID
goodsSnapshotService.saveBatch(goodsSnapshotList);

模块七：幂等保障

CDC 消费天然支持幂等——因为 ES / ClickHouse 的主键就是数据库主键：

// ES Bulk 请求中，IndexRequest 使用数据库主键作为 _id
IndexRequest indexRequest = Requests.indexRequest(indexName)
    .id(cdcData.getData().get("id").toString()) // 主键幂等
    .source(buildSource(cdcData));
bulkRequest.add(indexRequest);

// 同一 ID 的数据多次写入 = 更新操作，不会重复
// ClickHouse 同理，以主键 upsert

4. 技术栈与亮点归纳

技术栈清单

Maxwell（Binlog 监听，定制 ProducerFactory）
Kafka（消息总线，批量消费）
Elasticsearch RestHighLevelClient（Bulk API 写入）
ClickHouse（历史快照存储）
Redis + Redisson（分布式锁，索引创建协调）
MyBatis Plus（ClickHouse 写入）
设计模式：模板方法（Consumer 基类）、策略（按表/平台分发）

技术亮点

Binlog 零侵入同步：不修改一行业务代码，监听 Binlog 自动同步到 ES/ClickHouse
Bootstrap 历史回填：新链路上线时，先跑 Bootstrap 把历史数据迁移，不影响在线服务
字段级变化追踪：不仅记录变更结果，还记录变更了哪些字段（变更类型枚举）
平台分流消费：按 eBay / 其他平台分流，各自独立消费，互不干扰
索引预创建 + 分布式锁：多实例并发时只有一个实例创建索引，避免冲突
幂等消费：ES 主键 = DB 主键，同一消息处理两次不重复

5. 难点与亮点剖析

业务难点

历史数据迁移：已有上亿条数据，不能停服，Maxwell Bootstrap 分批扫描是最优解
字段过滤规则复杂：不同表需要记录的字段不同，同一表的 INSERT / UPDATE / DELETE 处理逻辑也不同
变更前后对比：不仅要记录最新值，还要记录"哪些字段从 A 变成了 B"（用于历史版本追溯）

技术难点

难点一：Binlog 位置管理

Maxwell 需要持久化记录"消费到哪个 Binlog 位置了"，否则重启后会重复消费。Maxwell 将 Position（binlog 文件 + offset）存储在 Maxwell 自己的 MySQL 表中（maxwell.positions）。

难点二：多表共用 Topic 的路由分发

BaseLogConsumer 监听的 Topic 合并了多个表的变更，需要在 Consumer 内部按 table 字段路由到不同的处理逻辑。

难点三：Maxwell 消费位点与业务 DB 主从延迟

Maxwell 读取 Binlog 有一定延迟（通常几百毫秒），如果业务写 MySQL 后立刻查 ES，可能查不到最新数据。设计方案：业务写 -> 等 500ms -> 查 ES（可接受范围内）。

实现亮点

按平台分流：eBay 商品同步独立链路，其他平台共用链路，互不干扰
old + new 双数据保留：完整保留变更前后数据，支持任意时间点回溯
变更类型枚举化：GoodsFieldChangeTypeEnum 将字段映射为枚举值，ES 查询时按枚举过滤即可
按月分索引：base:log:index-202404，按月清理过期数据

6. 替代方案对比

方案一：业务代码双写（已排除）

实现：MyBatis 拦截器 / AOP 在 INSERT/UPDATE 时同时写 ES
优劣：实现简单，但代码耦合，改表结构要同时改两边，漏写导致数据不一致
结论：ERP 系统历史包袱重，不可能重构所有业务代码

方案二：Canal（已排除）

实现：阿里开源的 MySQL Binlog 增量订阅
优劣：生态成熟，但 Kafka Producer 支持不如 Maxwell 完善；Maxwell 对 Bootstrap 模式支持更好
结论：Maxwell 的 Bootstrap 机制更适合历史数据回填场景

最终方案

采用 Maxwell + Kafka + 多 Consumer 分工 方案，Maxwell 负责 Binlog 解析和历史回填，Kafka 作为缓冲和分发总线，各 Consumer 负责各自领域的数据处理和存储。

7. 技术选型与方案抉择复盘

选型思路

2021 年需要将商品和库存数据同步到 ES 以支持模糊搜索和历史追溯。调研了双写、Canal、Maxwell 三种方案，最终选择 Maxwell，原因：

零侵入：不修改业务代码
Bootstrap 支持：解决了历史数据迁移问题
团队有积累：之前用过 Maxwell

优点

实时性强：Binlog 变更到 ES 通常 < 1 秒
数据一致性高：基于 Binlog，不存在漏写问题
可追溯：old + new 双数据，任意时间点状态可查

缺点 / 已知不足

延迟不确定：Binlog 延迟 + Kafka 消费延迟 + ES 写入延迟，总延迟不可精确控制
DDL 变更需要重启：新增字段需要同步修改 ES Mapping，通常手动处理
消费顺序不保证：Kafka 分区内的消息有序，但多分区场景下跨分区顺序不可控

可优化点

引入 Debezium（功能更全面，支持 Schema Changes 监控）
CDC 链路增加端到端延迟监控（Binlog 产生 → Consumer 消费 → ES 写入）
ES Mapping 变更支持自动化（解析 DDL 语句自动更新 Mapping）

8. 成果与数据

接入规模：商品（mall_goods）、库存（inv_stock）、组合品关系（pms_combo_sku_relation）等 10+ 张核心表 接入 CDC 链路
商品同步：eBay + 其他平台商品数据 100% 同步到 ES，模糊搜索覆盖率 100%
历史快照：商品变更记录 100% 入 ClickHouse，支持任意时间点回溯查询
Bootstrap 效率：历史数据回填速度约 5 万条/批，多批次并行，亿级数据可在数小时内完成
ES 查询延迟：商品模糊搜索 P99 延迟 < 200ms（相比 MySQL LIKE 搜索提升 10 倍以上）

9. 面试题清单

【基础】MySQL Binlog 的三种格式（Statement / Row / Mixed）有什么区别？Maxwell 默认用哪种？为什么？
- 考察点：Row 格式记录每行变化前后，Statement 记录 SQL 语句；Maxwell 推荐 Row 格式，避免函数/存储过程导致数据不一致
【基础】CDC（Change Data Capture）的原理是什么？和双写相比核心优势是什么？
- 考察点：监听数据库变更日志（Binlog/WAL），不侵入业务代码；双写代码耦合、漏写风险
【进阶】Maxwell 的 Bootstrap 机制是什么？它解决了什么问题？
- 考察点：历史数据回填、按主键范围分批扫描、bootstrap-insert 消息类型
【进阶】为什么 BaseLogConsumer 中的字段过滤逻辑那么复杂（20+ 个 isXxxTable / isXxxColumn 判断）？这种设计有什么问题？
- 考察点：配置化 vs 代码硬编码、可维护性问题、未来如何演进为规则引擎
【进阶】Kafka 批量消费中，max.poll.records 设置为 5000 和 100 有什么区别？大了会有什么风险？
- 考察点：批量大小与内存占用的关系、单次处理时间过长导致 Rebalance
【原理】ES Bulk 请求中，IndexRequest 的 id 字段决定了什么？如果两条消息 id 相同会发生什么？
- 考察点：ES 的 upsert 语义（id 相同 = 更新）；幂等消费的实现基础
【场景】如果 Maxwell 挂了重启，会重复消费已消费过的 Binlog 吗？Maxwell 是怎么记录消费位点的？
- 考察点：positions 表存储 (binlog_file, binlog_position, gtid)，重启后从记录的位置继续消费；可能重复消费，需业务方幂等
【场景】Binlog 变更到 ES 可查询的延迟大约是多少？有哪些环节会影响延迟？
- 考察点：Binlog 产生 → Maxwell 解析 → Kafka 消费 → ES Bulk 写入；MySQL 主从延迟也会影响
【设计】如果需要支持"ES 写入失败后自动重试并告警"，框架应该怎么改造？
- 考察点：SeekToCurrentErrorHandler + DLT；BulkResponse.hasFailures() 检查；死信队列
【延伸】和 Canal / Debezium 相比，Maxwell 的优势在哪里？什么时候应该切换到 Debezium？
- 考察点：Maxwell 轻量、Bootstrap 友好；Debezium 支持 Schema Change 监控、CDC 链路可监控、连接器生态更广

10. 简历提炼版

设计并实现 CDC 数据同步链路，基于 Maxwell 监听 MySQL Binlog 将变更数据发布到 Kafka，多 Consumer 分工处理商品/库存/订单同步到 ES（支持模糊搜索）+ ClickHouse（历史版本追溯）；Bootstrap 机制解决历史数据回填难题，字段级变化对比记录变更类型枚举；全链路零侵入业务代码，商品搜索 P99 延迟 < 200ms，亿级历史数据数小时内完成迁移。

关键源码路径

Maxwell 定制层

文件	路径
Maxwell 入口	`other/creeks-maxwell/.../maxwell/Maxwell.java`
Binlog 解析器	`other/creeks-maxwell/.../maxwell/replication/BinlogConnectorReplicator.java`
表过滤规则	`other/creeks-maxwell/.../maxwell/filtering/Filter.java`
Kafka 生产者	`other/creeks-maxwell/.../maxwell/producer/MaxwellKafkaProducer.java`
Bootstrap 控制器	`other/creeks-maxwell/.../maxwell/bootstrap/BootstrapController.java`

CDC 消费者层

文件	路径
统一数据模型	`creeks-cdc/client/.../cdc/client/model/CdcData.java`
数据转换工具	`creeks-cdc/server/.../cdc/utils/DataConvert.java`
消费者基类	`creeks-cdc/server/.../cdc/consumer/log/BaseLogConsumer.java`
ES 索引管理	`creeks-cdc/server/.../cdc/consumer/log/IndexHelper.java`
商品同步消费者	`creeks-cdc/server/.../cdc/consumer/goods/GoodsSyncConsumer.java`
商品快照消费者	`creeks-cdc/server/.../cdc/consumer/goods/GoodsSnapshotConsumer.java`
SKU 同步消费者	`creeks-cdc/server/.../cdc/consumer/products/SkuSyncConsumer.java`
库存快照消费者	`creeks-cdc/server/.../cdc/consumer/stock/StockSnapshotConsumer.java`
Kafka 消费者配置	`creeks-cdc/server/.../cdc/config/KafkaConfig.java`