VI
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish한국어日本語简体中文
Trang chủ / Kiến trúc & Thiết kế / Kafka

Event sourcing với Kafka: lưu sự kiện thay vì trạng thái

Cris The Dev · 12/10/2026 · 11 phút đọc · 158

1. Tổng quan

Phần lớn ứng dụng chỉ lưu trạng thái hiện tại. Khi một đơn hàng bị hủy, dòng trong bảng orders đổi thành CANCELLED. Không ai còn biết trước đó đơn đã có những món gì và đã thanh toán hay chưa.

Event sourcing làm ngược lại: ta chỉ lưu chuỗi sự kiện đã xảy ra, còn trạng thái là kết quả tính ra từ chuỗi đó. Trong bài này, bạn sẽ dùng Kafka làm event log cho aggregate Order với Spring Boot 3.x. Bạn sẽ dựng lại trạng thái bằng cách replay sự kiện, xây read model bằng projection và nắm các giới hạn thực tế của cách làm này.

2. Lưu trạng thái và lưu sự kiện

Với cách truyền thống, một đơn hàng là một bản ghi:

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

Với event sourcing, cùng đơn hàng đó là một chuỗi sự kiện chỉ được thêm vào, không sửa, không xóa:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="Khách đổi ý")

Trạng thái CANCELLED, total=250000 được tính bằng cách “gập” (fold) các sự kiện theo thứ tự. Cách lưu này mang lại ba lợi ích:

  • Audit tự nhiên: lịch sử đầy đủ chính là dữ liệu gốc.
  • Read model mới không cần migration: muốn thêm báo cáo, bạn replay log vào một projection mới.
  • Debug theo thời gian: bạn có thể dựng lại trạng thái tại bất kỳ thời điểm nào.

Kafka phù hợp với vai trò log này. Topic là một chuỗi append-only có thứ tự trong từng partition. Nếu dùng orderId làm key, mọi sự kiện của cùng một đơn sẽ rơi vào cùng partition và giữ đúng thứ tự.

3. Thiết lập dự án

3.1. Dependency

<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 quản lý version của spring-kafka, nên bạn không cần khai báo <version>. spring-boot-starter-json mang theo Jackson và module jsr310 để serialize Instant.

3.2. Cấu hình

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

Các điểm đáng chú ý:

  • acks: all và enable.idempotence: true đảm bảo sự kiện đã được ghi bền vững và producer retry không tạo bản ghi trùng.
  • Ta tắt type header của Spring. Loại sự kiện sẽ nằm ngay trong JSON qua thuộc tính type, nên consumer viết bằng ngôn ngữ khác vẫn đọc được.
  • auto-offset-reset: earliest cho phép consumer group mới đọc lại từ đầu log. Đây là điều kiện cần cho replay.

3.3. Topic giữ sự kiện vĩnh viễn

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

Mặc định Kafka xóa dữ liệu sau 7 ngày. Với event store, điều đó đồng nghĩa với mất dữ liệu gốc. retention.ms=-1 tắt việc xóa theo thời gian.

Lưu ý: Không dùng cleanup.policy=compact cho topic sự kiện. Compaction chỉ giữ bản ghi cuối cùng của mỗi key, tức là xóa toàn bộ lịch sử của đơn hàng. Compaction chỉ hợp với topic snapshot.

4. Định nghĩa sự kiện

Sự kiện là dữ liệu bất biến, mô tả việc đã xảy ra và được đặt tên ở thì quá khứ. sealed interface kết hợp record của Java 21 rất hợp cho việc này:

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 {}
}

Khi gửi một ItemAdded, payload trên Kafka sẽ có dạng:

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

Tên trong @JsonSubTypes là hợp đồng lâu dài. Bạn có thể đổi tên lớp Java, nhưng không được đổi chuỗi "ItemAdded", nếu không các sự kiện cũ sẽ không đọc lại được.

5. Aggregate: dựng trạng thái từ sự kiện

