Giao diện
@digiforce-nc/plugin-queue-adapter-redis
Plugin message queue qua Redis Streams — xử lý background job với Redis Streams và Consumer Groups, hỗ trợ concurrency control, retry, graceful shutdown.
Plugin này làm gì?
Nếu đã có Redis trong hệ thống, plugin này cung cấp adapter xử lý message queue mà không cần cài thêm RabbitMQ. Plugin sử dụng Redis Streams với Consumer Groups (XREADGROUP) để đảm bảo mỗi message chỉ được xử lý bởi một consumer duy nhất trong cluster.
Lưu ý kỹ thuật: Plugin sử dụng Redis Streams (
XADD/XREADGROUP), không phải BullMQ. Mỗi subscription tạo một Redis client riêng biệt.
Tính năng chính
| # | Tính năng | Mô tả |
|---|---|---|
| 1 | Consumer Groups | Mỗi message chỉ được xử lý bởi 1 consumer trong group |
| 2 | Concurrency control | Giới hạn số message xử lý đồng thời |
| 3 | Retry + max retries | Tự retry khi thất bại, dừng sau N lần |
| 4 | Timeout (AbortSignal) | Mỗi job có timeout signal |
| 5 | Graceful shutdown | Chờ job đang xử lý trước khi tắt |
| 6 | Auto-trim | Stream trim ~100,000 entries (MAXLEN approximate) |
| 7 | ACK + Delete | Message được ACK và xóa sau khi xử lý |
Biến môi trường
| Biến | Mặc định | Mô tả |
|---|---|---|
QUEUE_ADAPTER | (bắt buộc) | Phải đặt redis để kích hoạt adapter |
QUEUE_ADAPTER_REDIS_URL | (bắt buộc) | URL kết nối Redis |
QUEUE_CONNECTION_RETRY_MAX_ATTEMPTS | 20 | Số lần retry kết nối |
QUEUE_CONNECTION_RETRY_DELAY | 200 | Delay giữa các lần retry (ms) |
Kiến trúc nội bộ
| Thành phần | Vai trò |
|---|---|
RedisQueueAdapter | Implement IEventQueueAdapter — quản lý subscriptions |
publish(channel, content, options) | Gửi message vào stream (XADD, MAXLEN ~100000) |
subscribe(channel, options) | Tạo consumer group + Redis client riêng cho channel |
consume(channel) | Vòng lặp XREADGROUP BLOCK 1000 — đọc và xử lý message |
process(channel, { id, message }) | Xử lý message, retry khi lỗi, ACK + XDEL |
So sánh với RabbitMQ adapter
| Tiêu chí | Redis Streams | RabbitMQ |
|---|---|---|
| Setup | Dùng chung Redis | Cần cài riêng |
| Protocol | Redis Streams | AMQP |
| Routing | Đơn giản (channel name) | Exchange routing phức tạp |
| Reliability | Phụ thuộc Redis persistence | Cao (durable + persistent) |
| Monitoring | Redis Insight | Management UI built-in |
| Phù hợp | Đã có Redis, workload vừa | High throughput, mission-critical |
Dependencies
| Package | Vai trò |
|---|---|
@digiforce-nc/server | Server framework — IEventQueueAdapter interface |
redis (^4.6.10) | Redis client (node-redis v4) |