ZH
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish한국어日本語简体中文
首页 / 架构与设计 / Kafka

事件溯源和 Kafka：保存事件而非状态

Cris The Dev · 2026/10/12 · 11 分钟阅读 · 158

1. 概述

大多数应用程序只保存当前状态。当订单被取消时，orders 表中的行变为 CANCELLED。没有人知道之前订单有什么商品以及是否已付款。

事件溯源做法相反：我们只保存已发生的事件序列，而状态是从序列计算出来的结果。在本文中，你将使用 Kafka 作为 Order 聚合体的事件日志，通过重放事件来重建状态，通过 projection 构建读模型，并掌握实际限制。

2. 保存状态与保存事件

使用传统方式，订单是一条记录：

orders: id=A1 | status=CANCELLED | total=250000

使用事件溯源，同一订单是一个只能追加、不能修改、不能删除的事件序列：

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="客户改主意")

状态 CANCELLED, total=250000 通过按顺序“折叠”事件来计算。这种保存方式带来三个好处：

  • 天然审计：完整的历史就是源数据。
  • 新的读模型不需要迁移：想要添加报表，只需将日志重放到新的 projection。
  • 按时间调试：你可以重建任何时刻的状态。

Kafka 适合担当这个日志角色。Topic 是一个在每个分区内有序的仅追加序列。如果使用 orderId 作为 key，同一订单的所有事件都会落入同一分区并保持正确顺序。

3. 项目设置

3.1. 依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Spring Boot 管理 spring-kafka 的版本，所以你不需要声明 <version>。spring-boot-starter-json 包含 Jackson 和 jsr310 模块用于序列化 Instant。

3.2. 配置

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.value.default.type: com.nonla.orders.OrderEvent
        spring.json.trusted.packages: com.nonla.orders
        spring.json.use.type.headers: false

值得注意的几点：

  • acks: all 和 enable.idempotence: true 确保事件已被持久化记录且生产者重试不会创建重复记录。
  • 我们关闭 Spring 的类型头。事件类型将直接在 JSON 中通过 type 属性，所以用其他语言编写的 consumer 仍然可以读取。
  • auto-offset-reset: earliest 允许新的 consumer group 从日志开始处重新读取。这是重放的必要条件。

3.3. 持久化保存事件的 Topic

@Configuration
public class KafkaTopicConfig {

    @Bean
    NewTopic orderEvents() {
        return TopicBuilder.name("order-events")
                .partitions(3)
                .replicas(1)
                .config(TopicConfig.RETENTION_MS_CONFIG, "-1")
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
                .build();
    }
}

默认情况下 Kafka 在 7 天后删除数据。对于事件存储，这意味着丢失源数据。retention.ms=-1 禁用基于时间的删除。

注意： 不要为事件 topic 使用 cleanup.policy=compact。Compaction 只保留每个 key 的最后一条记录，即删除订单的整个历史。Compaction 仅适用于快照 topic。

4. 定义事件

事件是不可变数据，描述已发生的事情，以过去式命名。sealed interface 结合 Java 21 的 record 非常适合这个目的：

package com.nonla.orders;

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
        @JsonSubTypes.Type(value = OrderEvent.Created.class, name = "OrderCreated"),
        @JsonSubTypes.Type(value = OrderEvent.ItemAdded.class, name = "ItemAdded"),
        @JsonSubTypes.Type(value = OrderEvent.Paid.class, name = "OrderPaid"),
        @JsonSubTypes.Type(value = OrderEvent.Cancelled.class, name = "OrderCancelled")
})
public sealed interface OrderEvent {

    String orderId();
    Instant occurredAt();

    record Created(String orderId, String customerId, Instant occurredAt) implements OrderEvent {}
    record ItemAdded(String orderId, String sku, BigDecimal price, Instant occurredAt) implements OrderEvent {}
    record Paid(String orderId, BigDecimal amount, Instant occurredAt) implements OrderEvent {}
    record Cancelled(String orderId, String reason, Instant occurredAt) implements OrderEvent {}
}

发送 ItemAdded 时，Kafka 上的 payload 将如下所示：

{"type":"ItemAdded","orderId":"A1","sku":"COFFEE","price":150000,"occurredAt":"2026-10-11T03:15:00Z"}

@JsonSubTypes 中的名称是长期约定。你可以更改 Java 类名，但不能更改字符串 "ItemAdded"，否则旧事件将无法重新读取。

5. Aggregate：从事件构建状态

Order 有两种方法：

  • Command（addItem、pay、cancel）：检查业务规则然后返回事件。Command 不会自己修改状态。
  • apply：接收事件并更新状态。这个方法不应该抛出业务异常，因为事件已经发生了。
public final class Order {

    public enum Status { NEW, PAID, CANCELLED }

    private String id;
    private String customerId;
    private Status status;
    private BigDecimal total = BigDecimal.ZERO;
    private final List<String> items = new ArrayList<>();
    private long version;

    public static Order replay(List<OrderEvent> history) {
        Order order = new Order();
        history.forEach(order::apply);
        return order;
    }

