Posted inDatabase

Transactional Outbox Pattern: Xuất bản Event Đáng tin cậy trong Kiến trúc Microservices

Database tutorial - IT technology blog
Database tutorial - IT technology blog

Bắt đầu Nhanh (5 Phút): Chấm dứt Cơn ác mộng Dual-Write

Hãy hình dung một luồng thanh toán điển hình: khách hàng đặt đơn trên trang web của bạn. Service của bạn commit đơn hàng vào PostgreSQL và gửi ngay một event order_created tới RabbitMQ hoặc Apache Kafka.

# Cách tiếp cận ngây thơ: anti-pattern dual-write
def create_order(order_data):
    db.orders.insert(order_data)  # Bước 1: Lưu vào Postgres
    message_broker.publish("order_created", order_data)  # Bước 2: Điều gì xảy ra nếu bước này thất bại?

Điều gì xảy ra nếu Kafka bị ngắt kết nối ngay sau khi database của bạn vừa commit? Database thì đã lưu đơn hàng, nhưng các service thanh toán, kho vận và thông báo lại không hề hay biết. Đảo ngược thứ tự xử lý cũng mang lại rủi ro tương tự. Nếu bạn publish lên broker trước rồi transaction trong database bị rollback, các worker ở hạ nguồn sẽ xử lý một đơn hàng “ma”.

Các kỹ sư gọi đây là bài toán dual-write. Vì cơ sở dữ liệu quan hệ và message broker không dùng chung một distributed transaction, việc điều phối các thao tác ghi trên cả hai hệ thống thường rất thiếu tin cậy. Two-Phase Commit (2PC) thì quá chậm và dễ đổ vỡ đối với microservices. Thay vào đó, chúng ta dựa vào Transactional Outbox Pattern.

Giải pháp Outbox trong 5 Phút

Hãy ngừng tương tác đồng thời với hai hệ thống riêng biệt trong cùng một request path. Hãy ghi cả business entity lẫn message cần gửi ra ngoài vào cùng một database trong một transaction ACID mang tính nguyên tử duy nhất.

-- Tạo bảng outbox trong cơ sở dữ liệu của bạn
CREATE TABLE outbox_messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(50) NOT NULL,
    aggregate_id VARCHAR(50) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    processed BOOLEAN DEFAULT FALSE
);

Bây giờ, hãy commit cả hai bản ghi cùng lúc:

def place_order(order_data):
    with db.transaction():
        order_id = db.orders.insert(order_data)
        db.outbox.insert({
            "aggregate_type": "order",
            "aggregate_id": str(order_id),
            "event_type": "OrderPlaced",
            "payload": order_data,
            "processed": False
        })

Chúng chia sẻ chung một local transaction. Cả hai bản ghi hoặc cùng được commit xuống ổ đĩa, hoặc không có bản ghi nào được ghi nhận. Sau đó, một background worker chuyên biệt sẽ đọc các dòng từ outbox_messages và chuyển tiếp chúng đến broker một cách an toàn.

Phân tích Chuyên sâu: Cơ chế Hoạt động của Transactional Outbox Pattern

Pattern này giúp tách rời logic nghiệp vụ (domain logic) khỏi quá trình vận chuyển message bất đồng bộ. Nó chuyển đổi các lệnh gọi mạng (network calls) bên ngoài theo thời gian thực thành một pipeline tin cậy được điều phối trực tiếp bởi database.

Các Thành phần Cốt lõi

  • Command Handler: Xử lý các lệnh gọi API đến, cập nhật các bảng domain như orders, và ghi các event chuẩn bị gửi vào outbox_messages trong cùng một transaction nội bộ.
  • Outbox Table: Đóng vai trò như một hàng đợi staging bền vững và có thể truy vấn ngay bên trong datastore chính của bạn (PostgreSQL, MySQL hoặc SQLite).
  • Message Relay: Một background worker bất đồng bộ đảm nhận việc trích xuất các message đang chờ xử lý từ bảng outbox và chuyển tiếp chúng tới broker (chẳng hạn như Kafka, RabbitMQ hoặc AWS SNS/SQS).

