Cấu trúc dữ liệu và Giải thuật Bài 96: Transactional Outbox Pattern và Debezium (CDC) - Bảo chứng toàn vẹn dữ liệu giữa Database và Kafka.

Ở Bài 95, hệ thống E-commerce của Hasaki đã áp dụng Saga Pattern để điều phối các Microservices. Nguyên lý cốt lõi của Saga là: Khi Service A làm xong việc, nó phải bắn một Sự kiện (Event/Message) vào Kafka để Service B biết đường làm tiếp.

Nhưng ở đây ẩn chứa một cạm bẫy vật lý chết người mang tên Dual-Write (Ghi kép): Làm sao để đảm bảo việc Lưu vào MySQL và Bắn message vào Kafka thành công cùng một lúc?

1. Nỗi đau của bài toán Dual-Write (Ghi kép)

Giả sử đoạn code Golang của bạn xử lý việc tạo Đơn hàng như sau:

Lưu Đơn hàng vào MySQL.
Bắn Message "OrderCreated" vào Kafka để gọi Payment Service.

Nếu thao tác 1 thành công, nhưng khoảnh khắc sang thao tác 2 mạng bị đứt, hoặc Kafka đang bị sập (Broker down). Kết quả: Đơn hàng đã nằm trong Database, nhưng Message không bao giờ đến được Kafka. Khách hàng ngồi chờ mòn mỏi mà không thấy popup thanh toán hiện ra, đơn hàng kẹt mãi mãi ở trạng thái "Pending".

Ngược lại, nếu bạn bắn Message vào Kafka trước rồi mới lưu Database. Message bay đi thành công, Payment Service trừ tiền khách hàng, nhưng khi vòng lại lưu MySQL thì database báo lỗi Deadlock. Lúc này tiền đã trừ, nhưng đơn hàng không tồn tại!

Để giải quyết bài toán "Sống cùng sống, chết cùng chết" giữa hai hệ thống hoàn toàn khác biệt (MySQL và Kafka), chúng ta sử dụng kiến trúc Transactional Outbox Pattern.

2. Cứu cánh: Transactional Outbox Pattern

Ý tưởng của Outbox Pattern rất đơn giản nhưng cực kỳ uy lực: Mượn tính toàn vẹn (ACID) của cơ sở dữ liệu quan hệ để bảo vệ Kafka.

Thay vì cố gắng bắn message trực tiếp ra Kafka từ trong Application, chúng ta tạo thêm một bảng mới trong MySQL gọi là bảng outbox_events. Bảng này đóng vai trò như một "Hộp thư đi". Khi có đơn hàng mới, hệ thống mở một Transaction duy nhất và thực hiện 2 lệnh INSERT đồng thời:

SQL

BEGIN;
INSERT INTO orders (id, total, status) VALUES (101, 500000, 'pending');
INSERT INTO outbox_events (aggregate_id, event_type, payload) VALUES (101, 'OrderCreated', '{"id":101, "total":500000}');
COMMIT;

Nhờ đặc tính Nguyên tử (Atomicity) của Database Transaction, nếu lệnh lưu đơn hàng thất bại, sự kiện cũng bị hủy. Nếu lưu thành công, cả đơn hàng và sự kiện chắc chắn 100% đã nằm trên ổ cứng của MySQL.

Lúc này, API của Go có thể phản hồi 200 OK cho khách hàng ngay lập tức. Phần còn lại của bài toán chỉ là: Làm sao để bốc dữ liệu từ bảng outbox_events đẩy sang Kafka một cách trơn tru?

3. Quái vật hút dữ liệu: Change Data Capture (CDC) với Debezium

Cách cổ điển để đọc bảng outbox_events là viết một Cronjob (Polling) cứ 1 giây lại quét SELECT * FROM outbox_events WHERE status = 'NEW', đẩy ra Kafka rồi UPDATE status = 'SENT'. Cách này chạy được, nhưng tạo ra tải (Load) khủng khiếp lên Database và có độ trễ cao.

Công nghệ hiện đại giải quyết vấn đề này bằng CDC (Change Data Capture), và đại diện xuất sắc nhất là Debezium.

Debezium không chạy lệnh SELECT. Nó hoạt động như một gián điệp cấp thấp:

Trong MySQL, mọi thao tác INSERT, UPDATE, DELETE đều được hệ quản trị cơ sở dữ liệu âm thầm ghi lại vào một file log trên ổ cứng gọi là Binlog (Binary Log). Đây là file dùng để nhân bản dữ liệu từ Master sang Slave.
Debezium ngụy trang thành một Slave Database, kết nối với Master và liên tục "đọc trộm" (tail) file Binlog này.
Bất cứ khi nào có một dòng mới xuất hiện trong bảng outbox_events, Debezium lập tức bắt được luồng byte ở cấp độ hệ điều hành, biến nó thành Message và đẩy thẳng vào Kafka với độ trễ tính bằng mili-giây (Sub-millisecond).

