JA
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish한국어日本語简体中文
ホーム / アーキテクチャ・設計 / Kafka

イベントソーシング with Kafka：状態ではなくイベントを保存する

Cris The Dev · 2026/10/12 · 11 分で読めます · 158

1. 概要

ほとんどのアプリケーションは現在の状態のみを保存します。注文がキャンセルされるとき、ordersテーブルの行はCANCELLEDに変わります。その前に注文に何が入っていたのか、支払いが済んだのかは、誰も知ることができなくなります。

イベントソーシングはその逆です。我々は発生した一連のイベントのみを保存し、状態はそのシーケンスから計算される結果です。この記事では、Spring Boot 3.xを使用して、KafkaをOrderアグリゲートのイベントログとして使用します。イベントをリプレイして状態を再構築し、プロジェクションで読み取りモデルを構築し、このアプローチの実際の制限を理解するでしょう。

2. 状態の保存 対 イベントの保存

従来のアプローチでは、注文は1つのレコードです：

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は、イベントを順序に従って「折りたたむ」(fold)ことで計算されます。このように保存することで、3つの利点がもたらされます：

  • 自然な監査証跡：完全な履歴が生のデータそのものです。
  • 新しい読み取りモデルのマイグレーション不要：新しいレポートを追加するには、ログを新しいプロジェクションにリプレイするだけです。
  • 時間旅行デバッグ：任意の時点での状態を再構築できます。

Kafkaはこのログの役割に適しています。トピックは各パーティション内で順序付けられた追記のみのシーケンスです。orderIdをキーとして使用すれば、同じ注文のすべてのイベントは同じパーティションに落ち、正しい順序を保ちます。

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 と Instantをシリアライズするためのjsr310モジュールを提供します。

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 のタイプヘッダーを無効にしています。イベントタイプはtypeプロパティを通じて JSON に含まれるため、他の言語で書かれたコンシューマーでも読み取ることができます。
  • auto-offset-reset: earliestは、新しいコンシューマーグループがログの開始から読み取ることを許可します。これはリプレイの必要条件です。

3.3. イベントを永久に保持するトピック

@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は時間ベースの削除を無効にします。

注意： イベントトピックにcleanup.policy=compactを使用しないでください。圧縮は各キーの最後のレコードのみを保持します。つまり、注文の履歴全体を削除します。圧縮はスナップショットトピックにのみ適しています。

4. イベントの定義

イベントは不変のデータで、発生したことを説明し、過去形で名付けられます。Java 21 のsealed interfaceと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 上のペイロードは次のようになります：

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

@JsonSubTypes内の名前は長期的な契約です。Java クラス名は変更できますが、文字列"ItemAdded"は変更できません。そうでなければ、古いイベントは逆シリアル化できません。

5. アグリゲート：イベントから状態を構築する

Orderには 2 種類のメソッドがあります：

  • コマンド（addItem、pay、cancel）：ビジネスルールをチェックしてイベントを返す。コマンド自体は状態を変更しません。
  • 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. アグリゲートをテストする

アグリゲートは 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("支払い待ち状態");
    }
}

最初のテストは、4 つのイベントが 1 つの一貫した状態に折りたたまれ、versionがアプライされたイベント数に等しいことを示しています。2 番目のテストは、履歴から再構築された状態に基づいてビジネスルールがチェックされることを示しています。

6. イベントを Kafka に書き込む

6.1. イベントストア

コマンドを処理するには、1 つの注文の履歴が必要です。Kafka は「キーが A1 のすべてのレコードをくれ」のようなクエリをサポートしていないため、メモリにコピーを保持します。このコピーは、アプリケーションが起動するたびにトピック全体を読み取ることによって読み込まれます：

@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をキーとして使用します。これにより、1 つの注文のすべてのイベントが同じパーティションに落ちて、正しい順序を保ちます。
  • .get(...)はブローカーの確認を待ちます。コマンドはイベントがログに入ったときだけ成功と見なされます。
  • リスナーはランダムなgroupIdを使用します。起動するたびに新しいグループで、earliestと組み合わせて、履歴全体がマップに読み込まれます。

6.2. コマンドサービス

@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);
    }
}

処理フローは常に同じです：履歴を読み込む → リプレイする → コマンドを呼ぶ → イベントを追加する。UPDATEステートメントはありません。

注意： メモリ内マップはリスナーを通じて非同期に更新されます。createOrder直後にaddItemを呼ぶ場合、マップはCreatedを受け取るのに間に合わない可能性があります。また、Kafka には「現在のバージョンが 4 の場合のみ書き込む」のようなオプティミスティック同時実行制御がありません。同じ注文を処理する 2 つのインスタンスは矛盾したイベントを書き込むことができます。本番環境では、1 つのorderIdのすべてのコマンドを 1 つのコンシューマーにルーティングする（キーベースのパーティショニングを使用して）か、Kafka Streams を状態ストアで使用する必要があります。別のオプションは、データベースまたは専用イベントストアを書き込みソースとして使用し、Kafka に発行することです。

