Posted inDevelopment

Hướng dẫn sử dụng BullMQ và Redis quản lý Job Queue trong Node.js: Concurrency, Delayed Jobs và Retry tự động

Sự cố nghẽn server lúc 2 giờ sáng

Đúng 2 giờ sáng, chuông cảnh báo PagerDuty réo liên hồi. Trên màn hình Datadog, API gateway trả về lỗi 504 Gateway Timeout đỏ rực. CPU của 4 instance EC2 chạm ngưỡng 100%, RAM cạn dần và process Node.js crash liên tục vì lỗi Out of Memory (OOM).

Nguyên nhân bắt nguồn từ một cronjob email marketing chạy lúc nửa đêm. Hệ thống nhận lệnh và xử lý ngay trong request handler của Express. Backend vừa render HTML template, vừa nén file PDF hóa đơn 2MB, vừa gọi API gửi mail cho hơn 20.000 khách hàng cùng lúc.

Nhồi nhét tác vụ nặng vào request-response cycle là sai lầm kinh điển. Dù bạn có bọc trong Promise.all(), Node.js vẫn chỉ chạy trên một Single Thread Event Loop. Khi có hàng trăm tác vụ ngốn CPU hoặc chờ I/O bên ngoài, Event Loop sẽ nghẽn cứng. Thậm chí endpoint /healthz nhẹ nhất cũng bị timeout theo. Để giải quyết triệt để, bạn cần tách rời luồng tiếp nhận request và luồng thực thi nền bằng kiến trúc Job Queue với Redis và BullMQ.

Mô hình Job Queue hoạt động thế nào?

Kiến trúc hàng đợi hoạt động dựa trên mô hình Producer – Consumer quen thuộc:

  • Producer (Web API): Nhận request từ client, đóng gói dữ liệu công việc (job payload) rồi đẩy vào Redis. Ngay sau đó, API trả về HTTP 202 Accepted cho người dùng. Thao tác này chỉ mất khoảng 2-5ms thay vì bắt người dùng chờ 10-30 giây.
  • Queue Broker (Redis): Đóng vai trò vùng đệm lưu trữ hàng đợi trong RAM với tốc độ đọc ghi sub-millisecond.
  • Worker / Consumer: Một tiến trình Node.js riêng biệt, chạy nền độc lập với web server. Worker liên tục kéo job từ Redis về xử lý tuần tự hoặc song song theo quota tài nguyên bạn cấp phép.

BullMQ là thư viện quản lý queue hàng đầu hiện nay cho Node.js, được viết hoàn toàn bằng TypeScript. Khác với Bull cũ dùng Redis pub/sub và polling, BullMQ khai thác triệt để Redis Streams kết hợp cùng Lua script nguyên tử (atomic). Nhờ đó, hệ thống phân phối tác vụ chính xác, hỗ trợ concurrency, delayed jobs và retry mà không lo mất dữ liệu nếu worker đột ngột tắt nguồn.

Triển khai BullMQ và Redis từng bước

1. Chuẩn bị môi trường

Khởi chạy nhanh một container Redis 7 bằng Docker:

docker run -d --name redis-queue -p 6379:6379 redis:7-alpine

Cài đặt hai thư viện cốt lõi vào dự án:

npm install bullmq ioredis

2. Cấu hình kết nối Redis an toàn

Tạo file redis.js để tái sử dụng kết nối. Lưu ý quan trọng: bạn bắt buộc phải đặt maxRetriesPerRequest: null. BullMQ sử dụng các lệnh blocking như BRPOPLPUSH / BLMOVE để chờ job mới, nên nếu để giới hạn retry mặc định của ioredis, kết nối sẽ bị ngắt giữa chừng.

// redis.js
const IORedis = require('ioredis');

const redisConnection = new IORedis({
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
});

module.exports = redisConnection;

3. Định nghĩa Queue và đẩy Job

Trong queue.js, bạn khởi tạo hàng đợi và cấu hình chiến lược lưu trữ cũng như cơ chế thử lại (Retry Strategy):

// queue.js
const { Queue } = require('bullmq');
const redisConnection = require('./redis');

const emailQueue = new Queue('email-processing', {
  connection: redisConnection,
  defaultJobOptions: {
    // Giữ lại 1.000 job hoàn tất gần nhất để kiểm tra
    removeOnComplete: {
      count: 1000,
    },
    // Lưu 5.000 job lỗi để kỹ sư kiểm tra log sau này
    removeOnFail: {
      count: 5000,
    },
    // Thử lại tối đa 5 lần nếu gặp sự cố
    attempts: 5,
    // Giãn cách thử lại tăng dần theo hàm mũ: 3s, 6s, 12s, 24s...
    backoff: {
      type: 'exponential',
      delay: 3000,
    },
  },
});

async function addEmailJob(data) {
  return await emailQueue.add('send-welcome-email', data);
}

module.exports = { emailQueue, addEmailJob };

4. Thiết lập Delayed Jobs (Hẹn giờ chạy tác vụ)

Thay vì chạy một cronjob mỗi phút để quét bảng orders trong database, bạn có thể hẹn giờ kích hoạt job trực tiếp qua tham số delay. Đơn vị tính bằng mili-giây:

