7.7K 39 114

Đã đăng vào khoảng 8 giờ trước 18 phút đọc

📨🧠 Change Data Capture & Event Sourcing với Kafka – Khi Dữ Liệu Trở Thành Câu Chuyện Liên Tục - Kafka 101 P11

Change Data Capture & Event Sourcing với Kafka – Khi Dữ Liệu Trở Thành Câu Chuyện Liên Tục

Trong các hệ thống microservices hiện đại, có một bài toán kinh điển luôn làm đau đầu các kỹ sư thiết kế hệ thống: Làm sao để đồng bộ hóa dữ liệu thời gian thực giữa cơ sở dữ liệu chính (như PostgreSQL, MySQL) với các hệ thống chuyên biệt khác (như Elasticsearch phục vụ tìm kiếm, Redis phục vụ caching, hay Data Warehouse phục vụ phân tích)?

Khi một bản ghi thay đổi, toàn bộ các hệ thống vệ tinh phải lập tức cập nhật theo. Nếu bạn chọn cách để ứng dụng chính tự gọi API cập nhật đến từng hệ thống đó, bạn đang tự đào hố chôn mình bằng những lỗi mất đồng bộ dữ liệu không thể kiểm soát.

Bài viết này sẽ mổ xẻ hai mô hình thiết kế tối thượng giúp duy trì tính đồng bộ dữ liệu thời gian thực mà không tạo gánh nặng khóa bảng (locking overhead) hay làm chậm ứng dụng chính: Change Data Capture (CDC) và Event Sourcing, với Apache Kafka đóng vai trò là trục xương sống kết nối mọi thứ.

Câu chuyện từ phòng vận hành: Thảm họa mất đồng bộ dữ liệu tại trang thương mại điện tử

Hãy tưởng tượng bạn đang vận hành một trang web thương mại điện tử quy mô lớn. Hệ thống sử dụng PostgreSQL làm cơ sở dữ liệu transactional chính để lưu thông tin sản phẩm và đơn hàng. Để phục vụ tính năng tìm kiếm sản phẩm nhanh và gợi ý thông minh, hệ thống sử dụng một cụm Elasticsearch Cluster.

Để giữ Elasticsearch đồng bộ với PostgreSQL, nhóm phát triển ban đầu đã triển khai giải pháp "kép" ngay trong code ứng dụng:

@Transactional
public void updateProduct(Product product) {
    // Bước 1: Lưu thông tin sản phẩm vào PostgreSQL
    productRepository.save(product); 
    
    // Bước 2: Gọi API cập nhật sang Elasticsearch
    elasticsearchClient.updateIndex(product); 
}

Mọi thứ chạy rất mượt mà trong môi trường staging. Nhưng khi đưa lên production với lượng traffic tăng vọt, thảm họa bắt đầu xuất hiện.

Vào một tối thứ Sáu, mạng kết nối giữa ứng dụng và cụm Elasticsearch bị chập chờn (network timeout) hoặc cụm Elasticsearch bị quá tải và phản hồi rất chậm.

Ứng dụng thực hiện bước 1: Ghi thành công thông tin sản phẩm mới vào PostgreSQL.
Đến bước 2: Ứng dụng gọi Elasticsearch API nhưng bị timeout. Ứng dụng crash hoặc quăng ngoại lệ (exception).

Vì bước 2 nằm ngoài tầm kiểm soát của database transaction, cơ sở dữ liệu PostgreSQL đã lưu thông tin sản phẩm mới (ví dụ: giá sản phẩm giảm từ $100 xuống $50). Tuy nhiên, chỉ mục trên Elasticsearch vẫn giữ giá cũ là $100.

Kết quả là: Khách hàng tìm kiếm sản phẩm trên trang web thấy giá hiển thị là $100 (đọc từ Elasticsearch), nhưng khi bấm vào chi tiết sản phẩm hoặc thanh toán thì giá lại là $50 (đọc từ PostgreSQL). Sự lệch pha này lập tức tạo ra hàng loạt khiếu nại, phá hỏng trải nghiệm người dùng và gây thất thoát doanh thu nghiêm trọng.