4. Triển khai Outbox Transaction bằng Golang

Dưới đây là mô phỏng cách ứng dụng Backend của bạn thực hiện ghi dữ liệu an toàn vào Database, chuẩn bị sẵn sàng mồi nhử cho Debezium hút đi:

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"

	_ "github.com/go-sql-driver/mysql"
)

// Cấu trúc Payload của Sự kiện
type OrderEventPayload struct {
	OrderID int     `json:"order_id"`
	Total   float64 `json:"total"`
	Status  string  `json:"status"`
}

// Xử lý tạo đơn hàng với Outbox Pattern
func CreateOrderWithOutbox(db *sql.DB, orderID int, total float64) error {
	ctx := context.Background()

	// 1. Mở một Database Transaction duy nhất
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	// Đảm bảo Rollback nếu có lỗi xảy ra giữa chừng
	defer tx.Rollback()

	// 2. Thực thi Business Logic: Lưu vào bảng Orders
	_, err = tx.ExecContext(ctx, "INSERT INTO orders (id, total, status) VALUES (?, ?, ?)", orderID, total, "pending")
	if err != nil {
		return fmt.Errorf("lỗi khi lưu orders: %v", err)
	}

	// 3. Chuẩn bị Event Payload
	payload := OrderEventPayload{
		OrderID: orderID,
		Total:   total,
		Status:  "pending",
	}
	payloadBytes, _ := json.Marshal(payload)

	// 4. Lưu vào bảng Outbox trong CÙNG một Transaction
	// Chú ý: Không gọi API Kafka ở đây. Chỉ ghi vào DB.
	_, err = tx.ExecContext(ctx, 
		"INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload) VALUES (?, ?, ?, ?)",
		"Order", orderID, "OrderCreated", string(payloadBytes),
	)
	if err != nil {
		return fmt.Errorf("lỗi khi lưu outbox event: %v", err)
	}

	// 5. Commit chốt dữ liệu
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("lỗi khi commit transaction: %v", err)
	}

	fmt.Printf("Giao dịch Thành công! Đơn hàng %d và Event đã được lưu an toàn xuống đĩa.\n", orderID)
	return nil
}

func main() {
	// (Mô phỏng) Kết nối Database
	// db, err := sql.Open("mysql", "user:pass@tcp(localhost:3306)/hasaki_db")
	// if err != nil { log.Fatal(err) }
	// defer db.Close()
	
	// CreateOrderWithOutbox(db, 1005, 750000)
	fmt.Println("Mô phỏng Ghi Outbox thành công. Debezium đang chờ ở nền để đọc Binlog...")
}

5. Xử lý "At-least-once" và Tính Idempotent

Sự kết hợp giữa Outbox và Debezium đảm bảo tin nhắn không bao giờ bị mất. Tuy nhiên, trong các hệ thống phân tán, do yếu tố đường truyền mạng rớt gói (network retry), Kafka có thể nhận được một tin nhắn hai lần (At-least-once delivery).

Điều này dẫn đến hệ lụy: Payment Service có thể trừ tiền khách hàng 2 lần cho cùng một hóa đơn. Để giải quyết triệt để, Service nhận (Consumer) bắt buộc phải được thiết kế có tính Idempotent (Không thay đổi trạng thái nếu gọi nhiều lần).

Cách xử lý kinh điển:

Payment Service duy trì một bảng processed_events lưu ID của các tin nhắn đã xử lý.
Khi nhận một Message từ Kafka, nó dùng lệnh INSERT IGNORE INTO processed_events (event_id) (Sử dụng Unique Key).
Nếu INSERT thành công, nó mới gọi API ngân hàng trừ tiền. Nếu báo lỗi Duplicate (tin nhắn trùng), nó lập tức Drop bỏ tin nhắn và không làm gì cả.

Với mảnh ghép Outbox Pattern và Debezium, chúng ta đã xây dựng xong một cầu nối bất khả xâm phạm giữa Database tĩnh và Dòng chảy dữ liệu Kafka động.

cấu trúc dữ liệu

1. Nỗi đau của bài toán Dual-Write (Ghi kép)

2. Cứu cánh: Transactional Outbox Pattern

3. Quái vật hút dữ liệu: Change Data Capture (CDC) với Debezium

4. Triển khai Outbox Transaction bằng Golang

5. Xử lý "At-least-once" và Tính Idempotent

Table of contents