---
id: scenario.data-sync
title: 数据同步：业务库到搜索引擎 / 向量库 / 报表库，outbox 与 CDC、幂等、顺序、对账和重建
type: scenario
domain: cross
stack: [data-pipeline, postgresql, mongodb, elasticsearch, vector-db, mq]
tags: [数据同步, outbox, CDC, 变更流, 幂等, 顺序, 对账, 全量重建]
level: advanced
prerequisites: [backend.data-pipeline.index, backend.mq.index]
est_hours: 3
related: [backend.elasticsearch.scenario.kb-search-sync, backend.data-pipeline.scenario.third-party-sync, backend.mongodb.scenario.conversation-store, backend.mq.reliability]
week: [6, 14]
---

# 数据同步：业务库到搜索引擎 / 向量库 / 报表库，outbox 与 CDC、幂等、顺序、对账和重建

## 业务背景与问题

业务数据在 PostgreSQL（订单、知识库文档）和 MongoDB（对话记录），需要同步到：ES（全文检索）、向量库（RAG）、报表库（报表 Agent）。现状是业务代码里“写完数据库顺手写 ES”：

1. 写 ES 失败时数据库已提交，两边不一致；
2. 同一条数据快速修改两次，ES 中留下的是旧版本；
3. 删除没有同步，搜索出已下架的商品；
4. 修改映射或换嵌入模型时不知道怎么重建。

## 约束

- 允许秒级到分钟级的延迟。
- 不能明显增加业务写入的耗时。
- 下游可以从源头完全重建（源头是唯一可信的数据）。

## 方案设计

1. **捕获变更**（二选一或组合）：
   - **outbox 模式**：业务事务中同时写 outbox 表（事件类型、实体 id、版本号），后台进程投递到消息队列 → 保证“数据库提交 ⇔ 有事件”；
   - **CDC**：读取数据库日志（PostgreSQL 逻辑复制 / Debezium，MongoDB 变更流），业务代码零侵入。
2. **消费与写入**：
   - 消息只带 id 和版本，消费者回源读取最新数据再写下游（避免消息里的旧快照覆盖新数据）；
   - 下游文档 id = 业务主键，写入带外部版本号，旧版本被拒绝；
   - 删除事件 → 删除下游文档（或标记删除）；
   - 同一实体的事件按实体 id 分区，保证顺序；
   - 失败重试、死信、告警。
3. **对账**：每天比较源头与下游的数量和版本（按分片哈希），差异自动修复。
4. **全量重建**：新建索引 / 集合 → 全量导入（记录开始时间）→ 回放该时间之后的增量 → 校验 → 切换别名 → 保留旧索引回滚。
5. **监控**：同步延迟（事件产生到写入下游的时间）、积压量、失败数、对账差异。

## 关键代码

outbox（同一事务中写业务数据和事件）：

```sql
BEGIN;
UPDATE documents SET body = $2, version = version + 1, updated_at = now() WHERE id = $1;
INSERT INTO outbox (aggregate, aggregate_id, event, version)
SELECT 'document', id, 'updated', version FROM documents WHERE id = $1;
COMMIT;
```

消费者：回源 + 外部版本号（Elasticsearch Python 客户端，示意）：

```python
def handle(event: dict) -> None:
    doc = repo.get(event["aggregate_id"])                 # 回源读取最新数据
    if doc is None or doc.deleted:
        es.options(ignore_status=404).delete(index="kb", id=event["aggregate_id"])
        return
    es.index(index="kb", id=doc.id, document=to_search_doc(doc),
             version=doc.version, version_type="external")   # 版本小于等于已有版本时 ES 返回冲突，直接忽略
```

基础组件的可运行示例：[增量同步管道](../topics/backend/data-pipeline/examples/incremental-sync/)、[MongoDB 变更流](../topics/backend/mongodb/examples/conversations/)、[ES 别名切换](../topics/backend/elasticsearch/examples/hybrid-search/)。

## 取舍

| 决策 | outbox | CDC |
| --- | --- | --- |
| 侵入性 | 需要改业务代码 | 业务零侵入 |
| 事件语义 | 业务事件，含义清晰 | 行级变更，需要转换 |
| 运维 | 简单 | 需要部署和维护 CDC 组件、处理日志保留 |
| 适合 | 中小团队、事件少 | 表多、多个下游 |

## 易错点

- 业务代码中双写（数据库 + ES），没有事务保证。
- 消息中带完整快照，乱序时旧快照覆盖新数据。
- 忘记同步删除和权限变更。
- 重建期间的增量没有回放，切换后丢数据。
- 没有同步延迟监控，下游落后几小时才发现。

## 面试怎么讲

> “我们把‘写库顺手写 ES’改成了 outbox：业务事务里同时写 outbox 表，后台投递到队列，保证提交和事件一致。消费者只拿 id 回源读最新数据，用业务主键作为文档 id、版本号作为外部版本，乱序和重复都不会让旧数据覆盖新数据；删除也同步。每天按分片哈希对账；换映射或嵌入模型时新建索引全量导入、回放增量、校验后切别名。监控同步延迟和积压。”