// Gửi email nhắc giỏ hàng bỏ quên sau đúng 15 phút (900.000 ms)
await emailQueue.add(
  'send-cart-reminder',
  { userId: 'user_102', cartId: 'cart_883' },
  { delay: 15 * 60 * 1000 }
);

5. Xây dựng Worker và kiểm soát Concurrency

Worker chính là nơi tiêu thụ CPU và RAM. Nếu server chỉ có 2 vCPU, bạn không nên đặt concurrency lên 50 hay 100. Giữ con số này ở mức vừa phải (từ 5 đến 10) giúp bảo vệ CPU và tránh làm tràn connection pool của PostgreSQL hay MySQL.

// worker.js
const { Worker } = require('bullmq');
const redisConnection = require('./redis');

// Giả lập gửi mail qua bên thứ ba (SendGrid, SES) có tỉ lệ lỗi 30%
async function mockSendEmail(jobData) {
  if (Math.random() < 0.3) {
    throw new Error('503 Service Unavailable: Gateway phía nhà cung cấp mail timeout!');
  }
  console.log(`[OK] Đã gửi mail thành công đến: ${jobData.email}`);
}

const emailWorker = new Worker(
  'email-processing',
  async (job) => {
    console.log(`[XỬ LÝ] Job ${job.id} - Thử lần ${job.attemptsMade + 1}`);
    await mockSendEmail(job.data);
  },
  {
    connection: redisConnection,
    concurrency: 5, // Xử lý song song tối đa 5 jobs trên tiến trình này
  }
);

emailWorker.on('completed', (job) => {
  console.log(`[HOÀN TẤT] Job ${job.id} chạy thành công.`);
});

emailWorker.on('failed', (job, err) => {
  console.error(`[THẤT BẠI] Job ${job.id} lỗi ở lần thử ${job.attemptsMade}: ${err.message}`);
});

6. Kiểm thử cơ chế tự phục hồi

Tạo file server.js bắn 20 jobs vào hàng đợi để quan sát hành vi hệ thống:

// server.js
const { addEmailJob } = require('./queue');

async function bootstrap() {
  console.log('Bắt đầu nạp 20 jobs vào hàng đợi...');
  for (let i = 1; i <= 20; i++) {
    await addEmailJob({
      email: `customer_${i}@example.com`,
      subject: 'Kích hoạt tài khoản',
    });
  }
  console.log('Producer đã đẩy xong 20 jobs chỉ trong vài mili-giây.');
}

bootstrap();

Mở terminal thứ nhất và khởi động worker:

node worker.js

Mở tiếp terminal thứ hai để nạp tải:

node server.js

Kết quả hiển thị trên terminal worker chứng minh cơ chế bảo vệ của hàng đợi:

  1. Hệ thống chỉ kéo đúng 5 job xử lý cùng một thời điểm, giữ CPU ổn định quanh mức 15-25%.
  2. Những job vấp phải lỗi timeout giả lập không bị vứt bỏ. Chúng tự động đưa vào danh sách chờ thử lại với khoảng cách thời gian tăng dần theo Exponential Backoff.
  3. Khi bên nhận mail ổn định trở lại, toàn bộ job lỗi được hoàn tất trọn vẹn mà không cần kỹ sư can thiệp thủ công.

Bài học vận hành thực chiến trên Production

Tách rời producer và worker giúp hạ tải tức thì, nhưng cũng làm tăng độ phức tạp trong quản lý trạng thái.

Đầu tiên là nguyên tắc Lũy đẳng (Idempotency). Khi worker crash giữa chừng hoặc API gửi mail bị timeout ở tầng network, BullMQ sẽ chạy lại job đó. Nếu code không kiểm tra trạng thái trước khi gửi, khách hàng có thể nhận 3-4 email trùng lặp, hoặc nghiêm trọng hơn là bị trừ tiền hai lần. Hãy luôn kiểm tra cờ trạng thái trong database hoặc dùng Idempotency Key trước khi thực hiện tác vụ chính.

Thứ hai là bài toán dung lượng RAM Redis. Trung bình một job metadata kèm payload chiếm từ 1KB đến 4KB bộ nhớ. Nếu hệ thống xử lý 200.000 job mỗi ngày mà bạn quên cấu hình removeOnComplete, Redis sẽ ngốn thêm gần 1GB RAM mỗi ngày chỉ để chứa rác lịch sử. Sau 2-3 tuần, cụm Redis chắc chắn cạn RAM và sập.

Cuối cùng, luôn tách worker và API server thành các container hoặc droplet riêng biệt. Khi worker cần render video hoặc nén file PDF nặng ngốn sạch CPU, API server vẫn tiếp nhận request của người dùng mượt mà mà không lo bị treo lây.

Tóm kết

Đưa tác vụ nặng ra khỏi luồng request chính là giải pháp nền tảng giúp triệt tiêu lỗi 504 Timeout và giữ cho Node.js luôn phản hồi nhanh. Bằng cách kết hợp BullMQ và Redis, bạn nắm trong tay quyền kiểm soát tải hệ thống qua concurrency, tự động hóa retry khi dịch vụ thứ ba chập chờn và lên lịch công việc linh hoạt mà không cần lạm dụng cronjob.

Share:
Tags: