CDC + ES 同步方案 面试题库

项目背景：商品表数据量过大，查询性能瓶颈。通过 Maxwell 监听 MySQL binlog → Kafka → 商品服务消费 → ES，实现无侵入式数据同步。

---

初级（理解原理）

1. 什么是 CDC？为什么不用轮询而用 binlog？

核心回答要点：

• CDC（Change Data Capture）是变更数据捕获，通过监听 binlog 获取数据变更，属于事件驱动
• 轮询的问题：
  • 延迟高：轮询间隔决定最小延迟
  • 性能差：需要加 update_time 索引，全表扫描成本高
  • 漏数据：两次轮询之间的多次变更可能丢失
  • 资源浪费：无变更时也要查
• Binlog 是 MySQL 原生机制，变更即触发，延迟毫秒级

2. 简述 Maxwell 的工作原理

核心回答要点：

MySQL (binlog) → Maxwell 伪装成 Slave → 解析 binlog event → 序列化为 JSON → 推送至 Kafka

• Maxwell 伪装成一个 MySQL 从库，执行 BINLOG DUMP 协议
• 通过 MySQL 的 Java 客户端（BinaryLogClient）连接主库，接收 binlog 事件流
• 解析 binlog 中的 INSERT/UPDATE/DELETE 事件，提取变更的表名、数据库名、新旧数据
• 将事件序列化为 JSON，发送至 Kafka Producer

3. Maxwell 和 Canal 的区别？为什么选 Maxwell？

核心回答要点：

维度	Maxwell	Canal
开发语言	Java	Java
输出格式	JSON（统一）	Protobuf / JSON
Schema 追踪	内置（自动捕获 DDL 变更）	需额外处理
Bootstrap 回放	内置支持	不支持
Topic 路由	支持模板（%{database}_%{table}）	默认单 canal-destination
过滤	内置 JavaScript 过滤器	需自行配置

• 决定性因素：Maxwell 内置 Bootstrap 能力，支持全表/部分数据回放（--where 条件），是存量数据初始化的刚需
• Canal 要实现类似功能需要自己写全量导出脚本 + binlog 增量同步拼接，复杂度高

4. Binlog 有哪几种格式？Maxwell 要求哪种？

核心回答要点：

• STATEMENT：记录 SQL 语句（原始语句）
• ROW：记录每行数据的变更（每行的前像/后像）
• MIXED：以上两者混合
• Maxwell 要求 ROW 格式，因为需要解析每行的具体变更数据（旧值、新值）
• STATEMENT 格式只记录 SQL，无法提取结构化字段变更

-- 查看当前 binlog 格式
SHOW VARIABLES LIKE 'binlog_format';
-- 修改
SET GLOBAL binlog_format = 'ROW';

---

中级（架构设计）

5. Maxwell 如何保证消费位置不丢失？故障恢复怎么做？

核心回答要点：

• Maxwell 将读取的 binlog 位置（文件名 + offset）持久化到自身数据库的 positions 表中，每秒检查一次，有更新则写入
• 重启时从 positions 表查询最后位置，从该位置继续读取
• 故障恢复时，K8S 负责实例回滚/重建，Maxwell 重新启动后自动从上次 position 恢复
• GTID 模式：开启 GTID 后，位置记录变为 GTID set，更可靠，不依赖 binlog 文件名，主从切换也不丢位置

追问：如果 positions 表写入失败会怎样？

• 会导致重启后从头或从最新 binlog 位置开始，可能重复消费或丢失
• 解决方案：开启 GTID，positions 表的写入重试，监控告警

Mermaid 时序图：Maxwell 位置持久化流程

sequenceDiagram
    participant MB as MySQL Master
    participant M as Maxwell
    participant DB as Maxwell DB (positions表)
    participant K as Kafka

    MB->>M: 发送 binlog events
    M->>K: 解析后推送 JSON 到 topic
    Note over M: 每秒检查一次
    M->>DB: 写入最新 binlog position
    DB-->>M: ACK
    Note over M: 进程重启
    M->>DB: 查询上次 position
    DB-->>M: 返回 mysql-bin.00029:23264509
    M->>MB: 从该位置继续 BINLOG DUMP

6. Maxwell 首次启动时，如何决定从哪个 binlog 位置开始？

核心回答要点：