[PostgreSQL (Đã cập nhật: $50)] <--- MẤT ĐỒNG BỘ ---> [Elasticsearch (Vẫn giữ: $100)]

Khi đội ngũ kỹ sư cố gắng vá lỗi bằng cách chạy các script quét PostgreSQL để cập nhật lại Elasticsearch mỗi đêm, họ lại đối mặt với một vấn đề mới: script chạy quá nặng, gây khóa bảng (table locking) và làm chậm toàn bộ hệ thống giao dịch của khách hàng đang mua sắm ban đêm.

Niềm tin phổ biến 1: "CDC là viết Trigger trong Database để bắn sự kiện"

Tại sao niềm tin này nghe rất hợp lý?

Khi cần bắt các sự kiện thay đổi dữ liệu (insert, update, delete) trong database, giải pháp trực quan nhất xuất hiện trong đầu các lập trình viên SQL lâu năm là sử dụng Database Triggers.

Họ nghĩ rằng: "Chúng ta chỉ cần viết một Trigger trên bảng sản phẩm. Mỗi khi có dòng dữ liệu nào thay đổi, Trigger sẽ tự động chạy và gọi một hàm bên ngoài (như gọi HTTP REST API hoặc thực thi một thư viện gửi tin nhắn lên Kafka Broker)." Cực kỳ trực quan, không cần can thiệp vào code ứng dụng!

Tại sao nó lại phá hủy hệ thống production?

Database Trigger hoạt động bên trong cùng một context giao dịch (transaction) của câu lệnh ghi dữ liệu.

Tăng tải CPU/RAM cho DB: Cơ sở dữ liệu chính vốn dĩ đã phải chịu áp lực ghi dữ liệu lớn. Việc bắt nó phải gánh thêm logic chạy Trigger sẽ làm tiêu tốn thêm tài nguyên CPU và bộ nhớ của database server.
Nguy cơ sập dây chuyền (Cascading Failure): Nếu dịch vụ nhận sự kiện (API hoặc Kafka) bị chậm hoặc nghẽn mạng, Trigger sẽ bị treo để chờ phản hồi. Kết quả là câu lệnh ghi dữ liệu ban đầu (ví dụ: khách đặt hàng) cũng bị treo theo, dẫn đến nghẽn kết nối database (connection pool exhaustion) và kéo sập toàn bộ hệ thống.
Gây khóa bảng: Trigger chạy đồng bộ sẽ giữ các khóa hàng/khóa bảng (locks) lâu hơn, làm tăng khả năng xảy ra tranh chấp tài nguyên (deadlocks) giữa các giao dịch đồng thời.

Niềm tin phổ biến 2: "Event Sourcing chỉ đơn giản là ghi audit log vào một bảng phụ"

Nhiều lập trình viên tin rằng họ đang áp dụng kiến trúc Event Sourcing bằng cách tạo một bảng phụ trong database gọi là audit_logs hoặc product_history. Mỗi khi cập nhật bảng sản phẩm chính, họ chèn thêm một dòng ghi chú vào bảng lịch sử để ghi nhận ai đã sửa cái gì và lúc nào.

Tại sao đây là một sự hiểu lầm bản chất?

Cách tiếp cận trên thực chất chỉ là State Sourcing kết hợp với ghi chép lịch sử thụ động. Nguồn chân lý duy nhất (Single Source of Truth) của bạn vẫn là trạng thái hiện tại của thực thể được lưu trong bảng sản phẩm chính. Bảng lịch sử chỉ đóng vai trò thứ yếu để con người vào đọc khi cần đối soát.

Trong Event Sourcing thực thụ:

Không tồn tại bảng trạng thái hiện tại làm nguồn chân lý.
Nguồn chân lý duy nhất là một chuỗi các sự kiện nghiệp vụ (events) được ghi dưới dạng append-only (chỉ chèn thêm) vào một kho lưu trữ gọi là Event Store.
Trạng thái hiện tại của một đối tượng (ví dụ: Số dư ví, thông tin đơn hàng) không được lưu trữ trực tiếp. Khi ứng dụng cần biết trạng thái hiện tại, nó bắt buộc phải đọc lại toàn bộ chuỗi sự kiện từ quá khứ của đối tượng đó và thực hiện tái dựng (Replay/Fold) từ đầu.

State Sourcing (Truyền thống):
[Update Product Table] ──► [Product: Price = $50] (Chỉ giữ trạng thái cuối)

Event Sourcing:
[ProductCreated] ──► [PriceUpdated: $100] ──► [PriceUpdated: $50] (Lưu chuỗi sự kiện)

Góc nhìn mới: Đọc trực tiếp Transaction Log và Tách biệt luồng ghi

Để giải quyết bài toán đồng bộ dữ liệu một cách an toàn và hiệu năng cao, chúng ta cần thay đổi tư duy: Tuyệt đối không để ứng dụng chính chịu trách nhiệm đồng bộ, và tuyệt đối không dùng Trigger trong database.

Giải pháp chuẩn công nghiệp là sử dụng Change Data Capture (CDC) hoạt động ở tầng thấp hơn: Đọc trực tiếp Transaction Log của database.

1. Bản chất của Transaction Log

Mọi cơ sở dữ liệu quan hệ hiện đại (PostgreSQL có Write-Ahead Log - WAL, MySQL có Binary Log - binlog) đều sử dụng một cơ chế ghi chép cực nhanh gọi là Transaction Log. Trước khi bất kỳ thay đổi nào được áp dụng vào các bảng dữ liệu thực tế trên đĩa cứng, database sẽ ghi tuần tự thay đổi đó vào Transaction Log dưới dạng append-only. Quá trình này cực kỳ nhanh vì đĩa cứng ghi tuần tự không tốn chi phí tìm kiếm (seek time).

Công cụ CDC (như Debezium) sẽ chạy như một tiến trình độc lập, lắng nghe và đọc trực tiếp file Transaction Log này từ đĩa cứng (hoặc qua giao thức replication của database). Vì nó đọc bất đồng bộ và trực tiếp từ log, nó hoàn toàn không gây ảnh hưởng đến hiệu năng của database chính, không giữ bất kỳ khóa bảng nào, và hoạt động hoàn toàn độc lập với ứng dụng backend.

2. Kiến trúc Kafka Connect

Để biến dòng thay đổi từ Transaction Log thành dữ liệu hữu ích, chúng ta sử dụng Kafka Connect – một framework tích hợp dữ liệu mạnh mẽ đi kèm với Apache Kafka.

[App Client] ──Write──► [PostgreSQL (WAL)]
                             │
                             ▼ (Debezium Source Connector)
                      [Kafka Broker] (Topic: db.products)
                             │
                             ▼ (Elasticsearch Sink Connector)
                      [Elasticsearch Index]

Kafka Connect chia làm hai thành phần chính:

Source Connector: Lắng nghe nguồn dữ liệu (ở đây là PostgreSQL WAL thông qua Debezium), chuyển các thay đổi thành các message có cấu trúc và đẩy vào Kafka topic.
Sink Connector: Tiêu thụ dữ liệu từ Kafka topic và đẩy vào hệ thống đích (ở đây là Elasticsearch, Redis hoặc BigQuery).

Toàn bộ quá trình đồng bộ này diễn ra bất đồng bộ (asynchronous), tách biệt hoàn toàn (decoupled). Nếu Elasticsearch bị sập, Kafka Connect Sink Connector sẽ dừng lại và tự động phục hồi (retry) khi Elasticsearch hoạt động trở lại. Dữ liệu không bao giờ bị mất vì nó vẫn nằm an toàn trên Kafka topic.