Order có hai loại phương thức:

  • Command (addItem, pay, cancel): kiểm tra quy tắc nghiệp vụ rồi trả về sự kiện. Command không tự sửa trạng thái.
  • apply: nhận một sự kiện và cập nhật trạng thái. Hàm này không được ném lỗi nghiệp vụ, vì sự kiện đã xảy ra rồi.
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, "Chỉ thêm món vào đơn mới");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "Đơn không ở trạng thái chờ thanh toán");
        if (total.signum() == 0) {
            throw new IllegalStateException("Không thể thanh toán đơn rỗng");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("Đơn đã bị hủy");
        }
        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; }
}

Vì OrderEvent là sealed, switch trong apply được compiler kiểm tra đầy đủ. Khi bạn thêm loại sự kiện mới mà quên xử lý, code sẽ không biên dịch. Đây là một lưới an toàn đáng giá.

5.1. Kiểm thử aggregate

Aggregate không phụ thuộc Kafka, nên test rất nhanh:

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", "Khách đổi ý", 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", "Hết hàng", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("chờ thanh toán");
    }
}

Test thứ nhất cho thấy bốn sự kiện được gập thành một trạng thái nhất quán, với version bằng số sự kiện đã áp dụng. Test thứ hai cho thấy quy tắc nghiệp vụ được kiểm tra dựa trên trạng thái dựng lại từ lịch sử.

6. Ghi sự kiện vào Kafka

6.1. Event store

Để xử lý command, bạn cần lịch sử của một đơn hàng. Kafka không hỗ trợ truy vấn “đưa tôi mọi bản ghi có key A1”, nên ta giữ một bản sao trong bộ nhớ. Bản sao này được nạp bằng cách đọc lại toàn bộ topic mỗi khi ứng dụng khởi động:

@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("Không ghi được sự kiện " + 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);
    }
}

Giải thích từng phần:

  • send(TOPIC, event.orderId(), event) dùng orderId làm key. Nhờ vậy mọi sự kiện của một đơn nằm cùng partition và giữ đúng thứ tự.
  • .get(...) chờ broker xác nhận. Command chỉ được coi là thành công khi sự kiện đã nằm trong log.
  • Listener dùng groupId ngẫu nhiên. Mỗi lần khởi động là một group mới, kết hợp với earliest nên toàn bộ lịch sử được đọc lại vào map.

6.2. Command service

@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("Không tìm thấy đơn " + orderId);
        }
        return Order.replay(history);
    }
}

Luồng xử lý luôn giống nhau: load lịch sử → replay → gọi command → append sự kiện. Không có câu lệnh UPDATE nào cả.

Lưu ý: Map trong bộ nhớ được cập nhật bất đồng bộ qua listener. Nếu bạn gọi addItem ngay sau createOrder, có thể map chưa kịp nhận Created. Ngoài ra, Kafka không có optimistic concurrency kiểu “chỉ ghi nếu version hiện tại là 4”. Hai instance cùng xử lý một đơn có thể ghi hai sự kiện mâu thuẫn. Trong production, bạn nên định tuyến mọi command của một orderId về một consumer duy nhất (command topic có key theo orderId), hoặc dùng Kafka Streams với state store. Một lựa chọn khác là dùng event store chuyên dụng hoặc database làm nguồn ghi chính, rồi publish sang Kafka.

7. Projection: xây read model

Event log rất tốt để ghi, nhưng không tiện để truy vấn kiểu “liệt kê đơn đã thanh toán của khách C7”. Projection giải quyết việc này. Nó là một consumer đọc sự kiện và duy trì một mô hình được tối ưu cho việc đọc. Đây chính là phần “Q” trong CQRS.

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

Ví dụ này dùng ConcurrentHashMap cho gọn. Thực tế, bạn sẽ ghi vào PostgreSQL, Elasticsearch hoặc Redis. Điểm quan trọng là projection có groupId cố định, nên Kafka nhớ offset và chỉ xử lý sự kiện mới sau mỗi lần restart.

Projection phải idempotent. Kafka đảm bảo at-least-once, nên một sự kiện có thể được giao lại sau khi rebalance. Với database, bạn có thể lưu kèm offset đã xử lý cuối cùng theo partition và bỏ qua những bản ghi cũ hơn.

7.1. Replay để dựng lại read model

Giả sử projection có bug khiến total bị tính sai. Bạn sửa code, xóa dữ liệu read model, dừng ứng dụng rồi reset offset của group về đầu:

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