1. 查询 maxwell.positions 表，看是否有上次的位置记录
2. 如果有，从该位置恢复
3. 如果没有（首次启动），发送 SHOW MASTER STATUS 获取当前 binlog 位置，从该位置开始监听
4. 同时调用 SchemaStore 初始化当前数据库的表结构快照

注意：首次启动时，不会回溯历史 binlog，只会从启动时刻的位置开始。如果需要同步存量数据，需要使用 Bootstrap 功能。

7. Bootstrap（数据回放）的过程是同步还是异步？会不会阻塞实时 binlog 消费？

核心回答要点：

• 默认是异步的（--bootstrapper=async）
• 异步模式下：
  • 非 Bootstrap 的表由主线程正常复制，不受影响
  • Bootstrap 表的 binlog 事件会排队等待，保存在 RowMapBuffer（LinkedList，内存 1000 条，超出后溢出到磁盘）
  • Bootstrap 完成后，调用 pushSkippedRows() 将排队的事件重新发送
• 结论：不会阻塞实时 binlog 消费，主链路正常处理

追问：如果同步模式会怎样？

• 同步模式下（--bootstrapper=sync），Bootstrap 过程会阻塞 binlog 的读取，所有表的变更都会被暂停，生产环境不建议使用

8. 如何触发 Bootstrap？全量和部分回放怎么做？

核心回答要点：

方式一：命令行工具

# 全表回放
bin/maxwell-bootstrap \
  --user=maxwell --password=xxx --host=localhost \
  --database shop_dev --table goods

# 部分回放（指定 where 条件）
bin/maxwell-bootstrap \
  --user=maxwell --password=xxx --host=localhost \
  --database shop_dev --table goods \
  --where="update_time > '2026-01-01'"

方式二：往 maxwell.bootstrap 表插入记录

INSERT INTO maxwell.bootstrap (database_name, table_name)
VALUES ('shop_dev', 'goods');

-- 部分回放
INSERT INTO maxwell.bootstrap (database_name, table_name, where_clause)
VALUES ('shop_dev', 'goods', 'id >= 10000');

回放消息的 type 序列：

• bootstrap-start → bootstrap-insert（多条）→ bootstrap-complete

9. Maxwell 监听到表结构变更（DDL）后怎么处理？

核心回答要点：

• Maxwell 内置了 SchemaStore，首次启动时捕获所有表结构存入自身的 schemas 表
• 通过解析 binlog 中的 DDL 事件（CREATE TABLE、ALTER TABLE、DROP TABLE），实时更新内存中的 Schema
• 每次 binlog 解析时，使用当前 Schema 将行数据序列化为正确的 JSON 字段
• 即使 Maxwell 宕机期间发生了 DDL，重启后也能从 binlog 中的 DDL 事件恢复 Schema 状态

代码示例（Maxwell 日志）：

INFO AbstractSchemaStore - storing schema @Position[BinlogPosition[mysql-bin.00029:23264509]]
  after applying "ALTER TABLE goods ADD COLUMN new_field VARCHAR(255)"
  new schema id is 4

---

高级（实战问题）

10. Kafka Topic 的划分策略？为什么要按 {database}_{table} 拆分？

核心回答要点：

策略	示例	优点	缺点
单 Topic	cdc	消费入口统一	单 Topic 压力大，无法按表独立扩 partition
按数据库	cdc_{database}	数据库级别隔离	粒度仍粗，无法独立消费单个表
按表	cdc_{database}_{table}	灵活控制，可按表扩消费组	Topic 数量多

本项目选择按表拆分的原因：

• 商品表和其他表的变更量级不同，需要独立控制 partition 数量
• 不同服务可能只关心部分表（如商品服务只消费 goods 相关表），按表拆分可精准路由
• 配合 @KafkaListener(topics = {...}) 或 @KafkaListener(topicPattern = "cdc_goods_*") 灵活匹配

11. 如何保证 MySQL → Kafka → ES 的数据一致性？

核心回答要点：

这是一个经典的分布式数据一致性问题，分层来看：

1）MySQL → Kafka（Maxwell 层）：

• Maxwell 是"至少一次"（at-least-once）语义：binlog 读到 → 发 Kafka → 确认写入 → 更新 position
• 理论上不会丢数据，极端情况（Maxwell 崩溃未更新 position）可能重复发送

2）Kafka → 消费服务：

• 使用手动 ACK（AckMode.MANUAL），ES 写入成功才提交 offset
• 消费失败时 ack.nack() 触发重发
• 配合 Kafka 的 enable.auto.commit=false + acks=all + retries=3

3）消费服务 → ES：

• ES 使用 document ID（商品 ID）做幂等写入，天然支持覆盖更新
• 删除操作：ES deleteById，也是幂等的
• 最终一致性，不是强一致性

结论：整体是最终一致性模型，依赖 ES 的幂等写入 + Kafka 的手动 ACK 保障数据不丢失。

Mermaid 数据流图：一致性保障机制

flowchart TD
    A[MySQL 变更] -->|binlog| B[Maxwell]
    B -->|push JSON| C[Kafka Topic]
    C -->|poll| D[消费服务]
    D -->|写入 ES| E{ES 写入成功?}
    E -->|是| F[ack.acknowledge]
    E -->|否| G[ack.nack 重试]
    G --> D
    F --> H[Offset 提交]
    style F fill:#90EE90
    style G fill:#FFB6C1

12. 如果消费端处理失败，消息堆积了怎么办？

核心回答要点：

这是经典的Kafka 消息积压问题，分步骤排查和处理：

排查：

1. 检查消费服务日志，定位失败原因（ES 不可用？数据格式异常？）
2. 查看 Kafka Consumer Lag（kafka-consumer-groups.sh --describe）
3. 确认是否是偶发还是持续性问题

处理：

1. 修复根因：比如 ES 恢复、代码 bug 修复
2. 临时扩容：增加消费服务的 partition 数量 + 消费者实例数（消费者数 ≤ partition 数）
3. 跳过脏数据：如果是个别脏数据导致，可以考虑将失败的消息转存到死信队列（DLQ），继续消费后续消息
4. 加速消费：调整 max.poll.records 增大单次拉取量，提高消费吞吐

预防措施：

• 消费端加超时和熔断（如 Sentinel），ES 不可用时快速失败
• 失败消息写入死信 Topic，不阻塞主消费流
• 监控 Kafka Consumer Lag，设置告警阈值

13. Maxwell 的压力瓶颈在哪里？怎么调优？

核心回答要点：

潜在瓶颈：

1. MySQL 主库：Maxwell 作为从库拉取 binlog，如果主库写入量巨大，binlog 生成速度可能超过 Maxwell 解析速度
2. Maxwell 内存：Bootstrap 异步模式下，排队的 binlog 事件保存在内存（1000 条）+ 磁盘溢出，大量并发 Bootstrap 可能导致内存压力
3. Kafka Producer：如果 topic 数量多、消息量大，Producer 端的 buffer.memory、linger.ms、batch.size 配置不当会成为瓶颈

调优方向：

• MySQL 侧：开启 GTID，确保 binlog 格式为 ROW，合理设置 binlog_expire_logs_seconds
• Maxwell 侧：通过 --filter 排除无关表，减少处理量；调整 producer 的 batch 大小
• Kafka 侧：增加 partition 数量，调整 acks=all + retries + retry.backoff.ms

# config.properties 示例调优
kafka.retries=3
kafka.acks=all
kafka.retry.backoff.ms=1000
kafka.batch.size=65536
kafka.linger.ms=5

14. 如果商品表有 10 亿行，Bootstrap 全表回放需要多久？会不会影响线上？

核心回答要点：

• 全表 10 亿行一次性 Bootstrap 会非常耗时（可能数小时），且 Maxwell 需要遍历全表执行 SELECT，会对 MySQL 主库产生额外读压力
• 建议策略：
  1. 分批 Bootstrap：使用 --where 条件按 ID 范围分批，例如 id BETWEEN 1 AND 1000000、id BETWEEN 1000001 AND 2000000
  2. 低峰期执行：在业务低峰期（凌晨）执行 Bootstrap，减少对线上查询的影响
  3. 先初始化存量，再启动增量：先通过离线导出（SELECT INTO OUTFILE 或 mysqldump）批量导入 ES，再启动 Maxwell 增量同步
  4. 监控 MySQL 负载：Bootstrap 期间监控 MySQL 的 CPU、IO、连接数，避免影响线上业务