Thiết kế Event Sourcing thời gian thực

Khi kết hợp CDC với tư duy Event Sourcing, chúng ta có thể xây dựng các hệ thống có khả năng phục hồi và tái dựng dữ liệu phi thường.

Trong kiến trúc Event Sourcing:

Event Store: Kafka đóng vai trò là một Event Store tuyệt vời nhờ tính chất lưu trữ tuần tự, không thể thay đổi (immutable log) và khả năng lưu giữ dữ liệu lâu dài (retention policy có thể cấu hình vô hạn).
Aggregate State (Trạng thái tích lũy): Ứng dụng đọc dòng event từ Kafka để dựng lên trạng thái hiện tại trong bộ nhớ (in-memory state) hoặc ghi xuống một Read-Model database (mô hình CQRS).

Giải pháp thực tế: Đồng bộ PostgreSQL sang Elasticsearch bằng Kafka Connect

Dưới đây là cách bạn cấu hình một hệ thống đồng bộ tự động từ PostgreSQL sang Elasticsearch mà không cần viết một dòng code ứng dụng nào.

1. Cấu hình Debezium PostgreSQL Source Connector

Chúng ta gửi cấu hình JSON này đến Kafka Connect REST API để kích hoạt việc đọc WAL từ PostgreSQL:

{
  "name": "postgres-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz_pass",
    "database.dbname": "ecommerce",
    "database.server.name": "dbserver1",
    "table.include.list": "public.products",
    "plugin.name": "pgoutput",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.products"
  }
}

Sau khi chạy connector này, mọi thao tác Insert, Update, Delete trên bảng products của PostgreSQL sẽ ngay lập tức sinh ra các message có cấu trúc JSON chi tiết trong Kafka topic dbserver1.public.products.

2. Cấu hình Elasticsearch Sink Connector

Tiếp theo, chúng ta cấu hình Sink Connector để tiêu thụ dữ liệu từ topic Kafka và đẩy sang Elasticsearch:

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "dbserver1.public.products",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": "false",
    "schema.ignore": "true",
    "behavior.on.null.values": "delete"
  }
}

Với thuộc tính "behavior.on.null.values": "delete", nếu bạn xóa một sản phẩm trong PostgreSQL, Debezium sẽ phát ra một message có value là null (tombstone record), và Elasticsearch Sink Connector sẽ tự động hiểu để xóa chỉ mục tương ứng trên Elasticsearch. Hệ thống tự động đồng bộ 100%!

Đánh đổi cấu trúc (Architectural Trade-offs)

Mặc dù CDC và Event Sourcing giải quyết triệt để bài toán đồng bộ dữ liệu, chúng ta vẫn phải đối mặt với các đánh đổi kiến trúc:

Tính nhất quán cuối cùng (Eventual Consistency): Vì quá trình đồng bộ là bất đồng bộ qua Kafka, sẽ có một khoảng thời gian trễ rất nhỏ (vài mili-giây đến vài giây) mà Elasticsearch chưa cập nhật kịp dữ liệu mới từ PostgreSQL. Bạn phải thiết kế ứng dụng để chấp nhận sự trễ này ở phía giao diện người dùng.
Độ phức tạp vận hành tăng cao: Bạn phải vận hành và giám sát thêm cụm Kafka Connect, quản lý các schema registry, cấu hình dọn dẹp log, và thiết lập cảnh báo khi Connector bị dừng hoạt động do lỗi cấu hình hoặc lỗi mạng.

Góc khuất Production: Thảm họa Replay hàng triệu sự kiện khi thiếu Snapshot

Trong kiến trúc Event Sourcing, khi một thực thể (ví dụ: tài khoản ngân hàng) tích lũy hàng triệu sự kiện qua nhiều năm hoạt động, một lỗi thiết kế nghiêm trọng rất dễ xảy ra: Thiếu cơ chế Snapshot (Ảnh chụp trạng thái).

Thảm họa xảy ra thế nào?