Hai Chiến lược Message Relay Phổ biến

1. Polling Publisher

Một background worker sẽ truy vấn database theo chu kỳ định kỳ (ví dụ: mỗi 500ms một lần):

SELECT * FROM outbox_messages
WHERE processed = FALSE
ORDER BY created_at ASC
LIMIT 100
FOR UPDATE SKIP LOCKED;

Mệnh đề FOR UPDATE SKIP LOCKED là mấu chốt quan trọng ở đây. Nó thông báo cho Postgres bỏ qua bất kỳ hàng nào hiện đang bị các worker thread khác chiếm giữ (lock). Bạn có thể chạy song song từ 4 đến 8 relay instance mà không cần dùng distributed lock hay lo ngại race condition. Khi một batch đã được gửi thành công vào broker, worker sẽ cập nhật processed = TRUE hoặc xóa hẳn các hàng đó đi.

2. Transaction Log Tailing (Change Data Capture / CDC)

Thay vì chạy các câu lệnh polling liên tục vào các bảng chính, bạn có thể khai thác trực tiếp luồng transaction log của engine cơ sở dữ liệu. Các công cụ như Debezium sẽ đọc Write-Ahead Log (WAL) của PostgreSQL hoặc binlog của MySQL và truyền các thay đổi dữ liệu đến Kafka với độ trễ dưới 10ms. Cách tiếp cận này loại bỏ hoàn toàn các truy vấn polling và giải tỏa áp lực tải đọc trên database chính.

Sử dụng Nâng cao: Xử lý Concurrency, Thứ tự Event và Lỗi hệ thống

Thiết lập vòng lặp cơ bản chỉ mất một buổi chiều. Nhưng xử lý trơn tru các lỗi phát sinh trên production đòi hỏi sự cẩn trọng cao độ.

At-Least-Once Delivery & Tính Idempotency

Sự cố mạng là điều không thể tránh khỏi. Một relay worker có thể đã đẩy thành công message lên Kafka, nhưng lại gặp sự cố sập nguồn ngay trước khi kịp commit trạng thái processed = TRUE trở lại PostgreSQL. Khi worker này khởi động lại, nó sẽ gửi lại chính xác event đó. Outbox pattern chỉ đảm bảo cơ chế at-least-once delivery (phân phối ít nhất một lần), chứ không bao giờ đảm bảo strictly once (duy nhất một lần).

Các consumer ở tầng hạ nguồn bắt buộc phải có tính chất idempotent (bất biến với thao tác lặp lại). Hãy luôn lưu lại ID của các message đã xử lý vào một bảng deduplication riêng biệt:

def handle_order_placed(event):
    event_id = event["id"]
    with db.transaction():
        if db.processed_events.exists(event_id):
            return  # Đã nhận được message trùng lặp; bỏ qua an toàn
        
        apply_billing(event["payload"])
        db.processed_events.insert({"event_id": event_id})

Partitioning và Đảm bảo Thứ tự Event

Hãy xét trường hợp một đơn hàng được cập nhật 3 lần trong vòng 2 giây: vừa tạo, thay đổi địa chỉ, rồi bị hủy. Nếu các thông tin này đến service giao hàng không đúng trình tự, bạn có nguy cơ giao nhầm một đơn hàng vốn đã bị hủy. Khi xuất bản các dòng trong outbox lên Kafka, hãy gán message key bằng aggregate_id (chẳng hạn như order_id). Kafka sẽ băm (hash) key này để điều hướng mọi bản cập nhật của cùng một đơn hàng vào đúng một partition duy nhất, duy trì thứ tự FIFO nghiêm ngặt.

Dọn dẹp Bảng Outbox Khi Lưu lượng Lớn

