3. 增量同步原理：基于日志的增量捕获（CDC）、基于时间戳的增量抽取

增量同步，说白了就是只搬运变化的数据。全量同步像搬家，一次搬完所有东西；增量同步则像日常补货，只把新买的、过期的、坏掉的商品处理一下。这两种策略各有各的用武之地，但今天咱们重点聊聊增量同步的两种主流实现方式：CDC（Change Data Capture） 和 基于时间戳的增量抽取。

我个人习惯把增量同步比作「数据侦探」的工作——你得找到线索，证明哪些数据变了，然后顺着线索把变化抓出来。CDC 和时间戳，就是两种截然不同的找线索方式。

3.1 基于日志的增量捕获（CDC）

CDC 的核心思路是：不碰业务表，直接从数据库的事务日志里读变化。你想想看，数据库每做一次 INSERT、UPDATE、DELETE，都会在日志里留下记录。CDC 就是去「偷看」这些日志，把变化解析出来，再推给下游。

CDC 的典型工作流程：

  1. 源数据库写入事务日志（如 MySQL 的 binlog、PostgreSQL 的 WAL、Oracle 的 Redo Log）。
  2. CDC 工具（如 Debezium、Canal、Maxwell）实时读取日志，解析成结构化事件。
  3. 事件被发送到消息队列（Kafka、Pulsar）或直接写入目标系统。
  4. 目标系统按事件顺序回放，完成数据同步。

我在项目中遇到过好几次这样的场景：业务表每天几千万条写入，如果用时间戳去扫表，数据库 CPU 直接飙到 90%，业务方投诉电话打爆。换成 CDC 之后，源库几乎零感知，同步延迟从分钟级降到了秒级。嗯，这里要注意——CDC 不是银弹，它也有自己的脾气。

3.1.1 CDC 的三种常见实现模式

模式 原理 典型工具 延迟 对源库影响
日志解析器 伪装成从库，直接读 binlog/WAL Debezium, Canal 秒级 极低（只读日志）
触发器 + 日志表 在源表上建触发器，写操作时同步写入日志表 自研方案 实时 较高（额外写操作）
API 轮询 通过数据库 API 查询变更记录（如 Oracle LogMiner） Oracle GoldenGate 分钟级 中等

我个人最推荐第一种——日志解析器。它不侵入业务代码，不增加源库写压力，而且能捕获所有 DML 操作，包括 DELETE。我曾经见过一个团队用触发器方案，结果业务高峰期触发器嵌套太深，直接把事务回滚了，数据丢了整整两小时。从那以后，我对触发器方案就格外谨慎。

3.1.2 CDC 的典型架构

下面这张图展示了我最常用的一套 CDC 架构。它把日志读取、事件分发、数据落地拆成了三个独立环节，每个环节都可以水平扩展。

CDC 增量同步典型架构 源数据库 MySQL / PostgreSQL 事务日志 CDC 连接器 Debezium / Canal 消息队列 Kafka / Pulsar 流处理器 Flink / Kafka Streams 目标存储 HDFS / ClickHouse 💡 架构要点 • 日志读取与数据落地解耦，连接器只负责「读」和「发」 • 消息队列做缓冲，应对源库写入峰值和下游消费抖动 • 流处理器负责数据清洗、格式转换、幂等去重

我的经验之谈：CDC 连接器最好独立部署，不要和业务服务混在一起。我曾经见过有人把 Debezium 嵌在 Java 应用里，结果连接器 OOM 把整个服务拖垮了。独立部署之后，连接器挂了不影响业务，重启也快。

3.2 基于时间戳的增量抽取

时间戳方案就直白多了：在源表上加一个「最后修改时间」字段，每次同步时，只拉取时间戳大于上次同步时间的数据。说白了，就是让业务系统自己给数据打上「我变了」的标签。

这种方案的好处是简单、可控、不依赖第三方工具。但坑也不少。我记得有一次，业务方说「我们表里都有 update_time，直接用吧」，结果上线后发现大量数据漏同步——因为某些批量更新操作没走应用层，直接 DBA 在数据库里 UPDATE 了，update_time 没更新。嗯，这就是典型的「你以为你有了，其实你没有」。

3.2.1 时间戳方案的实现步骤

  1. 建时间戳字段：在源表增加 update_time（或 modified_at），每次 INSERT/UPDATE 时自动更新。
  2. 记录同步水位：在目标系统或元数据中心，记录上次成功同步的最大时间戳。
  3. 执行增量查询：SELECT * FROM source_table WHERE update_time > last_sync_time。
  4. 数据写入目标：对拉取到的数据进行清洗、转换、写入。
  5. 更新水位：将本次同步的最大时间戳写入元数据中心。

⚠️ 我曾经踩过的坑：

  • 时间精度问题：MySQL 的 datetime 精度到秒，同一秒内多条数据可能漏掉。建议用 datetime(6) 或 timestamp(6)。
  • 时钟回拨：如果源库服务器时间被回拨，新数据的时间戳可能小于旧数据，导致漏同步。建议用数据库自带的 SYSTIMESTAMP 或 NOW()，不要依赖应用服务器时间。
  • DELETE 操作无法捕获：时间戳方案只能捕获 INSERT 和 UPDATE，DELETE 操作不会留下时间戳。需要配合软删除（逻辑删除）或额外的 DELETE 日志表。

3.2.2 时间戳 vs CDC：怎么选？

很多同学问我：「到底该用时间戳还是 CDC？」我的回答是：看你的场景对延迟和侵入性的容忍度。下面这张表是我自己总结的选型参考：

对比维度 基于时间戳 基于 CDC
实现复杂度 低，只需加字段 + 写 SQL 高，需要部署连接器、消息队列
对源库侵入性 中，需要改表结构 低，只读日志不碰业务表
实时性 分钟级（取决于轮询频率） 秒级甚至毫秒级
捕获 DELETE 不支持（需额外方案） 原生支持
数据一致性 可能漏数据（时间精度、时钟问题） 高（基于事务日志，顺序保证）
运维成本 低，一个定时任务搞定 高，需要监控连接器、消息队列
适用场景 数据量不大、延迟要求不高、表结构可控 高并发、低延迟、数据一致性要求高

我的建议：

  • 如果源表数据量在百万级以下，同步周期可以接受 5-10 分钟，优先用时间戳方案。简单、稳定、好排查。
  • 如果源表每天几千万写入，或者业务要求秒级同步，直接上 CDC。前期投入大，但后期省心。
  • 如果两者都不满足……嗯，可以考虑混合方案：白天用 CDC 做实时同步，夜间用时间戳做全量对账，互相兜底。

3.3 两种方案的代码示例

光说不练假把式。我写两个简单的示例，帮你直观感受一下两者的区别。

3.3.1 基于时间戳的增量抽取（Python 伪代码）

import mysql.connector
from datetime import datetime

def incremental_sync(last_sync_time):
    conn = mysql.connector.connect(host='source_db', user='sync_user', password='***')
    cursor = conn.cursor(dictionary=True)
    
    # 拉取增量数据
    query = """
        SELECT * FROM orders 
        WHERE update_time > %s 
        ORDER BY update_time ASC
        LIMIT 1000
    """
    cursor.execute(query, (last_sync_time,))
    rows = cursor.fetchall()
    
    if not rows:
        print("没有增量数据")
        return last_sync_time
    
    # 写入目标系统（伪代码）
    for row in rows:
        write_to_target(row)
    
    # 更新水位
    new_sync_time = rows[-1]['update_time']
    update_watermark('orders', new_sync_time)
    
    print(f"同步完成，共 {len(rows)} 条，水位更新至 {new_sync_time}")
    return new_sync_time

# 定时调用
last_sync = datetime(2024, 1, 1, 0, 0, 0)
while True:
    last_sync = incremental_sync(last_sync)
    time.sleep(60)  # 每分钟轮询一次

3.3.2 基于 CDC 的增量捕获（Debezium + Kafka 配置片段）

# Debezium MySQL 连接器配置（JSON）
{
  "name": "mysql-connector-orders",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "source_db",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "***",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "ecommerce",
    "table.include.list": "ecommerce.orders",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.ecommerce",
    "include.schema.changes": "false",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "false"
  }
}

# 消费端（Kafka Console Consumer）查看变更事件
# kafka-console-consumer --bootstrap-server kafka:9092 \
#   --topic dbserver1.ecommerce.orders \
#   --from-beginning \
#   --property print.key=true \
#   --property print.value=true

小提示：CDC 配置里的 snapshot.mode 很关键。第一次启动时，Debezium 会先做一次全量快照，把表里已有的数据全部读出来，然后再切到增量模式。这个「全量→增量」的无缝衔接，正是咱们这门课的核心主题。后面几章我会专门讲这个切换的细节。

3.4 避坑指南：我踩过的那些坑

做增量同步这么多年，我踩过的坑能写一本小册子。挑几个最典型的分享给你：

  • 时间戳字段没建索引：有一次我接手一个项目，orders 表 5000 万行，每次增量查询都全表扫描，一次同步跑 20 分钟。加了个 INDEX(update_time)，直接降到 3 秒。你想想看，一个索引能省多少事。
  • CDC 连接器挂了没人知道：Debezium 连接器如果挂了，Kafka 里就收不到新事件了。但源库还在正常写入，下游系统以为数据没变，一直用旧数据。后来我加了监控告警：如果 Kafka topic 的消费延迟超过 5 分钟，立刻钉钉报警。
  • 事务日志被清理：MySQL 的 binlog 默认只保留几天。如果 CDC 连接器因为故障停了超过这个时间，重启后会发现 binlog 已经被 purge 了，只能重新做全量快照。我的建议是：把 binlog 保留时间设长一点（比如 7 天），同时做好连接器的高可用。
  • 时间戳方案遇到跨天问题：如果同步任务在 23:59:59 执行，拉到了 23:59:58 的数据，但同一秒还有一条 23:59:59 的数据没被拉取。下次同步时，水位是 23:59:58，那条 23:59:59 的数据就永远漏掉了。解决方案：水位用 >= 配合去重，或者把水位往前调 1 秒做冗余。

⚠️ 最重要的一个教训：无论用哪种方案，一定要做数据对账。我见过太多团队上了增量同步之后就不管了，结果某天发现目标库和源库差了 10 万条数据，根本不知道从什么时候开始漏的。我的习惯是：每天凌晨跑一次全量对账，把源库和目标库的 count、checksum 对比一遍，有差异立刻告警。

← 上一章 📖 返回目录 下一章 →