Bối cảnh & Lý do
Hầu hết các ứng dụng web trong môi trường production hiện nay vẫn phụ thuộc vào các thao tác CRUD trực tiếp trên cơ sở dữ liệu quan hệ. Trong một dịch vụ thanh toán hoặc thương mại điện tử, việc điều chỉnh số dư tài khoản thường diễn ra như sau:
UPDATE accounts SET balance = balance - 50.00, updated_at = NOW() WHERE id = 'acc-102';Câu lệnh đơn lẻ đó đã âm thầm xóa sạch toàn bộ lịch sử. Khi bộ phận tài chính yêu cầu kiểm toán sau sáu tháng, hoặc khi hai background worker đồng thời thực hiện thao tác trừ tiền, bạn sẽ không thể trả lời các câu hỏi điều tra cơ bản: Tại sao số dư lại giảm? Ai đã cấp quyền cho giao dịch đó? Đó là khoản trừ tiền định kỳ tự động, điều chỉnh phí thủ công hay một giao dịch hoàn tiền?
Nhược điểm này bắt nguồn từ tính chất thay đổi được của dữ liệu (state mutability). Khi write model kiêm luôn vai trò của read model, các bản ghi chỉ phản ánh snapshot mới nhất. Bạn đã vứt bỏ toàn bộ dòng thời gian diễn biến. Trong các kiến trúc có thông lượng cao xử lý hàng nghìn lượt ghi mỗi giây, cấu hình này dẫn đến tình trạng tranh chấp khóa hàng (row-lock contention), deadlock và sai lệch đồng bộ dữ liệu âm thầm giữa các dịch vụ.
Event Sourcing kết hợp với Command Query Responsibility Segregation (CQRS) giải quyết triệt để vấn đề này bằng cách tách biệt hoàn toàn luồng ghi và luồng đọc:
- Event Sourcing: Các thay đổi về state trở thành những domain event bất biến, chỉ ghi thêm (append-only) như
AccountOpenedhayMoneyDeposited. Số dư hiện tại là kết quả thuần túy của phép fold trên stream event đó. - CQRS: Các command handler thực hiện xác thực đầu vào và ghi event vào một log chỉ ghi thêm. Các projection handler tách biệt sẽ đọc các event đó để duy trì các read view đã denormalize, thân thiện với việc truy vấn.
Khi đội ngũ của chúng tôi chuyển đổi một sổ cái thanh toán cốt lõi sang kiến trúc này trong đợt cao điểm đạt 1.400 giao dịch mỗi giây, hiện tượng lỗi double-spend đã giảm về 0. Đội ngũ compliance có được một audit trail hoàn chỉnh, có thể xác thực và tái hiện lại tình trạng tài chính của bất kỳ người dùng nào tại bất kỳ thời điểm nào trong lịch sử.
Cài đặt
Để giữ cho ví dụ tập trung và dễ tái hiện mà không cần triển khai một cụm Kafka hay PostgreSQL cồng kềnh, chúng ta sẽ xây dựng một engine in-process sử dụng Node.js, Express và better-sqlite3 để thực hiện ghi đĩa đồng bộ với độ trễ bằng không.
Khởi tạo một workspace dự án sạch và cài đặt các package cần thiết:
mkdir node-cqrs-eventsourcing
cd node-cqrs-eventsourcing
npm init -y
npm install express better-sqlite3 uuid
npm install --save-dev nodemonThêm "type": "module" vào tệp package.json của bạn để kích hoạt cú pháp import ES module gốc:
{
"name": "node-cqrs-eventsourcing",
"version": "1.0.0",
"type": "module",
"scripts": {
"start": "node server.js",
"dev": "nodemon server.js"
}
}Cấu hình
Hệ thống được chia thành ba thành phần riêng biệt: Event Store chỉ ghi thêm (append-only), Command Aggregate và Read Projector.
1. Thiết lập Storage Schema
Trong tệp database.js, khởi tạo cả hai bảng: log event chỉ ghi thêm và view projection được tối ưu hóa cho việc đọc.
import Database from 'better-sqlite3';
const db = new Database('system.db');
// Tối ưu hóa SQLite cho các thao tác đọc và ghi đồng thời
db.pragma('journal_mode = WAL');
db.exec(`
CREATE TABLE IF NOT EXISTS event_store (
id INTEGER PRIMARY KEY AUTOINCREMENT,
stream_id TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
version INTEGER NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(stream_id, version)
);
CREATE TABLE IF NOT EXISTS read_account_balances (
account_id TEXT PRIMARY KEY,
owner TEXT NOT NULL,
balance REAL NOT NULL,
last_version INTEGER NOT NULL
);
`);
export default db;Hãy chú ý kỹ đến ràng buộc UNIQUE(stream_id, version). Ràng buộc cơ sở dữ liệu này mang lại cơ chế optimistic concurrency control có tính tất định. Nếu hai request đồng thời cố gắng ghi nối tiếp version 3 cho cùng một stream, SQLite sẽ từ chối ngay lập tức thao tác ghi thứ hai.
2. Event Store và Projection Engine
Tiếp theo, tạo tệp eventEngine.js để xử lý việc lưu trữ event, stream hydration và dispatch dữ liệu sang read model:
import db from './database.js';
export class EventStore {
static getEvents(streamId) {
const rows = db.prepare(
'SELECT event_type, payload, version FROM event_store WHERE stream_id = ? ORDER BY version ASC'
).all(streamId);
return rows.map(r => ({
type: r.event_type,
payload: JSON.parse(r.payload),
version: r.version
}));
}
static append(streamId, expectedVersion, eventType, payload) {
const newVersion = expectedVersion + 1;
const insertEvent = db.prepare(`
INSERT INTO event_store (stream_id, event_type, payload, version, created_at)
VALUES (?, ?, ?, ?, ?)
`);
try {
insertEvent.run(streamId, eventType, JSON.stringify(payload), newVersion, new Date().toISOString());
// Trong cấu hình cục bộ này, thực hiện projection ngay sau khi commit
Projector.handle({
streamId,
type: eventType,
payload,
version: newVersion
});
} catch (err) {
if (err.code === 'SQLITE_CONSTRAINT_UNIQUE') {
throw new Error(`ConcurrencyConflictException: Stream ${streamId} đã bị cập nhật bởi một tiến trình khác.`);
}
throw err;
}
}
}
export class Projector {
static handle(event) {
const { streamId, type, payload, version } = event;
if (type === 'AccountOpened') {
db.prepare(`
INSERT INTO read_account_balances (account_id, owner, balance, last_version)
VALUES (?, ?, ?, ?)
`).run(streamId, payload.owner, payload.initialBalance, version);
} else if (type === 'MoneyDeposited') {
db.prepare(`
UPDATE read_account_balances
SET balance = balance + ?, last_version = ?
WHERE account_id = ?
`).run(payload.amount, version, streamId);
} else if (type === 'MoneyDebited') {
db.prepare(`
UPDATE read_account_balances
SET balance = balance - ?, last_version = ?
WHERE account_id = ?
`).run(payload.amount, version, streamId);
}
}
}3. Tái tạo State của Aggregate và Xử lý Command
Các command tuyệt đối không bao giờ được truy vấn read projection để đưa ra quyết định nghiệp vụ. Read model có tính chất eventually consistent và có thể bị trễ. Thay vào đó, command tải trực tiếp lịch sử từ event store, tái tạo lại state nội bộ trong bộ nhớ, thực thi các kiểm tra bất biến (invariants) và tạo ra event mới.
Thêm ranh giới aggregate vào tệp accountAggregate.js:
import { EventStore } from './eventEngine.js';
export class BankAccount {
constructor(id) {
this.id = id;
this.balance = 0;
this.version = 0;
this.isOpened = false;
}
loadFromHistory() {
const events = EventStore.getEvents(this.id);
for (const evt of events) {
this.apply(evt.type, evt.payload);
this.version = evt.version;
}
}
apply(type, payload) {
switch (type) {
case 'AccountOpened':
this.balance = payload.initialBalance;
this.isOpened = true;
break;
case 'MoneyDeposited':
this.balance += payload.amount;
break;
case 'MoneyDebited':
this.balance -= payload.amount;
break;
}
}
open(owner, initialBalance) {
if (this.isOpened) throw new Error('Tài khoản đã tồn tại.');
if (initialBalance < 0) throw new Error('Số tiền gửi ban đầu không được âm.');
EventStore.append(this.id, this.version, 'AccountOpened', { owner, initialBalance });
}
deposit(amount) {
if (!this.isOpened) throw new Error('Tài khoản không tồn tại.');
if (amount <= 0) throw new Error('Số tiền gửi phải lớn hơn 0.');
EventStore.append(this.id, this.version, 'MoneyDeposited', { amount });
}
debit(amount) {
if (!this.isOpened) throw new Error('Tài khoản không tồn tại.');
if (amount <= 0) throw new Error('Số tiền ghi nợ không hợp lệ.');
if (this.balance < amount) throw new Error(`Không đủ số dư: số dư hiện tại là ${this.balance}, yêu cầu rút ${amount}.`);
EventStore.append(this.id, this.version, 'MoneyDebited', { amount });
}
}4. Phơi bày CQRS API
Tập hợp HTTP server trong tệp server.js, duy trì sự tách biệt rõ ràng giữa các command ghi và query đọc qua từng route:
import express from 'express';
import { v4 as uuidv4 } from 'uuid';
import db from './database.js';
import { BankAccount } from './accountAggregate.js';
const app = express();
app.use(express.json());
// PHÍA COMMAND: Xác thực business rules và append event
app.post('/api/accounts', (req, res) => {
try {
const accountId = uuidv4();
const account = new BankAccount(accountId);
account.open(req.body.owner, Number(req.body.initialBalance) || 0);
res.status(201).json({ accountId, message: 'Đã khởi tạo account stream thành công.' });
} catch (err) {
res.status(400).json({ error: err.message });
}
});
app.post('/api/accounts/:id/debit', (req, res) => {
try {
const account = new BankAccount(req.params.id);
account.loadFromHistory();
account.debit(Number(req.body.amount));
res.json({ message: 'Ghi nhận giao dịch trừ tiền thành công.' });
} catch (err) {
res.status(400).json({ error: err.message });
}
});
// PHÍA QUERY: Đọc trực tiếp từ projection đã được tổng hợp trước
app.get('/api/accounts/:id', (req, res) => {
const row = db.prepare('SELECT * FROM read_account_balances WHERE account_id = ?').get(req.params.id);
if (!row) return res.status(404).json({ error: 'Không tìm thấy tài khoản.' });
res.json(row);
});
app.listen(3000, () => {
console.log('Dịch vụ Node CQRS đang lắng nghe trên cổng 3000');
});Kiểm thử & Giám sát
Khởi động server để kiểm tra việc xử lý command, tái tạo state và replay projection dưới tải làm việc thực tế:
node server.js1. Kiểm thử Chức năng
Mở một tài khoản mới với số tiền gửi ban đầu là $200.00:
curl -s -X POST http://localhost:3000/api/accounts \
-H "Content-Type: application/json" \
-d '{"owner": "Alex M.", "initialBalance": 200}'Dịch vụ phản hồi với một stream ID duy nhất:
{"accountId":"7f3a8f90-1c05-4f2b-8a8b-1e240b3cb1a4","message":"Đã khởi tạo account stream thành công."}Tiếp theo, gửi command trừ tiền $75.00 tới stream đó:
curl -s -X POST http://localhost:3000/api/accounts/7f3a8f90-1c05-4f2b-8a8b-1e240b3cb1a4/debit \
-H "Content-Type: application/json" \
-d '{"amount": 75}'Gọi projection để kiểm tra read state đã được tính toán:
curl -s http://localhost:3000/api/accounts/7f3a8f90-1c05-4f2b-8a8b-1e240b3cb1a4Projection trả về số dư và sequence number chính xác:
{"account_id":"7f3a8f90-1c05-4f2b-8a8b-1e240b3cb1a4","owner":"Alex M.","balance":125,"last_version":2}Bây giờ, hãy thử nghiệm việc thực thi ranh giới nghiệp vụ bằng cách cố gắng trừ $500.00. Command này sẽ thất bại ngay lập tức bên trong accountAggregate.js với mã lỗi HTTP 400 cùng với {"error": "Không đủ số dư: số dư hiện tại là 125, yêu cầu rút 500."}. Không có event nào được ghi, và read projection vẫn được giữ nguyên ở version 2.
2. Khôi phục sau Sự cố: Replay Projection
Read model thực chất là các bộ nhớ cache có thể loại bỏ và tạo lại bất kỳ lúc nào. Nếu một bug làm sai lệch bảng projection hoặc yêu cầu schema thay đổi, bạn không cần phải khôi phục bản backup cơ sở dữ liệu. Bạn hoàn toàn có thể xóa bảng projection và replay lại log event thô từ version 1.
Viết một công cụ khôi phục vận hành trong tệp replay.js:
// replay.js
import db from './database.js';
import { Projector } from './eventEngine.js';
console.log('Đang xóa dữ liệu bảng projection...');
db.exec('DELETE FROM read_account_balances;');
const events = db.prepare('SELECT stream_id, event_type, payload, version FROM event_store ORDER BY id ASC').all();
console.time('Thời gian Replay');
for (const row of events) {
Projector.handle({
streamId: row.stream_id,
type: row.event_type,
payload: JSON.parse(row.payload),
version: row.version
});
}
console.timeEnd('Thời gian Replay');
console.log(`Đã replay thành công ${events.length} event vào các projection view.`);Chạy script để tái tạo lại toàn bộ read model từ đầu:
node replay.jsTruy vấn lại /api/accounts/:id giờ đây sẽ trả về trạng thái hoàn toàn đồng nhất. Trong các bài kiểm tra hiệu năng, SQLite xử lý khoảng 45.000 event được replay mỗi giây trên ổ cứng NVMe tiêu chuẩn.
3. Các Metric Giám sát trong Production
Các chỉ số sử dụng CPU và bộ nhớ tiêu chuẩn sẽ không thể phát hiện ra các nút thắt cổ chai của Event Sourcing. Thay vào đó, hãy giám sát ba chỉ số vận hành sau:
- Độ trễ Projection (Projection Lag): Theo dõi khoảng chênh lệch giữa timestamp
created_attrongevent_storevà timestamp ghi trên các read view của bạn. Trong các hệ thống bất đồng bộ chạy trên nền Kafka hoặc RabbitMQ, hãy kích hoạt cảnh báo bất cứ khi nào độ trễ của consumer vượt quá 500ms hoặc 2.500 message chưa được xử lý. - Tính Liên tục của Stream Version: Xác minh rằng các version tăng đều đặn đúng +1. Bất kỳ bước nhảy thứ tự nào (chẳng hạn từ version 2 nhảy thẳng lên version 4) đều cảnh báo một message đã bị rớt hoặc một partition consumer xử lý sai thứ tự.
- Tỷ lệ Xung đột Khóa Lạc quan (Optimistic Lock Collision Rate): Theo dõi các đợt tăng đột biến của
ConcurrencyConflictException. Mặc dù các xung đột thỉnh thoảng xảy ra là điều bình thường khi chịu tải, nhưng tỷ lệ vượt quá 2% tổng lưu lượng ghi là dấu hiệu của hiện tượng hot-partitioning. Khắc phục vấn đề này bằng cơ chế exponential backoff ở phía client hoặc định tuyến cố định (sticky routing) theo stream ID tại bộ cân bằng tải (load balancer).