Một dịch vụ thương mại điện tử xử lý 50.000 đơn hàng mỗi ngày sẽ tạo ra 1,5 triệu dòng outbox mỗi tháng. Nếu không được kiểm soát, hiện tượng phình to bảng (table bloat) sẽ làm chậm quá trình quét dữ liệu và làm cạn kiệt dung lượng đĩa. Bạn có thể áp dụng 3 giải pháp thực tế sau:

  1. Hard Delete ngay khi Publish: Thực thi lệnh DELETE FROM outbox_messages WHERE id = :id ngay khi broker xác nhận đã nhận message. Cách này giữ cho bảng outbox luôn gọn nhẹ, hiếm khi vượt quá vài chục dòng.
  2. Khai báo Partitioning cho Bảng: Chia partition cho outbox_messages theo từng ngày. Khi một partition trong ngày đã được xử lý xong toàn bộ, chỉ cần chạy lệnh DROP TABLE. Thao tác này giải phóng dung lượng đĩa ngay tức thì mà không gây tốn tài nguyên vacuum từng dòng.
  3. Dọn dẹp Batch theo Lịch định kỳ: Chạy một cron job dọn dẹp vào ban đêm trong khung giờ ít truy cập để loại bỏ các bản ghi cũ: DELETE FROM outbox_messages WHERE processed = TRUE AND created_at < NOW() - INTERVAL '3 days' LIMIT 10000;.

Lời khuyên Thực tế: Checklist Triển khai và Kinh nghiệm Thực chiến

Hãy ghi nhớ những nguyên tắc đã được kiểm chứng thực tế này khi triển khai bảng outbox lên môi trường production:

  • Giữ payload độc lập và đầy đủ: Đóng gói toàn bộ các thay đổi trạng thái trực tiếp vào event payload. Các consumer không nên phải quay ngược lại và gọi REST API của service chỉ để lấy thêm thông tin chi tiết về khách hàng còn thiếu.
  • Lưu trữ dữ liệu có cấu trúc native: Ưu tiên dùng kiểu JSONB của PostgreSQL thay vì chuỗi string không index. Nó cho phép bạn thực hiện các câu truy vấn quản trị trực tiếp trên các thuộc tính của event mỗi khi cần debug các payload bị treo.
  • Đánh index tối ưu cho truy vấn polling: Hãy bổ sung một partial composite index: CREATE INDEX idx_unprocessed_outbox ON outbox_messages (created_at ASC) WHERE processed = FALSE;. Index này đảm bảo tốc độ tra cứu luôn cực nhanh ngay cả khi các bản ghi lịch sử tích tụ nhiều.
  • Thiết lập cảnh báo độ trễ relay: Giám sát thời gian tồn tại của bản ghi cũ nhất đang có trạng thái processed = FALSE. Nếu có bất kỳ message nào chưa được xử lý quá 30 giây, hãy lập tức gửi cảnh báo tới kỹ sư on-call qua PagerDuty.

Khi chuẩn bị dữ liệu mẫu cho môi trường staging hoặc mock các payload thử nghiệm, việc chuyển đổi dữ liệu thủ công thường rất tốn thời gian. Nếu bạn cần chuyển dữ liệu fixture từ CSV sang JSON để chạy thử nghiệm outbox, công cụ toolcraft.app/vi/tools/data/csv-to-json sẽ xử lý trực tiếp ngay trong trình duyệt mà không để lộ schema nội bộ ra ngoài máy của bạn. Điều này giúp bạn không phải mất công viết các đoạn script import tạm bợ khi kiểm thử các trường hợp edge case.

Gắn chặt việc phát sinh event vào các transaction trong database nội bộ sẽ ngăn chặn triệt để sự sai lệch dữ liệu ngay từ gốc, mang lại cho hệ thống microservices một nền tảng giao tiếp bền bỉ và vững vàng dưới tải cao.

Share:
Tags: