Elasticsearch Bài học 15: Xử lý xung đột dữ liệu (Conflict) trong môi trường đa luồng (Concurrency)

Ở Bài 11, chúng ta đã làm quen với Khóa lạc quan (Optimistic Concurrency Control - OCC) qua hai tham số _seq_no và _primary_term để chống lại lỗi Over-selling (bán vượt tồn kho).

Nhưng trong môi trường thực chiến – ví dụ một đợt Mega Sale tại Hasaki hay luồng xử lý hàng vạn giao dịch quẹt thẻ AFC đổ về qua Kafka – OCC thôi là chưa đủ. Nếu 100 luồng (thread) cùng lao vào trừ tồn kho của một sản phẩm trong cùng một mili-giây, Elasticsearch sẽ cho 1 luồng thành công và ném lỗi 409 Version Conflict vào mặt 99 luồng còn lại.

Bài học này sẽ hướng dẫn cách thiết kế kiến trúc Backend để xử lý triệt để 99 luồng thất bại đó, và giải quyết bài toán đồng bộ dữ liệu chéo hệ thống.

1. Xử lý "Thundering Herd" với Retry & Exponential Backoff

Khi hàng loạt request đâm sầm vào nhau gây ra lỗi 409, ném thẳng lỗi này về phía Frontend (bắt user click lại) là một trải nghiệm tồi tệ.

Tầng 1: Elasticsearch tự retry Như đã nói ở Bài 11, bạn có thể đẩy trách nhiệm này cho Elasticsearch bằng tham số retry_on_conflict=N. Tuy nhiên, nếu số lượng conflict quá lớn, việc Elasticsearch tự retry liên tục sẽ làm nghẽn Data Node. Bạn chỉ nên set số này ở mức thấp (vd: 3 hoặc 5).

Tầng 2: Vòng lặp Exponential Backoff ở Backend Nếu Elasticsearch đã retry 3 lần vẫn thất bại và trả về 409, code Backend (Golang hoặc PHP) của bạn phải đứng ra xử lý bằng thuật toán Exponential Backoff (Lùi bước theo hàm mũ). Thay vì retry ngay lập tức (dễ gây kẹt thêm), luồng của bạn sẽ "ngủ" (sleep) một khoảng thời gian tăng dần:

• Lần thất bại 1: Sleep 50ms rồi thử lại GET -> Tính toán -> PUT.

• Lần thất bại 2: Sleep 100ms.

• Lần thất bại 3: Sleep 200ms.

Tầng 3: Đẩy vào Dead Letter Queue (DLQ) Nếu thử đến lần thứ 5 vẫn thất bại, đừng block request API nữa. Hãy bắt Exception, đẩy payload giao dịch đó vào một message broker (như RabbitMQ hoặc Beanstalkd) dưới dạng Dead Letter Queue. Một worker chạy ngầm sẽ âm thầm tiêu thụ hàng đợi này vào lúc hệ thống thấp điểm hơn để đảm bảo không rớt một giao dịch nào.

2. External Versioning: Khi Elasticsearch không phải là "Nguồn chân lý"

Trong các hệ thống phân tán chuẩn mực, RDBMS (MySQL, PostgreSQL) mới là Source of Truth. Elasticsearch chỉ đóng vai trò read-replica thông qua luồng đồng bộ Change Data Capture (Debezium + Kafka).

Vấn đề: Hệ thống mạng không bao giờ hoàn hảo. Có thể có một độ trễ khiến gói tin cập nhật bản v2 (từ Kafka) chạy đến Elasticsearch chậm hơn gói tin cập nhật bản v3. Nếu bạn dùng API Update thông thường, bản v2 đến sau sẽ vô tình đè bẹp bản v3 mới nhất. Bạn không thể dựa vào _seq_no của Elasticsearch vì hệ thống gốc (MySQL) không hề biết _seq_no là gì.

Giải pháp: version_type=external Elasticsearch cho phép bạn bỏ qua bộ đếm phiên bản nội bộ, và ép nó sử dụng phiên bản từ hệ thống khác gửi sang. Bạn có thể dùng updated_at (chuyển thành Unix timestamp) hoặc cột version trong MySQL làm chốt chặn.

Cú pháp:

JSON

PUT products/_doc/1?version=1697200000&version_type=external
{
  "stock": 100,
  "updated_at": "2023-10-13 10:00:00"
}

Luật chơi của external: Elasticsearch sẽ chỉ chấp nhận lệnh ghi nếu version bạn gửi lên LỚN HƠN (>) version hiện tại đang lưu trữ. Nếu gói tin v2 đến sau gói tin v3, số version của nó nhỏ hơn, Elasticsearch sẽ tự động Reject với lỗi 409, bảo vệ toàn vẹn dữ liệu từ MySQL.

3. Distributed Lock (Khóa phân tán): Xử lý giao dịch chéo

Có những nghiệp vụ cực kỳ phức tạp yêu cầu phải sửa nhiều Document ở nhiều Index cùng một lúc, hoặc tính toán logic liên quan đến API của đối tác thứ ba trước khi lưu vào Elasticsearch.

Vì Elasticsearch không có cơ chế Transaction (Giao dịch) như MySQL (không có BEGIN...COMMIT...ROLLBACK), bạn không thể đảm bảo tính ACID khi thao tác trên nhiều bảng.

Giải pháp: Phải kéo cơ chế khóa ra khỏi database và đặt ở tầng hạ tầng bằng Redis Distributed Lock (Redlock).

• Bước 1: Khi API Backend nhận request trừ tồn kho, thay vì lao thẳng vào Elasticsearch, code (Golang/Nodejs) sẽ xin một khóa từ Redis: SET lock:product:1 "token" EX 2 NX (Khóa sản phẩm ID 1 trong 2 giây).

• Bước 2: Nếu lấy khóa thất bại (tức là có luồng khác đang giữ), luồng hiện tại trả về lỗi hoặc retry.

• Bước 3: Nếu lấy khóa thành công, Backend thoải mái gọi GET từ Elasticsearch, tính toán, gọi API logistics kiểm tra hành trình, rồi gọi PUT cập nhật Elasticsearch.

• Bước 4: Xóa khóa trên Redis (DEL lock:product:1).

Sử dụng Redis giúp bạn biến luồng xử lý Elasticsearch từ đa luồng (Multi-thread) thành xử lý tuần tự (Sequential) đối với những Document quan trọng, loại bỏ hoàn toàn khả năng xảy ra Version Conflict từ trong trứng nước.

Kết thúc Bài 15, chúng ta chính thức khép lại Phần 3: Thao tác dữ liệu nâng cao. Bạn đã nắm vững toàn bộ vòng đời của một Document từ khi được tạo ra, băm nhỏ, phân tán, cho đến khi đối mặt với bão traffic đa luồng.

Bây giờ là lúc chuyển sang Phần 4: Mapping & Phân tích văn bản. Tại sao Elasticsearch lại tìm được từ "chống" khi user gõ "chong", và làm sao để dạy nó hiểu tiếng Việt?