Cơn ác mộng: Khi Database và Message Broker “lệch pha”
Bạn hãy tưởng tượng: Khách hàng nhấn thanh toán đơn hàng 2 triệu đồng, hệ thống trừ tiền và lưu database thành công. Tuy nhiên, đúng lúc đó mạng chập chờn khiến service kho không nhận được message để đóng gói hàng. Kết quả là khách mất tiền nhưng đơn hàng vẫn nằm im, còn bạn phải thức đến 2 giờ sáng để đối soát log và sửa dữ liệu bằng tay.
Tôi từng trực tiếp xử lý một sự cố mất gần 5% số lượng message sự kiện trong một hệ thống thanh toán lớn. Lúc đó, team dùng cách đơn giản: Lưu đơn hàng vào PostgreSQL xong thì gọi publishMessage() sang RabbitMQ. Khi traffic tăng cao, Database commit xong nhưng lệnh gửi message bị timeout. Đơn hàng nằm chết trong DB, còn các service hạ tầng phía sau hoàn toàn mù tịt.
Sự cố đó khiến chúng tôi mất 3 ngày ròng rã để viết script bù dữ liệu. Kinh nghiệm xương máu ở đây là: Đừng bao giờ tin rằng hai hệ thống độc lập sẽ luôn hoạt động đồng bộ nếu thiếu cơ chế bảo vệ chặt chẽ.
Tại sao các cách làm thông thường lại dễ thất bại?
1. Dual Write (Ghi đồng thời)
Đây là sai lầm phổ biến nhất. Bạn thực hiện hai hành động độc lập trong code: Ghi vào DB và Ghi vào Message Queue.
async function createOrder(orderData) {
const order = await db.orders.create(orderData); // (1) Thành công
await rabbitMQ.publish('order_created', order); // (2) Nếu mạng lỗi ở đây, dữ liệu sẽ bị lệch!
}Vấn đề nằm ở chỗ bước (1) và (2) không nằm trong một transaction duy nhất. Nếu bước (2) lỗi, bước (1) không thể tự động rollback, dẫn đến tình trạng “râu ông nọ cắm cằm bà kia”.
2. Distributed Transactions (2PC)
Cách này cố gộp cả DB và Broker vào một giao dịch chung. Tuy nhiên, 2PC cực kỳ phức tạp và làm giảm hiệu năng hệ thống đáng kể. Trong môi trường Microservices hiện đại, 2PC thường là “cơn ác mộng” về bảo trì và khả năng mở rộng.
Transactional Outbox Pattern: Giải pháp từ tính chất ACID
Thay vì cố gửi message ngay, chúng ta sẽ tận dụng chính tính chất ACID của PostgreSQL. Ý tưởng cốt lõi rất thực tế:
- Tạo một bảng phụ gọi là
outboxngay trong Database của bạn. - Khi lưu đơn hàng, bạn chèn thêm một bản ghi vào bảng
outboxtrong cùng một Transaction. - Một tiến trình riêng (Relay) sẽ quét bảng
outbox, gửi message đi và đánh dấu thành công.
Vì việc lưu đơn hàng và lưu message vào outbox dùng chung một transaction, nên hoặc là cả hai cùng thành công, hoặc là không có gì thay đổi. Dữ liệu của bạn luôn ở trạng thái nhất quán tuyệt đối.
Hướng dẫn triển khai thực tế với PostgreSQL và Node.js
Bước 1: Thiết kế bảng Outbox
Chúng ta cần một cấu trúc bảng đủ linh hoạt để chứa nhiều loại sự kiện khác nhau.
CREATE TABLE orders (
id SERIAL PRIMARY KEY,
customer_name TEXT NOT NULL,
total_amount DECIMAL NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE outbox (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
aggregate_type TEXT NOT NULL, -- Ví dụ: 'Order'
aggregate_id TEXT NOT NULL, -- ID đơn hàng
type TEXT NOT NULL, -- Event: 'OrderCreated'
payload JSONB NOT NULL, -- Dữ liệu thực tế
status TEXT DEFAULT 'PENDING', -- PENDING, PROCESSED, FAILED
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);Bước 2: Viết Business Logic chuẩn Transaction
Sử dụng thư viện pg, bạn phải dùng chung một client để đảm bảo tính nguyên tử (Atomicity). Nếu một câu lệnh lỗi, toàn bộ thay đổi sẽ bị hủy bỏ.
async function createOrder(customerName, amount) {
const client = await pool.connect();
try {
await client.query('BEGIN');
// 1. Lưu đơn hàng vào bảng chính
const orderRes = await client.query(
'INSERT INTO orders (customer_name, total_amount) VALUES ($1, $2) RETURNING id',
[customerName, amount]
);
const orderId = orderRes.rows[0].id;
// 2. Lưu message vào bảng outbox (Cùng transaction)
const outboxPayload = { orderId, customerName, amount };
await client.query(
'INSERT INTO outbox (aggregate_type, aggregate_id, type, payload) VALUES ($1, $2, $3, $4)',
['Order', orderId.toString(), 'OrderCreated', JSON.stringify(outboxPayload)]
);
await client.query('COMMIT');
console.log('Giao dịch thành công!');
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}Bước 3: Xây dựng Message Relay (Worker chuyển tiếp)
Để đưa message từ DB sang Broker, Polling là cách tiếp cận đơn giản và hiệu quả nhất cho các hệ thống vừa và nhỏ.
async function relayMessages() {
const client = await pool.connect();
try {
// Lấy 10 message chưa xử lý, khóa dòng để tránh worker khác đọc trùng
const res = await client.query(
"SELECT * FROM outbox WHERE status = 'PENDING' FOR UPDATE SKIP LOCKED LIMIT 10"
);
for (const row of res.rows) {
try {
await messageBroker.publish(row.type, row.payload);
await client.query(
"UPDATE outbox SET status = 'PROCESSED' WHERE id = $1",
[row.id]
);
} catch (err) {
console.error(`Lỗi gửi message ${row.id}:`, err);
}
}
} finally {
client.release();
}
}
setInterval(relayMessages, 5000);Kỹ thuật FOR UPDATE SKIP LOCKED là chìa khóa ở đây. Nó cho phép bạn chạy nhiều instance Worker cùng lúc mà không sợ gửi trùng một message nhiều lần.
Kinh nghiệm tối ưu khi vận hành thực tế
Khi áp dụng pattern này vào dự án có hàng triệu transaction, bạn cần lưu ý ba điểm sau:
- Quản lý dung lượng DB: Bảng outbox sẽ phình to rất nhanh. Nếu mỗi ngày có 1 triệu đơn hàng, sau một tháng bảng sẽ có 30 triệu dòng, làm chậm câu lệnh query. Hãy lập lịch xóa (cleanup) các bản ghi
PROCESSEDsau 3-7 ngày. - Xử lý trùng lặp (Idempotency): Pattern này đảm bảo message “chắc chắn được gửi ít nhất một lần”. Tuy nhiên, nếu worker crash ngay sau khi gửi nhưng trước khi update DB, message sẽ bị gửi lại. Phía nhận (Consumer) bắt buộc phải kiểm tra trùng lặp.
- Cân nhắc dùng CDC: Nếu hệ thống đạt ngưỡng >1000 transaction/giây, việc Polling liên tục sẽ gây áp lực lên DB. Lúc này, hãy cân nhắc dùng Debezium để đọc file WAL của PostgreSQL, giúp chuyển tiếp dữ liệu mà không cần query trực tiếp.
Lời kết
Transactional Outbox không chỉ là một kỹ thuật code, nó là tư duy thiết kế hệ thống bền bỉ. Thay vì tin vào sự may mắn của kết nối mạng, chúng ta tin vào tính nhất quán của Database. Chúc các bạn áp dụng thành công và ngủ ngon hơn mỗi khi hệ thống có cao điểm!
