数据同步

用变更数据捕获、冲突解决和回填保持两系统同步。

为什么

当应用跨两数据库（如 Postgres 事务、Elasticsearch 搜索或 CRM 镜像），需一者变更出现在另一者。每 5 分钟轮询慢且负载重；通过 Postgres 逻辑复制或 Debezium 的变更数据捕获（CDC）实时流每行变更。冲突解决（最后写赢或协作字段的 CRDT）防两系统在都可写时分叉。回填是不光鲜的一半：加新同步目标时需先复制所有现有数据再让流接管。无文档化同步策略，团队最终有'基本能用'的 cron job 和没人懂的凌晨 3 点对账脚本。

何时使用

有 ≥2 系统须反映同数据时。CDC 约 1-3 天。单数据库应用可跳过。

如何做

1. 选 CDC 源：Postgres 逻辑复制（wal2json）或 Debezium
   - Postgres 启 wal_level=logical，创建复制槽
   - Debezium 把行级 insert/update/delete 流为 Kafka 事件

2. 设消费者：Kafka Connect sink 或自定义流处理器
   - ES：sink 把每变更写搜索索引
   - CRM：把行转 CRM schema 再 upsert

3. 为双向同步定义冲突解决
   - 单向（DB→搜索）：无冲突，直接应用
   - 双向：用 updated_at 时间戳（最后写赢）或字段级 CRDT

4. 回填：启动流前复制现有数据
   - 跑批量导出→导入 job（分页、可恢复）
   - 回填完成后启动 CDC 流，用槽的起始位置

5. 监控延迟：测源提交到 sink 应用的延迟
   - 延迟 >60s 告警——卡住的消费者意味搜索索引过期
   - 暴露 /sync/status 端点显示最后同步 LSN 和延迟

示例

-- 启用 Postgres 逻辑复制
ALTER SYSTEM SET wal_level = 'logical';
SELECT pg_create_logical_replication_slot('sync_slot', 'wal2json');

-- Debezium 连接器配置（Kafka Connect）
{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg.internal",
    "database.dbname": "app",
    "table.include.list": "public.users,public.orders",
    "plugin.name": "pgoutput"
  }
}

-- 回填（启动连接器前跑）
COPY (SELECT * FROM users WHERE updated_at < '2026-01-01') TO '/tmp/users.csv';
-- 然后批量导入 ES，再从槽位置启动连接器

常见问题

⚠回填前启动 CDC 流，目标漏所有预存行。
⚠流中不处理删除，删除行在搜索索引留幽灵。
⚠双向同步无冲突解决，两系统循环互覆。
⚠消费者宕时忽略复制槽增长——Postgres WAL 增到磁盘满。

最佳实践

✓用带 max_slot_wal_keep_size 的复制槽，宕消费者不能填满磁盘。
✓在流处理器而非源里转换——保持 DB schema 无同步关注。
✓跑每日对账 job 数源和 sink 行；漂移 >0.1% 告警。
✓同步 schema 版本化：源加列时消费者须处理新旧事件形状。