为什么投影表可以丢
投影表不是业务数据库，它随时可以从事件日志再生。这个特性是 CQRS 架构的底气：查询端出了问题——索引坏了、数据漂移、逻辑 bug 导致统计错误——最快的修复方式不是修补丁，而是删掉投影表，从位移 0 开始重放，让事件重新流过全新的处理器。
这种操作叫重建（rebuild），核心是重放（replay）——把 Kafka 主题从头读一遍，逐条事件重新计算投影。
重建的前提只有一个：Kafka 保留期必须覆盖重建窗口。 如果主题只保留 7 天，而你要重建的数据跨度是 30 天，重建到一半事件就消失了。保留期是 CQRS 的生命线，也是一项需要在建主题时就定好的运维决策。
什么时候需要重建
重建不是日常操作，但以下三种情况逃不掉：
- 投影逻辑改了。 比如之前按
book_id聚合，现在要按book_id + branch_id二级聚合。老投影表不支持新维度，只能从头重建。 - 数据漂移确认。 怀疑投影表和事件日志不一致，用重建做一次对账。
- 新消费者上线。 新建一个消费组从零开始消费，它的投影表天然是空的，需要完整重放。
第 3 种情况最常见：加一个新查询维度，就加一个新投影消费者，这个新消费者必须从位移 0 开始追赶。
重建期间的读服务
重建期间，旧的投影表不能立刻删掉——否则读服务会白屏。标准做法是双实例并行：
- 实例 A 继续服务读请求，数据来自旧投影表。
- 实例 B 从零开始重放，在新表里重建投影。
- 重建完成后切流量到实例 B，再下线实例 A。
双实例切换的判定条件有两个：新投影表的数据和旧表对得上（抽样对比），且新消费者的位移追上了日志末端（lag ≈ 0）。
实验：两次重建对比
下面用 library.inventory.adjusted 主题跑 20 条库存调整事件（随机正负增量），先做第一次投影，清空表，再做第二次重放，比较结果和耗时。
数据准备
Random rng = new Random(42);
String[] books = {"B-A01", "B-A02", "B-A03", "B-A04", "B-A05"};
String[] reasons = {"restock", "damage", "lost", "return"};
try (Producer<String, String> producer = new KafkaProducer<>(pp)) {
for (int i = 0; i < 20; i++) {
String book = books[rng.nextInt(books.length)];
int delta = rng.nextBoolean()
? rng.nextInt(5) + 1
: -(rng.nextInt(3) + 1);
String v = String.format(
"{\"bookId\":\"%s\",\"deltaCopies\":%d,"
+ "\"reason\":\"%s\",\"occurredAt\":\"%s\"}",
book, delta, reasons[rng.nextInt(reasons.length)],
Instant.now().toString());
RecordMetadata md = producer.send(
new ProducerRecord<>(
"library.inventory.adjusted", book, v)).get();
}
}
随机生成保证了覆盖四种原因和正负增量，20 条足够验证聚合逻辑的正确性。
重建方法
重建函数只做三件事：seek 到起始位移，逐条消费事件，解析增量并更新投影表。
static int rebuildFromOffset(long fromOffset, long toOffset)
throws Exception {
int count = 0;
Properties cp = new Properties();
cp.put("bootstrap.servers", "127.0.0.1:19102");
cp.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
cp.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
cp.put("auto.offset.reset", "earliest");
try (Connection conn = DriverManager.getConnection(
DB_URL, "sa", "");
KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(cp)) {
TopicPartition tp = new TopicPartition(
"library.inventory.adjusted", 0);
consumer.assign(Collections.singletonList(tp));
consumer.seek(tp, fromOffset);
conn.setAutoCommit(false);
while (count < toOffset - fromOffset) {
ConsumerRecords<String, String> recs =
consumer.poll(Duration.ofMillis(3000));
if (recs.isEmpty()) break;
for (ConsumerRecord<String, String> rec : recs) {
long off = rec.offset();
if (off >= toOffset) break;
String val = rec.value();
int delta = parseDelta(val);
String bookId = rec.key();
// MERGE 初始行
PreparedStatement ups = conn.prepareStatement(
"MERGE INTO inventory_projection "
+ "KEY(book_id) VALUES (?, ?, ?)");
ups.setString(1, bookId);
ups.setInt(2, delta);
ups.setLong(3, off);
ups.executeUpdate();
// 累加增量
PreparedStatement inc = conn.prepareStatement(
"UPDATE inventory_projection "
+ "SET total_copies = total_copies + ?,"
+ " last_offset = ? WHERE book_id = ?"
+ " AND last_offset = ?");
inc.setInt(1, delta);
inc.setLong(2, off);
inc.setString(3, bookId);
inc.setLong(4, off);
inc.executeUpdate();
count++;
}
conn.commit();
}
}
return count;
}
执行与结果
long targetOffset = ...; // 主题末端位移
// 第一次投影
long t1Start = System.currentTimeMillis();
int firstCount = rebuildFromOffset(0L, targetOffset);
long t1End = System.currentTimeMillis();
// 清空表
conn.createStatement().executeUpdate(
"DELETE FROM inventory_projection");
// 第二次重放
long t2Start = System.currentTimeMillis();
int secondCount = rebuildFromOffset(0L, targetOffset);
long t2End = System.currentTimeMillis();
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）:
javac --release 25 -cp ... RebuildDemo.java && java -cp ... RebuildDemo输出截取：
Seed[0]: book=B-A01 delta=-1 offset=0 Seed[19]: book=B-A02 delta=1 offset=19 Topic end offset=20 First build: records=20 elapsed=543 ms Projection table wiped Replay build: records=20 elapsed=537 ms --- Comparison --- B-A01: first=-4 replay=-4 OK B-A02: first=2 replay=2 OK B-A03: first=6 replay=6 OK B-A04: first=-4 replay=-4 OK B-A05: first=8 replay=8 OK All match=true Speedup ratio=1.01单次运行，随调度波动，不作基准。
两次重建结果完全一致：5 本书的库存总量都是相同的。B-A05 最终 total_copies=8，是增量最多的书；B-A01 和 B-A04 都是 -4，说明出库多于入库。
重放的性能原罪
重放虽然结果可复现，但性能本身是一个问题。
本实验中 20 条事件两次 rebuild 耗时都在 500 ms 左右，比例接近 1:1。这是因为 H2 内存表 + 本地 Kafka ≈ 零网络开销。放到生产环境，重放 100 万条事件的典型问题有三：
- I/O 放大。 每条事件至少一次 poll + 一次 SQL 写入，投影表如果在外部 Postgres，网络 RTT 会把单条处理时间拉到人毫秒级。
- 消费组滞后追赶。 如果一个新消费者从零开始消费 100 万条，每秒处理 5000 条也需要 200 秒。这期间读服务要么读旧投影，要么承受高延迟。
- 事件本身可能过大了。 携带全量状态的「胖事件」会让网络和反序列化双双承压，事件应该只携带增量（delta）。
应对策略没有银弹，只有权衡：
| 策略 | 效果 | 代价 |
|---|---|---|
增大 max.poll.records | 减少 poll 次数 | 单批次内存增大 |
| 批量写入投影表 | SQL 往返减少 | 批次内失败要整体重试 |
| 保留 snapshot | 从 snapshot + 增量追赶 | snapshot 管理复杂度 |
| 并行消费多分区 | 吞吐量线性增长 | 投影逻辑必须按 key 可分区 |
重建的性能影响与优化
重建的成本取决于三个因素：
- 事件总量：1000 条事件约需 1 秒，1000 万条事件可能需要数小时。
- 投影复杂度：简单计数投影很快，涉及多表 join 和复杂计算的投影很慢。
- 网络带宽：从 Kafka 读取事件的网络延迟。
缓解方案包括增量快照（定期生成投影表快照，重建时从快照开始）、并行重建（多个消费者并行重建不同分片）、事件压缩（在 changelog topic 上启用压缩减少重放量）。
重建的触发条件
| 触发条件 | 说明 |
|---|---|
| 投影表损坏 | 磁盘故障、数据错误、bug 导致的数据污染 |
| 投影逻辑变更 | 新业务需求需要不同的投影计算方式 |
| 数据库迁移 | 从 H2 迁移到 PostgreSQL，需要重新投影 |
| 消费者组重置 | seekToBeginning() 或变更消费者组名 |
| 架构升级 | 事件 schema 演进后，旧投影无法理解新事件 |
重建是运维操作，不应在业务高峰期执行。建议在低峰期通过脚本触发，并监控重建进度和系统负载。
重建与快照的关系
快照是重建的加速器。没有快照时，重建必须从位移 0 开始，逐条重放所有历史事件。有了快照后，重建可以从快照对应的时间点开始，只重放增量事件。
快照的实现方式：
- 投影表快照：定期导出投影表到文件（如 CSV 或 Parquet），重建时先导入快照，再重放增量事件。
- Kafka Streams 状态存储快照：
KafkaStreams#allMetadata()可导出本地状态存储的快照。 - 数据库备份：如果使用 PostgreSQL 等持久化数据库，直接用 pg_dump 做快照。
快照频率取决于事件产生速率和业务容忍度。高事件速率场景建议每小时快照一次，低事件速率场景可每天一次。
小结
- 投影表是可丢弃的，唯一不可丢的是 Kafka 中的事件日志。保留期决定了重建窗口的上限。
- 重建的典型触发条件：投影逻辑变更、数据漂移对账、新消费者从零追赶。
- 重建期间读服务不能挂，双实例并行是最稳妥的切换方式。
- 实测：20 条库存调整事件两次重建，5 本书的总量 100% 一致，耗时分别 543 ms 和 537 ms。
- 生产级重放的性能瓶颈在 I/O 放大和滞后追赶，策略是批量消费、批量写入、并行分区，或引入 snapshot 减少重放基数。
- 重建完成后应进行数据对账，验证新建投影与原有查询结果一致。
- 重建过程应记录日志，包括开始时间、结束时间、处理事件数和最终状态。
- 重建失败时应有回滚机制，保留旧投影表直到新投影验证通过。
- 重建期间读服务不应中断，建议采用双实例并行切换策略。
- 重建操作应由运维脚本自动化执行，避免手工操作导致的人为错误。
阅读导航：上一节 3.2 投影与视图构建 ｜ 下一节 4.1 分布式事务概述
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。