Khi khởi động lại, projection đọc toàn bộ lịch sử và tạo ra read model đúng. Muốn thêm một báo cáo hoàn toàn mới, bạn chỉ cần viết listener mới với groupId mới. Báo cáo đó tự động có đủ dữ liệu từ ngày đầu tiên. Đây là lợi ích lớn nhất của việc lưu sự kiện thay vì trạng thái.

8. Lỗi hay gặp và khi nào không nên dùng

Thay đổi schema sự kiện. Sự kiện cũ tồn tại mãi mãi. Hãy chỉ thêm field mới có giá trị mặc định, và đừng xóa hay đổi nghĩa field cũ. Khi cần thay đổi lớn, tạo loại sự kiện mới (ItemAddedV2) hoặc viết “upcaster” chuyển đổi sự kiện cũ lúc đọc. Với hệ thống lớn, nên dùng Schema Registry kèm Avro hoặc Protobuf.

Nhét trạng thái vào sự kiện. Một sự kiện kiểu OrderUpdated { toàn bộ đơn hàng } chỉ là CRUD đội lốt event sourcing. Sự kiện phải mô tả ý định nghiệp vụ, như ItemAdded hay OrderCancelled.

Replay quá lâu. Một aggregate có hàng nghìn sự kiện sẽ load chậm. Giải pháp là snapshot: định kỳ lưu trạng thái kèm version vào một topic compacted, khi load thì đọc snapshot rồi chỉ replay các sự kiện sau đó.

Gọi side effect trong apply. Nếu apply gửi email, mỗi lần replay sẽ gửi lại email. apply chỉ được thay đổi trạng thái trong bộ nhớ.

Dữ liệu cá nhân và quyền được xóa. Log bất biến xung đột với yêu cầu xóa dữ liệu người dùng. Kỹ thuật thường dùng là crypto-shredding: mã hóa dữ liệu cá nhân bằng khóa riêng cho từng khách, khi cần xóa thì hủy khóa.

Bạn không nên dùng event sourcing khi:

  • Domain chủ yếu là CRUD đơn giản và không ai cần lịch sử.
  • Team chưa quen với eventual consistency giữa phía ghi và read model.
  • Bạn cần truy vấn nhất quán ngay sau khi ghi trên nhiều aggregate.

Lưu ý: Kafka là một event log tuyệt vời, nhưng không phải event store đầy đủ. Nó thiếu hai thứ: đọc nhanh theo từng aggregate và ghi có điều kiện theo version. Nhiều hệ thống dùng database hoặc event store chuyên dụng làm nguồn sự thật, rồi dùng Kafka (thường qua outbox pattern) để phân phối sự kiện cho các projection và service khác.

9. Kết luận

Trong bài này, bạn đã lưu đơn hàng dưới dạng chuỗi sự kiện bất biến trên Kafka thay vì một bản ghi bị ghi đè. Bạn đã mô hình hóa sự kiện bằng sealed interface và record, dựng lại trạng thái bằng Order.replay và kiểm tra quy tắc nghiệp vụ trước khi append. Bạn cũng đã xây read model bằng projection và dựng lại nó bằng cách reset offset.

Có ba điểm cần nhớ:

  • Dùng orderId làm key để giữ thứ tự sự kiện của từng đơn.
  • Tắt retention theo thời gian và không dùng compaction cho topic sự kiện.
  • Nhận thức rõ giới hạn về concurrency của Kafka trước khi coi nó là nguồn sự thật duy nhất.

Khi áp dụng đúng chỗ, event sourcing cho bạn lịch sử đầy đủ và khả năng tạo read model mới mà không cần migration.

kafkaevent-sourcingspring-bootcqrsspring-kafka
Ứng dụng của mìnhHoanTien.proMua Shopee, TikTok, Lazada được hoàn tiềniOS · Android · Web

Bài cùng chủ đề

Kafka · 12/10/2026 · 8 phút đọc · 312

Change Data Capture với Debezium và Kafka

Dùng Debezium đọc WAL của PostgreSQL, đẩy mọi thay đổi dữ liệu vào Kafka và tiêu thụ chúng bằng Spring Boot, không cần sửa code ghi dữ liệu.