Cạm bẫy Dual-Write trong kiến trúc Microservices phân tán
Hãy hình dung một luồng thanh toán điển hình: API của bạn cập nhật đơn hàng trong PostgreSQL và phát ra một message tới Kafka hoặc RabbitMQ. Mọi thứ nghe có vẻ rất đơn giản. Các service giao hàng và xuất hóa đơn ở phía hạ nguồn (downstream) dựa vào event đó để xử lý đơn hàng.
Ban đầu, hầu hết các kỹ sư thường viết mã tương tự như thế này:
// Antipattern: Ghi kép (Dual-Write) không an toàn
_, err := db.ExecContext(ctx, "UPDATE orders SET status = 'PAID' WHERE id = $1", orderID)
if err != nil {
return err
}
// Điều gì xảy ra nếu mạng bị ngắt hoặc pod bị crash ngay tại đây?
err = messageBroker.Publish(ctx, "order.paid", eventPayload)
if err != nil {
// Đơn hàng đã được đánh dấu PAID trong cơ sở dữ liệu, nhưng các downstream service hoàn toàn không biết gì về nó.
return err
}
Chào mừng bạn đến với cạm bẫy dual-write. Không có distributed transaction nào liên kết cơ sở dữ liệu quan hệ với message broker của bạn. Nếu ứng dụng bị crash trong khoảng thời gian giữa lúc commit cơ sở dữ liệu và gửi message đến broker, event sẽ biến mất vĩnh viễn. Khách hàng đã thanh toán tiền, nhưng kho hàng sẽ chẳng bao giờ đóng gói sản phẩm.
Đổi ngược thứ tự thực thi cũng không khá hơn. Việc publish event trước rồi commit sau tiềm ẩn nguy cơ xuất hiện các “event ma” (ghost event): nếu database transaction bị rollback, các downstream service sẽ xử lý một đơn hàng thực chất chưa hề được thanh toán.
Về mặt lý thuyết, Two-Phase Commit (2PC) có thể điều phối các transaction phân tán. Tuy nhiên trên thực tế, 2PC gây ra độ trễ tăng đột biến rất nghiêm trọng từ 50–100ms, tạo ra các điểm lỗi đơn lẻ (single points of failure), và không được hỗ trợ sẵn trong các broker hiện đại như Kafka. Chúng ta cần một giải pháp gửi event đáng tin cậy mà vẫn duy trì được throughput cao.
Cách thức hoạt động của Transactional Outbox Pattern
Transactional Outbox Pattern loại bỏ hoàn toàn nhu cầu về transaction phân tán. Thay vào đó, nó tận dụng các đảm bảo ACID đã được kiểm chứng của chính cơ sở dữ liệu cục bộ. Mọi thay đổi dữ liệu nghiệp vụ và event gửi đi tương ứng sẽ cùng chia sẻ một database transaction nguyên tử (atomic) duy nhất.
Quy trình 3 bước
- Ghi nguyên tử duy nhất (Single Atomic Write): Khi sửa đổi bảng
orders, ứng dụng đồng thời ghi một bản ghi vào bảngoutbox_events. Nếu database transaction commit thành công, cả hai bản ghi đều được lưu vào đĩa. Nếu thất bại, cả hai đều được rollback hoàn toàn. Tuyệt đối không có ghost event. - Chuyển tiếp message bất đồng bộ (Asynchronous Message Relay): Một background worker độc lập sẽ quét (poll) các bản ghi đang chờ xử lý từ
outbox_events, publish chúng tới broker, và đánh dấu là đã xử lý. - Phân phối ít nhất một lần (At-Least-Once Delivery): Worker có thể publish một event tới Kafka rồi bị crash trước khi kịp cập nhật lại PostgreSQL. Các downstream consumer đôi khi sẽ nhận được message trùng lặp, vì vậy hãy thiết kế chúng có tính idempotent.
Các công cụ Change Data Capture (CDC) như Debezium rất vượt trội đối với các kiến trúc phân tán quy mô lớn. Tuy nhiên, với các hệ thống xử lý từ 500 đến 5.000 transaction mỗi giây, một Go worker chạy in-process kết hợp với cơ chế FOR UPDATE SKIP LOCKED của PostgreSQL mang lại độ tin cậy vững chắc mà hầu như không tốn thêm chi phí vận hành.
Từng bước triển khai trong Go
1. Thiết kế Schema trong PostgreSQL
Bắt đầu bằng việc tạo schema cho dữ liệu nghiệp vụ và bảng log outbox event:
CREATE TABLE orders (
id UUID PRIMARY KEY,
customer_id UUID NOT NULL,
amount_cents BIGINT NOT NULL,
status VARCHAR(32) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE outbox_events (
id UUID PRIMARY KEY,
aggregate_type VARCHAR(64) NOT NULL,
aggregate_id UUID NOT NULL,
event_type VARCHAR(64) NOT NULL,
payload JSONB NOT NULL,
status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
processed_at TIMESTAMPTZ
);
-- Partial index giúp truy vấn dưới một mili-giây ngay cả khi có hàng triệu dòng lịch sử
CREATE INDEX idx_outbox_pending ON outbox_events (created_at) WHERE status = 'PENDING';
Hãy chú ý đến partial index trên điều kiện status = 'PENDING'. Cơ chế này giúp việc tra cứu index duy trì dưới 2ms, ngay cả sau khi bảng đã tích lũy hàng triệu bản ghi lịch sử.
2. Ghi dữ liệu nguyên tử trong Go
Gói thao tác cập nhật dữ liệu nghiệp vụ và chèn bản ghi outbox vào trong một database transaction duy nhất:
package main
import (
"context"
"database/sql"
"encoding/json"
"time"
"github.com/google/uuid"
)
type OrderPaidEvent struct {
OrderID uuid.UUID `json:"order_id"`
AmountCents int64 `json:"amount_cents"`
Timestamp time.Time `json:"timestamp"`
}
func CompleteOrder(ctx context.Context, db *sql.DB, orderID uuid.UUID, amountCents int64) error {
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
if err != nil {
return err
}
defer tx.Rollback()
// 1. Cập nhật trạng thái nghiệp vụ
updateQuery := `UPDATE orders SET status = 'PAID' WHERE id = $1`
if _, err := tx.ExecContext(ctx, updateQuery, orderID); err != nil {
return err
}
// 2. Tuần tự hóa (serialize) payload của event
payload, err := json.Marshal(OrderPaidEvent{
OrderID: orderID,
AmountCents: amountCents,
Timestamp: time.Now().UTC(),
})
if err != nil {
return err
}
// 3. Đưa event vào outbox trong cùng transaction
outboxQuery := `
INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload, status)
VALUES ($1, $2, $3, $4, $5, 'PENDING')
`
eventID := uuid.New()
if _, err := tx.ExecContext(ctx, outboxQuery, eventID, "Order", orderID, "OrderPaid", payload); err != nil {
return err
}
return tx.Commit()
}
3. Background Worker an toàn khi chạy nhiều Replica
Chạy nhiều worker instance giúp ngăn ngừa nghẽn cổ chai, nhưng việc polling thông thường sẽ gây ra tranh chấp khóa (lock contention). PostgreSQL giải quyết vấn đề này bằng FOR UPDATE SKIP LOCKED. Mỗi worker sẽ giữ một exclusive lock trên một batch chưa được gán, trong khi các worker khác sẽ lập tức bỏ qua để xử lý các hàng khả dụng tiếp theo.
package main
import (
"context"
"database/sql"
"log"
"time"
"github.com/google/uuid"
)
type OutboxRecord struct {
ID uuid.UUID
EventType string
Payload []byte
}
type MessagePublisher interface {
Publish(ctx context.Context, topic string, data []byte) error
}
func StartOutboxWorker(ctx context.Context, db *sql.DB, publisher MessagePublisher, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := processBatch(ctx, db, publisher); err != nil {
log.Printf("lỗi chuyển tiếp outbox: %v", err)
}
}
}
}
func processBatch(ctx context.Context, db *sql.DB, publisher MessagePublisher) error {
// Giới hạn thời gian thực thi của mỗi batch chuyển tiếp là 5 giây
batchCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
tx, err := db.BeginTx(batchCtx, nil)
if err != nil {
return err
}
defer tx.Rollback()
query := `
SELECT id, event_type, payload
FROM outbox_events
WHERE status = 'PENDING'
ORDER BY created_at ASC
LIMIT 50
FOR UPDATE SKIP LOCKED
`
rows, err := tx.QueryContext(batchCtx, query)
if err != nil {
return err
}
defer rows.Close()
var batch []OutboxRecord
for rows.Next() {
var rec OutboxRecord
if err := rows.Scan(&rec.ID, &rec.EventType, &rec.Payload); err != nil {
return err
}
batch = append(batch, rec)
}
if err := rows.Err(); err != nil {
return err
}
if len(batch) == 0 {
return nil
}
for _, event := range batch {
if err := publisher.Publish(batchCtx, event.EventType, event.Payload); err != nil {
// Rollback để giữ nguyên trạng thái PENDING cho lần thử lại ở chu kỳ tiếp theo
return err
}
updateQuery := `
UPDATE outbox_events
SET status = 'PROCESSED', processed_at = $1
WHERE id = $2
`
if _, err := tx.ExecContext(batchCtx, updateQuery, time.Now().UTC(), event.ID); err != nil {
return err
}
}
return tx.Commit()
}
Những điểm cốt lõi và kinh nghiệm triển khai Production
Transactional Outbox Pattern tách biệt việc ghi dữ liệu nghiệp vụ khỏi tình trạng hoạt động (uptime) của message broker bên ngoài. Nó đảm bảo rằng một event chỉ được phát đi khi và chỉ khi trạng thái dữ liệu trong cơ sở dữ liệu đã thay đổi thành công.
Trước khi đưa giải pháp này lên production, hãy lưu ý hai biện pháp bảo vệ khi vận hành sau:
- Giữ timeout của lệnh publish ở mức ngắn: Giới hạn thời gian gọi
publisher.Publishtrong khoảng 200–500ms. Do việc publish diễn ra bên trong một transaction đang hoạt động, các lời gọi broker bị chậm sẽ chiếm dụng kết nối cơ sở dữ liệu và giữ lock bảng. - Chủ động dọn dẹp các bản ghi đã xử lý: Một bảng tăng thêm 100.000 event mỗi ngày sẽ phình to rất nhanh. Hãy thiết lập một cron job chạy hàng ngày để xóa hoặc phân vùng (partition) các bản ghi có
status = 'PROCESSED'vàprocessed_at < NOW() - INTERVAL '7 days'.
