4. 迁移流水线总体架构：源数据库、变更捕获、转换引擎、目标数据库、监控与回滚

数据库迁移流水线并非简单的数据搬运工具，而是一个具备可观测性、可逆性、可重复性的工程系统。本章从领域驱动设计的视角，将流水线拆解为五个核心领域：源数据库、变更捕获、转换引擎、目标数据库、监控与回滚。每个领域承担明确的职责，并通过接口契约进行交互。

4.1 五阶段架构总览

下图展示了迁移流水线的完整拓扑结构，数据从左至右流动，控制流与监控流从右至左反馈。

4.2 源数据库领域

源数据库是流水线的起点，负责提供存量数据快照与增量变更日志。该领域需要解决的核心问题包括：

• 连接管理：维护连接池，支持主从切换、只读实例路由。
• 快照一致性：通过全局读锁或 MVCC 快照，保证存量数据的一致性视图。
• 日志保留策略：确保 Binlog / WAL 保留时间足够长，避免增量阶段因日志过期而丢失变更。

4.3 变更捕获领域

变更捕获（CDC）是流水线的数据感知层。它监听源数据库的变更事件，将其转化为统一的事件格式，并推送到下游。该领域包含两个子阶段：

1. 日志读取器（Log Reader）：从源数据库读取原始变更日志，维护偏移量（Offset）。
2. 事件转换器（Event Converter）：将原始日志转换为标准化的 ChangeEvent 消息。

// 标准化 ChangeEvent 结构（伪代码）
{
  "source": {
    "db": "order_db",
    "table": "orders",
    "server_id": 123,
    "timestamp_ms": 1699000000000
  },
  "before": { "id": 1001, "status": "PENDING", "amount": 299.00 },
  "after":  { "id": 1001, "status": "PAID",   "amount": 299.00 },
  "op": "u",  // c=create, u=update, d=delete, r=snapshot
  "ts_ms": 1699000001000
}

4.4 转换引擎领域

转换引擎是流水线的业务逻辑核心。它接收标准化事件，执行以下操作：

• Schema 映射：将源表结构映射到目标表结构，处理字段名变更、类型转换、默认值填充。
• 数据清洗：过滤无效数据、去重、格式标准化。
• 路由分发：根据业务规则将数据路由到不同的目标表或目标集群。
• 事务合并：将多个细粒度事件合并为批量写入，提升目标数据库吞吐量。

转换引擎通常采用无状态设计，便于水平扩展。状态（如偏移量、计数器）统一存储在外部中间件（如 Redis 或 ZooKeeper）中。

4.5 目标数据库领域

目标数据库是流水线的终点，负责高效写入与数据校验。该领域需要处理：

4.6 监控与回滚领域

监控与回滚是流水线的安全保障层。它不直接参与数据流动，但贯穿整个流水线生命周期。核心职责包括：

• 延迟监控：实时计算源端事件产生时间与目标端写入时间的差值，超过阈值时告警。
• 数据校验：周期性对源端与目标端的数据进行抽样对比，发现不一致时触发修复流程。
• 断点续传：记录每个阶段的处理进度（Offset），在组件重启后从断点继续。
• 回滚指令：当检测到严重数据错误或业务方发起回滚请求时，停止写入并启动回滚流程。

下图展示了监控与回滚领域内部的组件协作关系：

4.7 接口契约定义

流水线各阶段之间通过明确的接口契约进行解耦。以下是核心接口定义：

// 阶段间事件接口（基于 Apache Avro / Protobuf）
syntax = "proto3";

message ChangeEvent {
  string source_id = 1;          // 源数据库实例标识
  string table_name = 2;         // 表名
  int64  timestamp = 3;          // 事件产生时间（毫秒）
  string op_type = 4;            // c / u / d / r
  bytes  before = 5;             // 变更前数据（JSON 或 Avro 编码）
  bytes  after = 6;              // 变更后数据
  int64  offset = 7;             // 日志偏移量
  map<string, string> headers = 8; // 扩展头（如事务 ID、路由标签）
}

// 监控上报接口
message MonitorMetric {
  string pipeline_id = 1;
  string stage_name = 2;         // source / capture / transform / target
  int64  lag_ms = 3;             // 延迟毫秒数
  int64  processed_count = 4;    // 已处理事件数
  int64  error_count = 5;        // 错误事件数
  double throughput = 6;         // 当前吞吐量（events/s）
}

// 回滚指令接口
message RollbackCommand {
  string pipeline_id = 1;
  int64  rollback_to_offset = 2; // 回滚到指定偏移量
  string reason = 3;             // 回滚原因
  bool   dry_run = 4;            // 是否为预演模式
}

4.8 架构决策记录（ADR）

下表总结了本章架构设计中的关键决策及其依据：