7. プロジェクション：読み取りモデルを構築する

イベントログは書き込みには最適ですが、「顧客 C7 の支払済み注文をリストアップ」のようなクエリには不便です。プロジェクションがこれを解決します。これは、イベントを読み取り、読み取りに最適化されたモデルを維持するコンシューマーです。これは 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 に書き込みます。重要なのは、プロジェクションが固定の groupIdを持つため、Kafka はオフセットを覚えており、再起動後は新しいイベントのみを処理することです。

プロジェクションはべき等である必要があります。Kafka は at-least-once を保証するため、リバランス後にイベントが再配信される可能性があります。データベースでは、処理済みの最後のパーティションごとのオフセットを格納し、古いレコードをスキップできます。

7.1. 読み取りモデルを再構築するためのリプレイ

プロジェクションにバグがあり、totalの計算が間違っていたと仮定します。コードを修正し、読み取りモデルデータを削除し、アプリケーションを停止して、グループのオフセットを開始にリセットします：

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

再起動するとき、プロジェクションは履歴全体を読み取り、正しい読み取りモデルを作成します。完全に新しいレポートを追加するには、新しいgroupIdを持つ新しいリスナーを記述するだけです。そのレポートは自動的に初日からのすべてのデータを持っています。これが、状態ではなくイベントを保存することの最大の利点です。

8. よくあるエラーと非推奨の場合

イベントスキーマの変更。 イベントは永遠に存在します。デフォルト値を持つ新しいフィールドを追加するだけで、古いフィールドを削除したり意味を変更したりしないでください。大きな変更が必要な場合は、新しいイベントタイプ（ItemAddedV2）を作成するか、読み取るときに古いイベントを変換する「アップキャスター」を記述します。大規模なシステムでは、Avro または Protobuf で Schema Registry を使用する必要があります。

イベントに状態を入れる。 OrderUpdated { entire order }のようなイベントは、単なるイベントソーシング風の CRUD です。イベントはビジネス意図を説明する必要があり、ItemAddedまたはOrderCancelledのようなものです。

リプレイが長すぎる。 数千のイベントを持つアグリゲートは遅く読み込まれます。解決策はスナップショットです。定期的にversionと共に状態をコンパクト化されたトピックに保存し、読み込むときはスナップショットを読み取ってからそれ以降のイベントをリプレイします。

applyでサイドエフェクトを呼ぶ。 applyがメールを送信する場合、リプレイするたびにメールが再度送信されます。applyはメモリ内の状態を変更するだけです。

個人データと削除権。 不変ログはユーザーデータ削除の要件と矛盾します。よく使われる手法は暗号シュレッディングです。個人データを各顧客に固有のキーで暗号化し、削除する必要があるときはキーを破棄します。

次の場合はイベントソーシングを使用しないでください：

  • ドメインはほぼシンプルな CRUD で、誰も履歴を必要としません。
  • チームは読み取り側と読み取りモデル間の結果整合性に不慣れです。
  • 複数のアグリゲートにわたって書き込み後すぐに一貫したクエリが必要です。

注意： Kafka は素晴らしいイベントログですが、完全なイベントストアではありません。2 つのことが不足しています。アグリゲートで高速読み取りとバージョンごとの条件付き書き込みです。多くのシステムは、データベースまたは専用イベントストアを真実のソースとして使用し、Kafka（通常 outbox パターンを介して）を使用してイベントをプロジェクションおよび他のサービスに配布します。

9. まとめ

この記事では、単一の上書きされたレコードではなく、Kafka 上の不変のイベントシーケンスとして注文を保存しました。sealed interfaceとrecordでイベントをモデル化し、Order.replayで状態を再構築し、追加前にビジネスルールをチェックしました。また、プロジェクションで読み取りモデルを構築し、オフセットをリセットして再構築しました。

3 つのことを覚えておいてください：

  • orderIdをキーとして使用して、各注文のイベントの順序を保持します。
  • 時間ベースの保持を無効にし、イベントトピックに圧縮を使用しないでください。
  • それを唯一の真実のソースとして扱う前に、Kafka の同時実行の制限を明確に理解してください。

正しく適用されると、イベントソーシングは完全な履歴を提供し、マイグレーションなしで新しい読み取りモデルを作成する機能を提供します。

kafkaevent-sourcingspring-bootcqrsspring-kafka
著者のアプリHoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同じトピックの記事

Kafka · 2026/10/12 · 8 分で読めます · 312

Change Data CaptureとDebeziumとKafka

Debeziumを使ってPostgreSQLのWALを読み込み、すべてのデータ変更をKafkaにプッシュし、Spring Bootで消費。データ書き込みコードの変更は不要です。