Để tìm ra số dư hiện tại của tài khoản khách hàng, hệ thống bắt đầu đọc từ sự kiện đầu tiên lúc mở tài khoản cách đây 5 năm (ví dụ: AccountCreated), sau đó duyệt qua 2 triệu sự kiện cộng tiền, trừ tiền tiếp theo để cộng dồn ra số dư hiện tại.

Khi ứng dụng bị crash hoặc cần khởi động lại:

Ứng dụng buộc phải khôi phục lại trạng thái của hàng nghìn tài khoản bằng cách đọc và replay lại hàng tỷ sự kiện từ Event Store (Kafka).
Quá trình Replay này tiêu tốn dung lượng I/O đĩa khổng lồ, làm CPU của worker tăng vọt lên 100%.
Thời gian khởi động ứng dụng kéo dài từ vài giây lên tới hàng tiếng đồng hồ. Trong suốt thời gian này, ứng dụng hoàn toàn mất khả năng phục vụ khách hàng.

Replay 2 triệu sự kiện ──► [Quá tải CPU/Disk] ──► [Ứng dụng khởi động mất 3 tiếng] ──► [Dịch vụ tê liệt]

Giải pháp khắc phục:

Áp dụng mẫu thiết kế Snapshotting: Định kỳ (ví dụ: sau mỗi 1,000 event hoặc vào cuối mỗi ngày), ứng dụng sẽ tính toán trạng thái hiện tại và lưu một bản "ảnh chụp" (Snapshot) vào một database phụ (như Redis hoặc PostgreSQL).
Khi cần tái dựng trạng thái, ứng dụng chỉ cần load bản Snapshot gần nhất (ví dụ: số dư vào cuối ngày hôm qua), sau đó chỉ replay các event phát sinh từ thời điểm Snapshot đó đến nay (vài chục event của ngày hôm nay). Thời gian khôi phục sẽ giảm từ hàng tiếng xuống còn vài mili-giây.

Tổng kết bài học (Key Takeaways)

Giải phóng ứng dụng chính: Hãy để ứng dụng của bạn tập trung vào việc thực thi giao dịch ghi dữ liệu chính. Sử dụng CDC thông qua đọc Transaction Log để thực hiện các tác vụ đồng bộ bất đồng bộ, giúp bảo vệ hiệu năng và độ ổn định của database.
Kafka Connect là vũ khí đắc lực: Đừng tự viết code kết nối và đồng bộ dữ liệu thủ công. Hãy tận dụng hệ sinh thái phong phú của Kafka Connect để tích hợp PostgreSQL, Elasticsearch, S3, BigQuery một cách tự động và bền vững.
Event Sourcing yêu cầu Snapshot: Hãy luôn ghi nhớ rằng việc lưu trữ chuỗi sự kiện vô hạn cần đi kèm với một chiến lược tạo ảnh chụp trạng thái (Snapshot) định kỳ để bảo vệ hệ thống khỏi thảm họa quá tải khi khôi phục trạng thái.

💡 Về TechCraft

TechCraft được xây dựng với mong muốn giúp Developer phát triển tư duy hệ thống thông qua những nội dung có chiều sâu về Backend Engineering, Distributed Systems và Production Architecture.

Tại đây, bạn sẽ không chỉ học cách một công nghệ hoạt động, mà còn hiểu vì sao các hệ thống lớn lại được thiết kế theo cách đó.

Nếu muốn tiếp tục đào sâu hơn, bạn có thể khám phá Dev Insider — nơi tập trung các series chuyên sâu dành cho Backend Developer.

🚀 Dev Insider https://www.patreon.com/techcraft_official/posts/vi-sao-dev-ra-161163881?collection=2220113

📘 Facebook https://www.facebook.com/techcraft.official

🎥 YouTube https://www.youtube.com/@techcraft.official

🎵 TikTok https://www.tiktok.com/@techcraft.official

Từ Developer biết code -> Engineer hiểu hệ thống.