    public void apply(OrderEvent event) {
        switch (event) {
            case OrderEvent.Created e -> { id = e.orderId(); customerId = e.customerId(); status = Status.NEW; }
            case OrderEvent.ItemAdded e -> { items.add(e.sku()); total = total.add(e.price()); }
            case OrderEvent.Paid e -> status = Status.PAID;
            case OrderEvent.Cancelled e -> status = Status.CANCELLED;
        }
        version++;
    }

    public OrderEvent addItem(String sku, BigDecimal price) {
        requireStatus(Status.NEW, "只能向新订单添加商品");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "订单不在等待支付状态");
        if (total.signum() == 0) {
            throw new IllegalStateException("无法支付空订单");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("订单已被取消");
        }
        return new OrderEvent.Cancelled(id, reason, Instant.now());
    }

    private void requireStatus(Status expected, String message) {
        if (status != expected) throw new IllegalStateException(message);
    }

    public String id() { return id; }
    public Status status() { return status; }
    public BigDecimal total() { return total; }
    public List<String> items() { return List.copyOf(items); }
    public long version() { return version; }
}

因为 OrderEvent 是 sealed，apply 中的 switch 语句被编译器充分检查。当你添加新的事件类型忘记处理时，代码将无法编译。这是一个值得拥有的安全网。

5.1. 测试 Aggregate

Aggregate 不依赖 Kafka，所以测试非常快：

class OrderTest {

    private final Instant now = Instant.parse("2026-10-11T03:00:00Z");

    @Test
    void replayRebuildsStateFromHistory() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.ItemAdded("A1", "COFFEE", new BigDecimal("150000"), now),
                new OrderEvent.ItemAdded("A1", "MUG", new BigDecimal("100000"), now),
                new OrderEvent.Cancelled("A1", "客户改主意", now)));

        assertThat(order.status()).isEqualTo(Order.Status.CANCELLED);
        assertThat(order.total()).isEqualByComparingTo("250000");
        assertThat(order.items()).containsExactly("COFFEE", "MUG");
        assertThat(order.version()).isEqualTo(4);
    }

    @Test
    void cannotPayCancelledOrder() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.Cancelled("A1", "缺货", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("等待支付");
    }
}

第一个测试显示四个事件折叠成一致的状态，版本等于已应用的事件数。第二个测试显示业务规则基于从历史重建的状态进行检查。

6. 将事件写入 Kafka

6.1. 事件存储

要处理 command，你需要一份订单的历史。Kafka 不支持查询“给我所有 key 为 A1 的记录”，所以我们在内存中保存一个副本。这个副本通过在应用程序启动时重新读取整个 topic 来加载：

@Component
public class OrderEventStore {

    private static final String TOPIC = "order-events";

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final Map<String, List<OrderEvent>> streams = new ConcurrentHashMap<>();

    public OrderEventStore(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void append(OrderEvent event) {
        try {
            kafkaTemplate.send(TOPIC, event.orderId(), event).get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new IllegalStateException("无法写入事件 " + event, e);
        }
    }

    public List<OrderEvent> load(String orderId) {
        return List.copyOf(streams.getOrDefault(orderId, List.of()));
    }

    @KafkaListener(topics = TOPIC, groupId = "order-store-#{T(java.util.UUID).randomUUID()}")
    void on(OrderEvent event) {
        streams.computeIfAbsent(event.orderId(), id -> new CopyOnWriteArrayList<>()).add(event);
    }
}

各部分解释如下：

  • send(TOPIC, event.orderId(), event) 使用 orderId 作为 key。这样一个订单的所有事件都在同一分区且保持正确顺序。
  • .get(...) 等待 broker 确认。Command 仅在事件已进入日志时才被认为成功。
  • Listener 使用随机的 groupId。每次启动都是一个新的 group，结合 earliest，整个历史被重新读取到 map 中。

6.2. Command 服务

@Service
public class OrderCommandService {

    private final OrderEventStore store;

    public OrderCommandService(OrderEventStore store) {
        this.store = store;
    }

    public String createOrder(String customerId) {
        String orderId = UUID.randomUUID().toString();
        store.append(new OrderEvent.Created(orderId, customerId, Instant.now()));
        return orderId;
    }

    public void addItem(String orderId, String sku, BigDecimal price) {
        Order order = loadExisting(orderId);
        store.append(order.addItem(sku, price));
    }

    public void pay(String orderId) {
        store.append(loadExisting(orderId).pay());
    }

    private Order loadExisting(String orderId) {
        List<OrderEvent> history = store.load(orderId);
        if (history.isEmpty()) {
            throw new NoSuchElementException("未找到订单 " + orderId);
        }
        return Order.replay(history);
    }
}

处理流程总是相同的：加载历史 → 重放 → 调用 command → 追加事件。没有任何 UPDATE 语句。

注意： 内存中的 map 通过 listener 异步更新。如果你在 createOrder 之后立即调用 addItem，map 可能还没来得及接收 Created。此外，Kafka 没有“仅当当前版本为 4 时才写入”这样的乐观并发控制。两个实例处理同一订单可能会写入两个相互矛盾的事件。在生产环境中，你应该将同一 orderId 的所有 command 路由到一个唯一的 consumer（command topic 按 orderId 分键），或使用 Kafka Streams 配合 state store。另一个选项是使用专用事件存储或数据库作为主要写入源，然后发布到 Kafka。

7. Projection：构建读模型

事件日志非常适合写入，但不方便查询，如列出客户 C7 已支付的订单。Projection 解决了这个问题。它是一个读取事件并维护为读取优化的模型的 consumer。这正是 CQRS 中的 Q 部分。

public record OrderSummary(String orderId, String customerId, String status, BigDecimal total) {}

@Component
public class OrderSummaryProjection {

