Redis Streams Consumer Groups: Xử lý message tin cậy trong production

Redis Streams Consumer Groups giúp nhiều worker chia nhau xử lý một dòng sự kiện, nhưng không biến Redis thành hàng đợi exactly-once. Message đã giao nằm trong Pending Entries List cho đến khi XACK; worker chết cần được phát hiện và message phải được claim lại. Vì vậy consumer production phải idempotent, giới hạn retry, quản lý retention và quan sát pending backlog.

Bài viết dựa trên tài liệu chính thức về Redis Streams, XREADGROUP và XAUTOCLAIM. Kiểm tra cú pháp theo phiên bản Redis đang vận hành.

1. Stream, group và consumer giữ ba loại trạng thái

XADD nối entry vào stream và trả ID tăng theo thời gian. Mỗi consumer group có con trỏ last-delivered-id; mỗi consumer nhận một phần công việc; entry đã giao nhưng chưa ack nằm trong Pending Entries List (PEL).

XGROUP CREATE orders workers $ MKSTREAM
XADD orders * event_id evt-123 type order.created order_id 9821
XREADGROUP GROUP workers worker-a COUNT 20 BLOCK 5000 STREAMS orders >

Dấu > yêu cầu message mới chưa từng giao cho group. Đọc pending của chính consumer dùng ID khác, nên loop recovery và loop nhận mới cần được thiết kế rõ ràng.

2. Delivery là at-least-once

Worker có thể hoàn tất side effect rồi crash trước XACK. Message sẽ được giao lại và side effect có thể chạy lần hai. Ack trước khi xử lý lại có nguy cơ mất việc nếu worker chết. Không có thứ tự gọi nào tự tạo exactly-once giữa Redis và database/API khác.

Thiết kế đúng là ack sau khi side effect bền vững, đồng thời làm handler idempotent bằng event_id:

BEGIN;
INSERT INTO processed_events(event_id)
VALUES ('evt-123')
ON CONFLICT DO NOTHING;

-- Chỉ thực hiện thay đổi nghiệp vụ nếu insert ở trên thành công
COMMIT;

XACK orders workers 1710000000000-0

Dedup record và thay đổi nghiệp vụ nên cùng transaction database. Nếu gọi API ngoài, dùng idempotency key hoặc outbox/inbox phù hợp.

3. Vòng lặp consumer phải có backpressure

• Dùng COUNT vừa với concurrency và thời gian xử lý.
• Dùng BLOCK thay polling liên tục.
• Không đọc thêm vô hạn khi worker pool đã đầy.
• Đặt timeout cho dependency và dừng nhận mới khi shutdown.
• Hoàn tất hoặc để message pending có chủ đích; không ack mù.

Mỗi process cần consumer name duy nhất và ổn định trong vòng đời instance. Dùng cùng tên cho nhiều worker làm metric và ownership khó hiểu.

4. Recovery bằng XAUTOCLAIM

XAUTOCLAIM chuyển ownership của pending entry đã idle quá ngưỡng sang consumer khỏe, với cách quét gần giống cursor:

XAUTOCLAIM orders workers recovery-1 60000 0-0 COUNT 100

Lệnh trả cursor kế tiếp và các entry đã claim; tiếp tục cho đến khi cursor quay về 0-0. min-idle-time phải lớn hơn thời gian xử lý hợp lệ cao percentile. Nếu đặt quá thấp, recovery worker cướp message vẫn đang chạy và làm tăng duplicate.

Claim không chứng minh consumer cũ đã chết và cũng không khóa side effect bên ngoài. Idempotency vẫn bắt buộc. Quan sát delivery count để nhận diện poison message.

5. Retry có ngân sách và dead-letter flow

Redis Streams không tự cung cấp một dead-letter queue hoàn chỉnh theo policy ứng dụng. Consumer cần phân loại:

• lỗi tạm thời: giữ pending hoặc retry có backoff;
• lỗi dữ liệu vĩnh viễn: ghi entry cùng nguyên nhân sang dead-letter stream rồi ack bản gốc;
• lỗi không rõ: giới hạn delivery count, alert và quarantine.

