投影不是缓存，是派生视图
事件日志是一个 append-only 的序列，每次追加都表示一次状态变更。投影（projection）的任务就是把这个序列转换成一张可查询的表。
与缓存的区别在于三点：
- 投影是从事件派生，不是从命令模型派生。 缓存可以刷自命令端的 MySQL 表，投影必须消费事件流。这意味着投影的语义是「事件发生过什么」，不是「当前数据库状态是什么」。
- 投影可以重建。 缓存丢了只能回源重填，投影可以从 Kafka 位移 0 开始重放，逐条事件重新计算。
- 投影可以有多份。 同一个
library.loan.created主题，借阅团队投影成「按书的活跃借阅数」，通知团队投影成「需要提醒的逾期列表」。事件是素材，投影是制品。
投影处理器的基本结构
一个投影处理器只做四件事：
- 消费事件（从 Kafka 拉取）。
- 幂等写入（防止重复消费导致计数翻倍）。
- 更新投影表（按事件语义做加/减/设值）。
- 持久化位移（知道自己处理到哪了）。
Kafka topic ---> Consumer ---> 幂等检查 ---> 投影表更新 ---> Offset commit
^ |
+-------- 重试 / 跳过异常 ---------------+
第 2 步是关键。Kafka 消费者的语义是「至少一次」，不幂等的投影一旦重消费，统计数字就会偏。常见做法是在投影表里额外维护一个 proc_offset 字段，收到事件时先比大小，只处理比自己记录大的位移。
实验：从 Kafka 到 H2 投影
下面用 library.loan.created 主题生产三条事件（B-100 出现两次），然后用裸消费者把它们投影到 H2 的 active_loan_projection 表，验证幂等性和聚合正确性。
表结构
// 投影表：每本书的当前活跃借阅数
s.execute("CREATE TABLE active_loan_projection ("
+ "book_id VARCHAR(36) PRIMARY KEY,"
+ "active_count INT DEFAULT 0,"
+ "last_event_at TIMESTAMP,"
+ "proc_offset BIGINT)");
// 事件日志表：记录已消费的事件位移（幂等用）
s.execute("CREATE TABLE event_log ("
+ "offset_id BIGINT PRIMARY KEY,"
+ "topic_name VARCHAR(80),"
+ "evt_key VARCHAR(36),"
+ "payload VARCHAR(4000),"
+ "received_at TIMESTAMP)");
event_log 不是业务表，是投影处理器的内部去重簿。如果同一个位移被消费两次，第二次查到 offset_id 已存在就直接跳过。
生产测试事件
String[] books = {"B-100", "B-101", "B-100"}; // B-100 有两笔借阅
String[] loans = {"L-200", "L-201", "L-202"};
Properties pp = new Properties();
pp.put("bootstrap.servers", "127.0.0.1:19102");
pp.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
pp.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
pp.put("acks", "1");
try (Producer<String, String> producer = new KafkaProducer<>(pp)) {
for (int i = 0; i < 3; i++) {
String v = String.format(
"{\"loanId\":\"%s\",\"bookId\":\"%s\","
+ "\"memberId\":\"%s\",\"occurredAt\":\"%s\"}",
loans[i], books[i], members[i], Instant.now());
RecordMetadata md = producer.send(
new ProducerRecord<>("library.loan.created", books[i], v)
).get();
}
}
消费与投影
消费者用 assign 直接绑定分区，省去消费组再平衡的开销（适合单节点实验）。每次 poll 到记录后，先查 event_log 做幂等，再按「存在则更新 / 不存在则插入」两条路径处理投影表。
TopicPartition tp = new TopicPartition("library.loan.created", 0);
consumer.assign(Collections.singletonList(tp));
consumer.seekToEnd(Collections.singletonList(tp));
long endOffset = consumer.position(tp);
long startOffset = endOffset - 3; // 只读本次生产的 3 条
consumer.seek(tp, Math.max(0, startOffset));
conn.setAutoCommit(false);
// 循环 poll
for (int batch = 0; batch < 5; batch++) {
ConsumerRecords<String, String> recs =
consumer.poll(Duration.ofMillis(3000));
if (recs.isEmpty()) break;
for (ConsumerRecord<String, String> rec : recs) {
long off = rec.offset();
String key = rec.key(); // book_id
// 幂等：查 offset_id 是否已存在
PreparedStatement check = conn.prepareStatement(
"SELECT 1 FROM event_log WHERE offset_id = ?");
check.setLong(1, off);
if (check.executeQuery().next()) {
System.out.println("Skip dup offset=" + off);
continue;
}
// 记录事件日志
PreparedStatement ins = conn.prepareStatement(
"INSERT INTO event_log VALUES (?,?,?,?,?)");
ins.setLong(1, off);
ins.setString(2, rec.topic());
ins.setString(3, key);
ins.setString(4, rec.value());
ins.setTimestamp(5, Timestamp.from(Instant.now()));
ins.executeUpdate();
// 投影更新
PreparedStatement exists = conn.prepareStatement(
"SELECT active_count, proc_offset "
+ "FROM active_loan_projection WHERE book_id = ?");
exists.setString(1, key);
ResultSet ers = exists.executeQuery();
if (ers.next()) {
// 已有记录：活跃数 +1
PreparedStatement upd = conn.prepareStatement(
"UPDATE active_loan_projection "
+ "SET active_count = active_count + 1,"
+ " last_event_at = ?, proc_offset = ? "
+ "WHERE book_id = ?");
upd.setTimestamp(1, Timestamp.from(Instant.now()));
upd.setLong(2, off);
upd.setString(3, key);
upd.executeUpdate();
} else {
// 新记录：插入初始值
PreparedStatement insRow = conn.prepareStatement(
"INSERT INTO active_loan_projection "
+ "VALUES (?, 1, ?, ?)");
insRow.setString(1, key);
insRow.setTimestamp(2, Timestamp.from(Instant.now()));
insRow.setLong(3, off);
insRow.executeUpdate();
}
}
conn.commit();
}
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）:
javac --release 25 -cp ... ProjectionDemo.java && java -cp ... ProjectionDemo输出截取：
Produced: book=B-100 offset=7 Produced: book=B-101 offset=8 Produced: book=B-100 offset=9 Inserted offset=7 key=B-100 Inserted offset=8 key=B-101 Updated offset=9 key=B-100 rows=1 --- Projection --- book_id=B-100 active_count=2 proc_offset=9 book_id=B-101 active_count=1 proc_offset=8 Events logged=3 Consumer elapsed=3118 ms Last offset=9单次运行，随调度波动，不作基准。
注意 B-100 出现了两次，投影表的 active_count=2 正好对应两条独立借阅。第二次消费到同一 book 时没有重复插入，而是走了 UPDATE，幂等机制保证了只增不减。
投影的延迟
投影不是瞬时的。延迟来自三段：
- 生产延迟。 命令端写库后发出事件的网络往返。
- 消费延迟。 消费者 poll 间隔、批次攒积、再平衡等待。
- 写入延迟。 投影数据库的事务提交时间。
本实验中消费者是手工 assign 分区后直接 seek 到目标位移，中间没有消费组再平衡，所以延迟主要来自 poll 超时（本例中 3 秒 timeout 内就捕获了全部 3 条）。在生产环境里，如果投影处理器和生产者之间隔着 lag，延迟会达到秒级甚至分钟级。
这个延迟就是 CQRS 的代价：查询模型永远比命令模型慢一拍。 业务方必须提前确认这一拍是否可接受。
投影写在哪里
投影的目标存储没有唯一答案，取决于查询模式：
| 存储 | 适用场景 | 重建方式 |
|---|---|---|
| H2/Derby 内存表 | 实验、测试、轻量查询 | 重放 Kafka 事件流 |
| PostgreSQL | 复杂查询、需要 JOIN 和索引 | 重放重建或逻辑复制 |
| Redis | 热点数据、QPS 极高 | 全量重放或 snapshot 恢复 |
| Elasticsearch | 全文检索、聚合分析 | 全量重放 |
| Kafka Streams 状态存储 | 流内聚合、KTable | changelog 主题自动恢复 |
同一个事件主题可以有多个投影消费者各写各的存储，互不影响。这是 CQRS 最灵活的地方——「一份事件，多份视图」。
投影的错误处理与恢复
投影处理器在消费过程中可能遇到三类错误：
- 投影表写入失败：数据库连接超时、磁盘满、主键冲突（幂等逻辑 bug）。
- 事件解析失败：事件 schema 演进后，旧版消费者不认识新字段。
- 消费者自身崩溃：JVM OOM、GC 停顿超过
max.poll.interval.ms。
处理策略：
- 可重试错误（网络超时、连接断开）：指数退避重试，最多 3 次，仍失败则跳过该事件并记录死信。
- 不可重试错误（解析失败、业务规则违背）：直接跳过，记录到
projection_dlq表，由人工介入处理。 - 消费者崩溃：重启后从上次提交的 offset 继续消费。关键：如果崩溃前已写入投影但未提交 offset，重启后会重复消费，此时幂等机制必须生效。
一个实用的监控指标是「投影 lag」：Kafka 消费者组的 lag（未消费消息数）持续上升，说明投影处理器处理不过来。lag 阈值应设告警，例如超过 1000 条时触发。
投影的测试策略
投影处理器的单元测试应该验证三件事：
- 幂等性：同一事件处理两次，投影结果不变。
- 顺序敏感事件的正确性：版本号递增的事件必须按顺序处理。
- 异常事件的跳过：未知事件类型不抛异常，不破坏已有投影。
测试用例示例：
@Test
void idempotentProjection() {
projection.handle(new LoanCreated("L001", "B100"));
projection.handle(new LoanCreated("L001", "B100")); // 重复
assertEquals(1, projection.countByBook("B100"));
}
（本实验未在本机执行，仅作为预期行为说明。）
小结
- 投影是从事件流到查询模型的单向转换，不是缓存，可以重建。
- 幂等写入是投影的底线：同一事件消费多次不能导致计数错误。
- 实测：3 条 Kafka 事件（含同一 book 的重复事件）投影到 H2，
B-100的active_count=2，B-101的active_count=1。 - 投影延迟来自生产、消费、写入三段，查询模型永远比命令模型慢一拍。
- 投影目标按查询模式选：关系型数据库做结构化查询，Redis 做热点，ES 做搜索，Streams 状态存储做流内聚合。
阅读导航：上一节 3.1 CQRS 基础 ｜ 下一节 3.3 模型重建与重放
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。