    private final Map<String, OrderSummary> summaries = new ConcurrentHashMap<>();

    @KafkaListener(topics = "order-events", groupId = "order-summary-projection")
    void on(OrderEvent event) {
        summaries.compute(event.orderId(), (id, current) -> switch (event) {
            case OrderEvent.Created e -> new OrderSummary(id, e.customerId(), "NEW", BigDecimal.ZERO);
            case OrderEvent.ItemAdded e -> new OrderSummary(id, current.customerId(), current.status(),
                    current.total().add(e.price()));
            case OrderEvent.Paid e -> new OrderSummary(id, current.customerId(), "PAID", current.total());
            case OrderEvent.Cancelled e -> new OrderSummary(id, current.customerId(), "CANCELLED", current.total());
        });
    }

    public List<OrderSummary> paidOrdersOf(String customerId) {
        return summaries.values().stream()
                .filter(s -> s.customerId().equals(customerId) && s.status().equals("PAID"))
                .toList();
    }
}

这个例子为了简洁使用 ConcurrentHashMap。实际上，你会写入 PostgreSQL、Elasticsearch 或 Redis。重点是 projection 有固定的 groupId，所以 Kafka 记住偏移量，每次重启后只处理新事件。

Projection 必须是幂等的。Kafka 保证 at-least-once，所以一个事件可能在 rebalance 后被重新传递。对于数据库，你可以保存每个分区最后处理的偏移量，并跳过那些较旧的记录。

7.1. 重放以重建读模型

假设 projection 有 bug 导致 total 计算错误。你修复代码、删除读模型数据、停止应用程序，然后将 group 的偏移量重置到开头：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-summary-projection \
  --topic order-events \
  --reset-offsets --to-earliest --execute
GROUP                     TOPIC         PARTITION  NEW-OFFSET
order-summary-projection  order-events  0          0
order-summary-projection  order-events  1          0
order-summary-projection  order-events  2          0

重启时，projection 读取整个历史并创建正确的读模型。想要添加全新的报表，只需编写具有新 groupId 的新 listener。该报表会自动拥有从第一天开始的所有数据。这是保存事件而非状态的最大好处。

8. 常见错误和何时不应使用

改变事件 schema。 旧事件永远存在。只应该添加具有默认值的新字段，并且不要删除或改变旧字段的含义。需要大幅改变时，创建新的事件类型（ItemAddedV2）或编写 upcaster 在读取时转换旧事件。对于大系统，应该使用 Schema Registry 配合 Avro 或 Protobuf。

在事件中塞入状态。 OrderUpdated { 整个订单 } 这样的事件只是穿着 event sourcing 外套的 CRUD。事件必须描述业务意图，如 ItemAdded 或 OrderCancelled。

重放过于缓慢。 具有数千个事件的 aggregate 会加载缓慢。解决方案是快照：定期将状态和版本保存到 compacted topic，加载时读取快照，然后只重放之后的事件。

在 apply 中调用副作用。 如果 apply 发送邮件，每次重放都会重新发送邮件。apply 只应该改变内存中的状态。

个人数据和被遗忘权。 不可变日志与用户数据删除要求相冲突。常用的技术是 crypto-shredding：用每个客户的私钥加密个人数据，需要删除时就销毁密钥。

以下情况下不应该使用 event sourcing：

  • 域主要是简单的 CRUD 且没人需要历史。
  • 团队还不熟悉读写端之间的最终一致性。
  • 你需要在写入后立即在多个 aggregate 上进行强一致查询。

注意： Kafka 是一个优秀的事件日志，但不是完整的事件存储。它缺少两样东西：按每个 aggregate 快速读取和按版本有条件地写入。许多系统使用数据库或专用事件存储作为真实来源，然后使用 Kafka（通常通过 outbox pattern）来分发事件到其他 projection 和服务。

9. 结论

在本文中，你将订单保存为 Kafka 上的不可变事件序列，而不是被覆盖的单一记录。你使用 sealed interface 和 record 对事件进行建模，用 Order.replay 重建状态，在追加前检查业务规则。你还通过 projection 构建了读模型，通过重置偏移量来重建它。

有三个要点要记住：

  • 使用 orderId 作为 key 来保持每个订单事件的顺序。
  • 禁用基于时间的保留，并且不为事件 topic 使用 compaction。
  • 清楚了解 Kafka 的并发限制，才能把它视为唯一真实来源。

正确应用时，事件溯源为你提供完整历史和创建新读模型而无需迁移的能力。

kafkaevent-sourcingspring-bootcqrsspring-kafka
作者的应用HoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同主题文章