Thao tác đưa sang dead-letter và ack không atomic giữa hai command nếu chỉ gọi tuần tự. Có thể dùng transaction/Lua phù hợp, nhưng script cũng cần versioning, timeout và cluster key-slot awareness. Luôn giữ original stream ID, event ID, lỗi và số lần thử.

6. Retention và PEL là hai vấn đề khác nhau

MAXLEN ~ hoặc XTRIM giới hạn độ dài stream. Nhưng trim entry không có nghĩa mọi group đã xử lý xong. Nếu retention ngắn hơn consumer lag, dữ liệu cần thiết có thể biến mất trong khi trạng thái pending vẫn gây khó chẩn đoán.

Chọn retention từ worst-case outage, tốc độ ingest, dung lượng Redis và yêu cầu replay. Theo dõi ID cũ nhất cần thiết của từng group. Approximate trimming giảm chi phí nhưng không đảm bảo đúng một số lượng tuyệt đối.

7. Ordering chỉ có phạm vi

Stream ID tạo một thứ tự trong stream, nhưng nhiều consumer xử lý song song sẽ hoàn tất không theo thứ tự. Nếu mọi event của một aggregate phải tuần tự, có thể partition theo key thành nhiều stream hoặc dùng khóa/sequence check ở database. Một stream duy nhất với một consumer giữ thứ tự nhưng giới hạn throughput và availability.

Không suy luận thời gian nghiệp vụ tuyệt đối chỉ từ stream ID: producer khác nhau, clock và retry có thể khiến event-time khác ingestion-time. Mang theo occurred_at và aggregate version khi cần.

8. Producer cũng cần contract

• Có event_id toàn cục và schema version.
• Không đặt payload lớn/binary trực tiếp nếu có thể lưu reference.
• Xác định field bắt buộc và cách xử lý field lạ.
• Dùng transactional outbox nếu event phải đồng bộ với database write.
• Đặt max length có chủ đích, không trim tùy tiện ở mỗi producer.

XADD thành công không chứng minh nghiệp vụ trước đó đã commit. Dual-write database rồi Redis cần outbox hoặc cơ chế nhất quán tương đương.

9. Observability tối thiểu

Alert theo tuổi message và SLO xử lý, không chỉ theo số lượng. Một queue nhỏ nhưng message quan trọng kẹt nhiều giờ vẫn là sự cố.

10. Shutdown, deploy và bảo mật

Khi nhận signal, dừng XREADGROUP mới, chờ task đang chạy trong grace period, ack task đã commit và để task chưa hoàn tất trong PEL cho recovery. Không xóa consumer ngay nếu chưa hiểu pending ownership.

Dùng ACL giới hạn key pattern và command; TLS cho kết nối; không đưa secret/PII không cần thiết vào stream. Backup và failover cần được test với delivery semantics thực tế, không chỉ kiểm tra Redis “lên lại”.

Checklist production

• Handler idempotent bằng event ID bền vững.
• Ack chỉ sau khi side effect đã commit.
• Đọc mới bằng >; recovery pending là luồng riêng.
• XAUTOCLAIM có min idle dựa trên processing latency.
• Retry có giới hạn và dead-letter có metadata điều tra.
• Retention dài hơn outage/lag dự kiến và phù hợp memory.
• Ordering requirement được thiết kế theo aggregate/partition.
• Producer có schema version và outbox khi cần.
• Dashboard có lag, PEL, oldest idle, delivery count và DLQ.
• Graceful shutdown, crash và duplicate delivery đã được test.

Kết luận

Consumer Groups giải quyết phân phối công việc và theo dõi message đang bay, nhưng độ tin cậy cuối cùng nằm ở handler. Khi chấp nhận at-least-once, dùng idempotency, recovery có ngưỡng, retry hữu hạn và retention đo được, Redis Streams trở thành nền tảng gọn cho nhiều workflow thời gian thực mà không che giấu failure mode.