---

架构师级（方案设计）

15. 如果要将这个 CDC 方案推广到整个公司（多套 MySQL 实例、几十个微服务），架构怎么设计？

核心回答要点：

挑战：

• 多个 MySQL 实例 → 每个实例需要一个 Maxwell 实例（Maxwell 监听粒度是实例级别）
• 几十个微服务 → Topic 数量爆炸（每个服务的每张表一个 Topic）
• 数据安全问题 → 不同服务的 CDC 数据需要隔离

架构设计：

flowchart LR
    subgraph "MySQL Instances"
        M1[(MySQL-1\n商品库)]
        M2[(MySQL-2\n订单库)]
        M3[(MySQL-3\n用户库)]
    end

    subgraph "Maxwell Layer"
        MX1[Maxwell-1]
        MX2[Maxwell-2]
        MX3[Maxwell-3]
    end

    subgraph "Kafka Cluster"
        KT1[Topic: cdc_goods_*]
        KT2[Topic: cdc_order_*]
        KT3[Topic: cdc_user_*]
    end

    subgraph "CDC Service Layer"
        CS1[CDC-Goods-Svc]
        CS2[CDC-Order-Svc]
        CS3[CDC-User-Svc]
    end

    subgraph "Downstream"
        ES[(Elasticsearch)]
        R[(Redis)]
        CK[(ClickHouse)]
    end

    M1 --> MX1
    M2 --> MX2
    M3 --> MX3
    MX1 --> KT1
    MX2 --> KT2
    MX3 --> KT3
    KT1 --> CS1
    KT2 --> CS2
    KT3 --> CS3
    CS1 --> ES
    CS1 --> R
    CS2 --> ES
    CS2 --> CK
    CS3 --> R

关键设计决策：

1. Maxwell 与 MySQL 实例 1:1，避免一个 Maxwell 实例跨实例监听（Maxwell 不支持）
2. CDC 中间件层（CDC Service）：统一消费 Kafka 消息，转换为标准数据结构后分发给下游，隔离 CDC 框架变化
3. 按服务/模块拆分 Topic 命名空间：cdc_{业务域}_{表名}，如 cdc_goods_spu、cdc_goods_sku
4. Kafka 多集群/多 Topic 隔离：不同业务域使用不同 Topic，消费组独立
5. 监控与运维：搭建 Maxwell/Kafka 消费 Lag 监控，CDC 服务异常时告警

16. 如果中途需要将 Maxwell 替换为其他 CDC 框架（如 Flink CDC），如何做到对下游无感知？

核心回答要点：

核心思路是抽象数据格式，让下游不依赖 Maxwell 特有的 JSON 结构：

方案一：中间件层统一转换（推荐）

[Maxwell/Canal/Flink CDC] → [统一格式适配器] → [标准 CdcData JSON] → Kafka → 下游服务

• 在 CDC Service 中定义统一的 CdcData 数据结构（database, table, type, data, old, xid, timestamp）
• Maxwell 的消息经过适配器转为 CdcData 再发 Kafka
• 替换框架时只需更换适配器，下游消费的 CdcData 格式不变

方案二：CDC Service 内部转换

• Maxwell 直接发原始 JSON 到 Kafka
• CDC Service 消费时解析 Maxwell JSON → 转为 CdcData → 再处理
• 替换框架时只需修改 CDC Service 内的解析逻辑

关键： 下游服务只依赖 CdcData 接口，不关心底层是 Maxwell 还是 Canal。

// 统一的 CDC 数据结构
public class CdcData {
    private String database;       // 数据库名称
    private String table;          // 表名称
    private String type;           // insert / update / delete
    private Map<String, Object> data;     // 变更后数据
    private Map<String, Object> old;      // 变更前数据（仅 update）
    private Long xid;              // 事务 ID
    private Long timestamp;        // 变更时间戳
}

17. 商品涉及多张关联表（SPU、SKU、属性、图片），如何在 ES 中维护一个完整的商品文档？

核心回答要点：

这是 CDC 多表同步到 ES 的经典问题。

ES 存储方案选择：

