事件存储的角色
事件存储是事件溯源的底盘。它不只负责存储事件，还必须保证：
- 同一聚合根的事件顺序性。 一个订单的
Created、Paid、Shipped不能乱序，否则重建出来的状态就是错误的。 - 不可变。 已落盘的事件不能修改，只能追加。这是重建可复现性的前提。
- 可遍历。 从任意位移开始按顺序读取到任意位移结束，且遍历过程不能跳过中间事件。
- 全局唯一标识。 每条事件至少在一个聚合根的范围内有唯一序号。
这四条约束把「事件存储」与「普通消息队列」区分开。消息队列通常允许消费者乱序消费、跳过位移、甚至截断日志；事件存储必须把顺序和完整性放在第一位。
自建 vs 使用 Kafka
自建事件存储
自建方案通常是一个关系型数据库（Postgres 或 H2）的扩展，表结构类似：
CREATE TABLE events (
aggregate_id VARCHAR(64) NOT NULL,
seq INT NOT NULL,
event_type VARCHAR(64) NOT NULL,
payload JSONB NOT NULL,
recorded_at TIMESTAMP DEFAULT NOW(),
PRIMARY KEY (aggregate_id, seq)
);
优势是查询能力强。可以按 aggregate_id、event_type、recorded_at 任意组合过滤；可以 JOIN 业务表做关联分析；事务语义清晰，INSERT 事件和更新聚合状态可以在同一个数据库事务里完成。
劣势是吞吐扩展有限。单库写入受单机连接数和磁盘 IOPS 限制，水平分片后跨分片的顺序一致性变为分布式共识问题，复杂度陡增。
直接使用 Kafka
Kafka 天然顺序（分区级别）、天然不可变（日志只追加）、天然可遍历（seek 到任意位移），这些特性和事件存储的需求高度重叠。有些架构索性把 Kafka topic 当成事件存储，将事件直接发布到 Kafka，读取时从分区位移开始消费重建。
// 生产者视角：事件直接发到 Kafka
try (Producer<String, String> p = new KafkaProducer<>(props)) {
ProducerRecord<String, String> rec = new ProducerRecord<>(
"library.inventory.events",
bookId,
"{\"eventType\":\"Adjusted\",\"delta\":-2,...}");
p.send(rec).get();
}
Kafka 做事件存储的优势是吞吐天花板高：分区级别并行写入，底层顺序磁盘写入性能远超随机写入。劣势在查询能力：Kafka 只能按位移顺序遍历，不支持按 event_type 随机查；消息保留期受 retention.bytes / retention.ms 控制，超期后事件被删除，重建只能读到最近的数据。
混合方案：Kafka 做写入通道，SQLite 做本地归档
生产中最常见的折中是：
- 写入路径： 业务系统生成事件，发到 Kafka topic。
- 归档路径： 消费者从 Kafka 头读到尾，把事件按原序追加到本地 SQLite 或嵌入式数据库的日志表里。
- 重建路径： 从 SQLite 日志表按
seq顺序读取，不再需要经手 Kafka 网络层。
SQLite 作为嵌入式数据库的优势是没有服务端进程、文件就是数据库、单进程写入性能极高。它的局限是并发写入能力差，但事件存储天然是顺序单写，这个限制正好不冲突。
Kafka compacted topic 的局限
Kafka 提供了一种压缩策略 cleanup.policy=compact，它的设计目标是为 key-value 场景保留每个 key 的最新值。对于事件溯源，这是一个容易踩的坑：
- Compaction 只保留每个 key 的最新 message。 事件溯源要求保留所有历史事件，而
compacttopic 会在后台清理相同 key 的旧消息。如果一个商品有 1000 次库存调整，compact后只剩下最后一条，前面 999 条事件彻底消失。 - Compaction 是异步的。 即使 retention 还没触发，compact cleaner 也在后台按
min.cleanable.dirty.ratio运行，清理时间不可控。 - Compaction 不保证快照语义。 压缩后的消息可能不是原子快照，中间状态被截断，重建时序列不连续。
结论：Kafka compact topic 不能当事件存储用。 事件存储必须使用 cleanup.policy=delete 的普通日志 topic，并且把 retention.ms 设得足够长——或者干脆在 Kafka 端不设上限，由下游 SQLite 归档承担冷存储职责。
实测：消费者归档到 SQLite + 重放到任意版本
以下代码在 Docker 容器内执行，Kafka 运行在宿主机的 Docker 容器里，Java 代码通过共享网络访问 Kafka。生产者写入 20 条事件后，消费者把全部事件追加到 SQLite 本地表，再通过 version <= N 条件重放到任意版本。
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.sql.*;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
public class EventStoreDemo {
private static final String BOOTSTRAP = "127.0.0.1:9092";
private static final String TOPIC = "mq-arch-09-es-test";
public static void main(String[] args) throws Exception {
// 生成 20 条事件到 Kafka
int N = 20;
writeEvents(N);
System.out.println("Kafka 生产者写入 " + N
git
+ " 条事件到 topic: " + TOPIC);
// 消费归档到 SQLite
consumeAndStore(0L, -1L, "arch01");
// 重放到版本 5 和 10
replayFromVersion(5);
replayFromVersion(10);
// 一致性验证
Connection conn = getSqliteConn();
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM event_store");
rs.next();
System.out.println("\nSQLite 总事件数: " + rs.getInt(1));
ResultSet rs2 = stmt.executeQuery(
"SELECT SUM(CAST(json_extract(payload, '$.delta') AS INT)) " +
"FROM event_store WHERE event_type = 'InventoryAdjusted'");
rs2.next();
System.out.println("全量 delta SUM: " + rs2.getLong(1));
ResultSet rs3 = stmt.executeQuery(
"SELECT SUM(CAST(json_extract(payload, '$.delta') AS INT)) " +
"FROM event_store WHERE event_type = 'InventoryAdjusted' AND version <= 10");
rs3.next();
System.out.println("版本 <=10 的 delta SUM: " + rs3.getLong(1));
conn.close();
}
static void writeEvents(int n) throws Exception {
Properties pp = new Properties();
pp.put("bootstrap.servers", BOOTSTRAP);
pp.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
pp.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
Random rng = new Random(42);
String[] reasons = {"restock", "borrow", "return", "damage", "sale"};
String[] books = {"B-001", "B-002", "B-003", "B-004", "B-005"};
try (Producer<String, String> p = new KafkaProducer<>(pp)) {
for (int i = 1; i <= n; i++) {
String book = books[rng.nextInt(books.length)];
int delta = rng.nextBoolean()
? rng.nextInt(5) + 1 : -(rng.nextInt(3) + 1);
String payload = String.format(
"{\"version\":%d,\"bookId\":\"%s\",\"delta\":%d,\"reason\":\"%s\",\"occurredAt\":\"%s\"}",
i, book, delta, reasons[rng.nextInt(reasons.length)],
Instant.now().toString());
ProducerRecord<String, String> rec =
new ProducerRecord<>(TOPIC, book, payload);
RecordMetadata md = p.send(rec).get();
if (i <= 3 || i >= n - 2) {
System.out.printf(" sent[%d] offset=%d book=%s delta=%d%n",
i, md.offset(), book, delta);
}
}
}
}
static Connection getSqliteConn() throws SQLException {
return DriverManager.getConnection(
"jdbc:sqlite:/tmp/mq-arch-09-event-store.db");
}
static void consumeAndStore(long fromOffset, long toOffset, String tag)
throws Exception {
Properties cp = new Properties();
cp.put("bootstrap.servers", BOOTSTRAP);
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");
Connection conn = getSqliteConn();
Statement stmt = conn.createStatement();
stmt.execute("CREATE TABLE IF NOT EXISTS event_store (" +
"kafka_offset BIGINT PRIMARY KEY," +
"kafka_partition INT NOT NULL," +
"book_id TEXT NOT NULL," +
"event_type TEXT," +
"version INT," +
"payload TEXT NOT NULL," +
"stored_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)");
conn.close();
try (KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(cp)) {
TopicPartition tp = new TopicPartition(TOPIC, 0);
consumer.assign(Collections.singletonList(tp));
consumer.seek(tp, fromOffset);
int count = 0;
long endOffset = toOffset < 0
? getEndOffset(consumer, tp) : toOffset;
Connection wc = getSqliteConn();
wc.setAutoCommit(false);
PreparedStatement ps = wc.prepareStatement(
"INSERT OR IGNORE INTO event_store (" +
"kafka_offset,kafka_partition,book_id,event_type," +
"version,payload) VALUES (?,?,?,?,?,?)");
while (count < endOffset - fromOffset) {
ConsumerRecords<String, String> recs =
consumer.poll(Duration.ofMillis(3000));
if (recs.isEmpty()) break;
for (ConsumerRecord<String, String> rec : recs) {
if (toOffset >= 0 && rec.offset() >= toOffset) break;
String payload = rec.value();
int version = parseVersion(payload);
ps.setLong(1, rec.offset());
ps.setInt(2, rec.partition());
ps.setString(3, rec.key());
ps.setString(4, "InventoryAdjusted");
ps.setInt(5, version);
ps.setString(6, payload);
ps.addBatch();
count++;
}
ps.executeBatch();
wc.commit();
if (toOffset < 0 && count >= endOffset - fromOffset) break;
}
wc.close();
System.out.println("[" + tag + "] 消费并存储 " + count + " 条事件");
}
}
static long getEndOffset(KafkaConsumer<String, String> consumer,
TopicPartition tp) {
Map<TopicPartition, Long> endOffsets =
consumer.endOffsets(Collections.singletonList(tp));
return endOffsets.get(tp);
}
static int parseVersion(String json) {
String key = "\"version\":";
int idx = json.indexOf(key);
if (idx < 0) return 0;
int start = idx + key.length();
int end = json.indexOf(",", start);
if (end < 0) end = json.indexOf("}", start);
return Integer.parseInt(json.substring(start, end).trim());
}
static void replayFromVersion(int version) throws Exception {
Connection conn = getSqliteConn();
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery(
"SELECT book_id, CAST(json_extract(payload, '$.delta') AS INT) AS d " +
"FROM event_store WHERE version <= " + version +
" ORDER BY kafka_offset");
Map<String, Integer> totals = new LinkedHashMap<>();
int count = 0;
while (rs.next()) {
String book = rs.getString(1);
int delta = rs.getInt(2);
totals.put(book, totals.getOrDefault(book, 0) + delta);
count++;
}
System.out.println("\n--- 重放到版本 " + version +
"（" + count + " 条事件）---");
for (Map.Entry<String, Integer> e : totals.entrySet()) {
System.out.printf(" %s: total=%d%n", e.getKey(), e.getValue());
}
conn.close();
}
}
编译运行（Temurin 25.0.4+1 JDK / eclipse-temurin:25-jdk-alpine Docker / Kafka 4.3.0 KRaft / SQLite JDBC 3.47.0.0）：
javac --release 21 -cp ".:kafka-clients-4.3.0.jar:slf4j-api-2.0.20.jar" \n EventStoreDemo.java
java -cp ".:kafka-clients-4.3.0.jar:slf4j-api-2.0.20.jar:" +
"slf4j-simple-2.0.20.jar:sqlite-jdbc-3.47.0.0.jar" \n EventStoreDemo
输出：
sent[1] offset=20 book=B-001 delta=4
sent[2] offset=21 book=B-005 delta=-1
sent[3] offset=22 book=B-001 delta=-3
sent[18] offset=37 book=B-005 delta=-1
sent[19] offset=38 book=B-004 delta=3
sent[20] offset=39 book=B-001 delta=-2
Kafka 生产者写入 20 条事件到 topic: mq-arch-09-es-test
[arch01] 消费并存储 20 条事件
--- 重放到版本 5（5 条事件）---
B-001: total=0
B-005: total=-3
B-002: total=1
--- 重放到版本 10（10 条事件）---
B-001: total=-6
B-005: total=-3
B-002: total=1
B-004: total=1
B-003: total=-4
SQLite 总事件数: 20
全量 delta SUM: -11
版本 <=10 的 delta SUM: -11
单次运行，随调度波动，不作基准。
offset=20起因为 topic 之前已有实验消息。
这个实验验证了四个关键行为：
- Kafka 事件可以被消费并原样写入 SQLite。
INSERT OR IGNORE保证即使程序重启重复消费，也不会写入重复记录。 - SQLite 支持
json_extract。不需要手动反序列化，查询层可以直接从 JSON payload 提取字段做聚合。 - 重放到任意版本只需 SQL 过滤条件。
version <= 5和version <= 10的结果完全不同，说明版本回溯是精确可控的。 - 数据一致性可验证。 全量
SUM(delta)与按版本过滤后的SUM(delta)结果可交叉比对，SQLite 本地归档与 Kafka 源数据一一对应。
事件存储的表结构演进
SQLite 表是本地归档，但 schema 不可能一成不变。当事件格式升级时，归档表是否需要跟着改？通常有两种策略：
- 原样归档。 归档表存的是 Kafka message 的原始 payload，类型统一为
TEXT或BLOB，schema 变化不影响归档表。查询时用json_extract，如果新字段存在就提取，不存在就返回 NULL。 - 结构化归档。 归档表根据当前已知 schema 拆字段存储，schema 变更时通过 ALTER TABLE 添加新列。优势是查询性能好，劣势是 schema 演进需要迁移。
本地归档通常选第 1 种，因为简单、无迁移、存储开销对单进程场景可接受。
事件丢失的兜底
归档消费者是 Kafka 的一个消费者，它也有位移。如果消费者挂掉且位移提交失败，重启后会从上次成功提交的位置重新消费。这意味着两件事：
- 同一事件可能被写入 SQLite 两次。 这是幂等消费的问题，也是
INSERT OR IGNORE存在的理由。 - 位移提交不能太频繁也不能太松。 太频繁影响吞吐；太松则重启后重复消费窗口大。实践中每 1000 条事件或每 5 秒提交一次，根据业务容忍度调整。
Kafka retention 与归档的配合
虽然 SQLite 本地归档可以无限增长，但它只在消费者所在的节点本地存在。如果消费者所在机器挂了，SQLite 文件跟着丢失。为了避免这种单点风险，有两种做法：
- SQLite 文件定期异地备份。 用 rsync 或对象存储同步数据文件。由于 SQLite 是单文件，备份和恢复都很方便。
- 多消费者独立归档。 启动两个归档消费者写入各自节点的 SQLite，数据冗余两份。任何一份丢失，另一份可以补全。
Kafka 端的 retention 策略则保持保守：不清理，或清理周期远长于本地归档的频率。理想情况下，Kafka 的事件保留窗口 > 本地归档同步周期，这样即使本地文件丢失，也可以从 Kafka 重新消费补齐。
小结
- 事件存储要求顺序性、不可变性、可遍历性和全局唯一标识，这四条约束把事件存储与普通消息队列区分开。
- 自建事件存储（Postgres/H2）查询能力强但吞吐扩展有限；Kafka 吞吐高但查询能力弱且受 retention 限制。
- 混合方案最常用：Kafka 做写入通道，SQLite 做本地归档，重建时从 SQLite 读取不再经手网络。
- Kafka
compacttopic 不能当事件存储用，它会删除同 key 的历史消息，破坏事件溯源的完整性。 - 本机实测：20 条事件从 Kafka 消费归档到 SQLite，全量查询和版本过滤均正确，全量 delta SUM = -11，与重放结果一致。
阅读导航：上一节 9.2 快照与性能 ｜ 下一节 10.1 事件链路的追踪
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。