方案	说明	优点	缺点
宽表（嵌套对象）	一张大文档包含 SPU + SKU 列表 + 属性列表	查询简单，单次搜索即可	更新成本高（任一子表变更需重建整个文档）
父子文档	使用 ES 的 join 字段建立父子关系	子表变更只更新子文档	查询需要 join，性能差
反范式冗余	每个 SKU 存一个完整 SPU 信息的文档	查询快	数据冗余大，一致性难维护

本项目采用嵌套对象（宽表）方案：

public class GoodsDoc {
    private Long spuId;
    private String spuName;
    private String category;
    // ... SPU 字段
    private List<SkuDoc> skus;        // 嵌套对象
    private List<AttrDoc> attributes;  // 嵌套对象
}

当子表变更时（如 SKU 价格修改），CDC 消费端需要：

1. 从变更事件中提取关联的 SPU ID
2. 从 MySQL 查询该 SPU 的完整数据（包括所有 SKU、属性）
3. 重新构建 GoodsDoc 写入 ES（整体覆盖，ES 的 _id 基于 SPU ID 保证幂等）

关键挑战：

• 频繁的子表变更会导致大量的全量重建，需要防抖/合并（同 SPU 的多次变更合并为一次 ES 写入）
• 可以考虑引入延迟批量处理：收集同一 SPU 在 N 秒内的所有变更，合并后一次性重建文档

18. 如果业务要求 CDC 同步延迟不超过 1 秒，如何监控和保障？

核心回答要点：

监控指标：

端到端延迟 = Maxwell 解析延迟 + Kafka 传输延迟 + 消费处理延迟 + ES 写入延迟

1. Maxwell 侧：监控 binlog 读取位置与 MySQL 当前位置的差值（Maxwell 内部有心跳机制）
2. Kafka 侧：监控 Consumer Lag（kafka-consumer-groups.sh --describe），Lag × 消费速率 = 预估延迟
3. 消费服务侧：在 CdcData 中携带 timestamp，与当前时间差值即为端到端延迟

保障策略：

// 在消费端记录延迟指标
long syncDelayMs = System.currentTimeMillis() - cdcData.getTimestamp();
if (syncDelayMs > 1000) {
    log.warn("CDC 同步延迟超标: {}ms, database={}, table={}",
        syncDelayMs, cdcData.getDatabase(), cdcData.getTable());
    metrics.gauge("cdc.sync.delay.ms", syncDelayMs);
}

• Kafka 调优：linger.ms=0（不等待，立即发送），增加 partition 数量
• 消费端调优：增大 max.poll.records，ES 批量写入（bulk API）而非逐条
• 告警：设置 Consumer Lag 阈值、同步延迟阈值告警
• 兜底：定期全量校验（对比 MySQL 和 ES 的数据条数/关键字段）

---

高频追问链

以下是面试官可能的追问路径，练习时可以模拟完整对话：

追问链 A：Binlog 基础

"你知道 binlog 是什么吗？"
→ "binlog 有几种格式？"
→ "为什么 Maxwell 要求 ROW 格式？"
→ "ROW 格式的 binlog 内容长什么样？"
→ "如果 binlog 被误删了会怎样？"

追问链 B：数据一致性

"你如何保证 MySQL 和 ES 的数据一致？"
→ "是强一致还是最终一致？"
→ "如果 ES 写入失败了你怎么办？"
→ "那 Kafka 消息重复消费怎么处理？"
→ "如果消费了一半失败了，怎么保证幂等？"

追问链 C：性能与扩展

"商品表有几千万数据，怎么初始化到 ES？"
→ "全量导入需要多久？期间新增的数据怎么办？"
→ "如果 Bootstrap 过程中 MySQL 压力很大，怎么优化？"
→ "如果要同步到 ClickHouse 做分析，架构怎么改？"

---

项目亮点提炼（面试中主动展示）

1. 技术选型的思考过程：不是盲目选型，而是对比了 Maxwell/Canal/Flink CDC，基于 Bootstrap 能力做出决策
2. 无侵入式架构：通过中间件解耦，业务代码不感知 CDC 的存在
3. 最终一致性设计：理解分布式一致性不是追求强一致，而是通过幂等 + 重试 + 监控保障最终一致
4. 生产问题预判：提前考虑了 Topic 划分、消息积压、Bootstrap 性能、DDL 变更等生产问题
5. 可扩展性：CDC Service 的设计允许替换底层 CDC 框架而不影响下游