Posted inProgramming

Xây dựng Distributed Task Queue với BullMQ, Redis và TypeScript

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

So sánh các giải pháp xử lý tác vụ nền (Background Processing) trong Node.js

Khi một API endpoint phải xử lý các tác vụ nặng—chẳng hạn như tạo file PDF 50MB, chuyển mã (transcode) video 4K hoặc đồng bộ hàng nghìn bản ghi với webhook bên ngoài—thời gian phản hồi HTTP sẽ nhanh chóng bị suy giảm. Tách các tác vụ này ra khỏi chu trình request-response là yêu cầu tối quan trọng. Trong các hệ thống Node.js production, các đội ngũ phát triển thường lựa chọn một trong ba mô hình chính sau:

  • In-Memory Micro-Queues (ví dụ: fastq, EventEmitter): Triển khai nhanh nhất và không phụ thuộc vào dịch vụ bên ngoài. Tuy nhiên, nếu tiến trình Node.js khởi động lại khi triển khai mã nguồn mới, toàn bộ job đang chờ trong bộ nhớ RAM sẽ biến mất. Mô hình này cũng không thể phân phối công việc trên nhiều server instance.
  • Message Broker cấp doanh nghiệp (ví dụ: Apache Kafka, RabbitMQ): Rất mạnh mẽ trong việc xử lý luồng hàng triệu sự kiện mỗi giây và quản lý cấu trúc topology tin nhắn phức tạp. Nhược điểm là chi phí vận hành lớn, đòi hỏi nhiều thời gian tìm hiểu, và cấu hình phức tạp chỉ để hẹn giờ tác vụ đơn giản hoặc thực hiện cơ chế thử lại (delayed retry).
  • Hàng đợi phân tán trên nền Redis (BullMQ): Sự lựa chọn lý tưởng cho các microservice TypeScript và Node.js hiện đại. BullMQ tận dụng Redis streams, sorted sets và các Lua script nguyên tử (atomic) để xử lý việc lập lịch, thử lại và giới hạn tần suất (rate limiting) trên các worker phân tán mà không tạo ra gánh nặng lớn về hạ tầng.

Ưu và nhược điểm của kiến trúc BullMQ + Redis

Trước khi chuẩn hóa hệ thống với BullMQ, hãy đánh giá mức độ tương thích của kiến trúc này với các ràng buộc hạ tầng của bạn:

Ưu điểm

  • Hỗ trợ sẵn Delayed & Cron Jobs: Lên lịch chạy tác vụ sau 15 phút hoặc lặp lại hàng đêm vào lúc nửa đêm theo giờ UTC mà không cần duy trì các cron daemon bên ngoài.
  • Chuyển đổi trạng thái nguyên tử (Atomic State Transitions): BullMQ thực thi các Lua script trực tiếp bên trong Redis. Điều này đảm bảo trạng thái job chuyển đổi từ waiting sang active, completed hoặc failed một cách an toàn, loại bỏ hoàn toàn hiện tượng tranh chấp dữ liệu (race conditions).
  • Định kiểu TypeScript nghiêm ngặt: Hỗ trợ Generics cho dữ liệu đầu vào và kết quả trả về của job, giúp phát hiện sớm sự sai lệch schema giữa producer và consumer ngay từ thời điểm biên dịch.
  • Kiểm soát Concurrency & Rate Limiting chi tiết: Giới hạn số lượng tác vụ xử lý đồng thời trên mỗi worker và điều tiết tổng thông lượng (ví dụ: 50 request/giây trên toàn bộ 10 pod) nhằm tránh vượt quá giới hạn API của bên thứ ba.

Nhược điểm & Đánh đổi

  • Chiếm dụng bộ nhớ Redis: Payload và metadata thực thi của job được lưu trữ trực tiếp trên RAM. Việc xử lý 500.000 job mỗi ngày mà không có chính sách dọn dẹp nghiêm ngặt có thể làm cạn kiệt bộ nhớ Redis chỉ sau vài ngày.
  • Chi phí kết nối chuyên dụng: Các worker của BullMQ sử dụng các lệnh blocking của Redis để nhận job ngay lập tức. Mỗi hàng đợi và worker instance đều duy trì nhiều kết nối TCP riêng biệt.

Kiến trúc đề xuất và thiết lập cho môi trường Production

Một hệ thống xử lý tác vụ đáng tin cậy chia công việc thành hai vai trò riêng biệt: Producers (API server đẩy tác vụ vào hàng đợi) và Consumers (tiến trình worker chuyên dụng nhận và thực thi tác vụ). Cả hai giao tiếp với nhau thông qua một cụm Redis dùng chung.

Dưới đây là mô hình chuẩn cho một pipeline production:

  1. Primary Queue (Hàng đợi chính): Nhận các job tiêu chuẩn với cấu hình backoff hàm mũ (ví dụ: độ trễ ban đầu 2 giây và tăng gấp đôi sau mỗi lần thử lại).
  2. Dead-Letter Queue (DLQ): Lưu giữ các payload đã thất bại sau toàn bộ số lần thử lại, cho phép kỹ sư kiểm tra stack trace và kích hoạt chạy lại job mà không làm tắc nghẽn luồng xử lý thông thường.
  3. Cô lập kết nối: Cấu hình ioredis với tùy chọn maxRetriesPerRequest: null theo yêu cầu bắt buộc của BullMQ để hỗ trợ các lệnh blocking kéo dài.

Hướng dẫn triển khai

1. Khởi tạo dự án và cấu hình kết nối

Khởi tạo dự án TypeScript và cài đặt các dependency cần thiết:

npm install bullmq ioredis dotenv
npm install -D typescript @types/node tsx

Cấu hình kết nối Redis dùng chung tại src/config/redis.ts:

import { ConnectionOptions } from 'bullmq';

export const redisConnection: ConnectionOptions = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  password: process.env.REDIS_PASSWORD || undefined,
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
};

2. Định nghĩa kiểu dữ liệu cho Job và khởi tạo Producer

Định nghĩa các interface payload rõ ràng trong src/types/queue.ts để đảm bảo an toàn hợp đồng dữ liệu:

export interface EmailJobData {
  to: string;
  subject: string;
  templateId: string;
  variables: Record<string, string>;
}

export interface EmailJobResult {
  messageId: string;
  sentAt: string;
}

export const EMAIL_QUEUE_NAME = 'email-delivery';
export const EMAIL_DLQ_NAME = 'email-delivery-dlq';

Khởi tạo producer cho hàng đợi trong src/queues/emailQueue.ts:

import { Queue } from 'bullmq';
import { redisConnection } from '../config/redis';
import { EMAIL_QUEUE_NAME, EmailJobData, EmailJobResult } from '../types/queue';

export const emailQueue = new Queue<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 5,
      backoff: {
        type: 'exponential',
        delay: 2000, // Thử lại sau 2s, 4s, 8s, 16s, 32s
      },
      removeOnComplete: {
        age: 3600, // Giữ metadata của job hoàn thành trong 1 giờ
        count: 1000,
      },
      removeOnFail: false, // Giữ lại các job thất bại để kiểm tra
    },
  }
);

export async function scheduleEmail(data: EmailJobData, delayMs: number = 0) {
  const job = await emailQueue.add('send-transactional-email', data, {
    delay: delayMs,
  });
  return job.id;
}

3. Xây dựng Worker và định tuyến Dead-Letter Queue

Khi các API bên ngoài gặp sự cố mạng tạm thời, BullMQ sẽ tự động thử lại tác vụ bằng cơ chế exponential backoff. Nếu cả 5 lần thử đều thất bại, hệ thống sẽ ghi lại ngữ cảnh lỗi và đẩy trực tiếp vào Dead-Letter Queue để theo dõi và cảnh báo.

Tạo consumer worker trong src/workers/emailWorker.ts:

import { Worker, Queue, Job } from 'bullmq';
import { redisConnection } from '../config/redis';
import {
  EMAIL_QUEUE_NAME,
  EMAIL_DLQ_NAME,
  EmailJobData,
  EmailJobResult,
} from '../types/queue';

const deadLetterQueue = new Queue(EMAIL_DLQ_NAME, { connection: redisConnection });

export const emailWorker = new Worker<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  async (job: Job<EmailJobData, EmailJobResult>) => {
    // Giả lập lỗi API phía downstream
    if (job.data.to.includes('fail')) {
      throw new Error(`Nhà cung cấp SMTP phía upstream bị quá thời gian chờ (timeout) cho ${job.data.to}`);
    }

    return {
      messageId: `msg_${job.id}_${Date.now()}`,
      sentAt: new Date().toISOString(),
    };
  },
  {
    connection: redisConnection,
    concurrency: 10, // Xử lý đồng thời 10 job trên mỗi worker instance
    limiter: {
      max: 100,
      duration: 1000, // Giới hạn thông lượng tối đa 100 job mỗi giây cho mỗi worker
    },
  }
);

emailWorker.on('completed', (job: Job) => {
  console.log(`[Job ${job.id}] Đã gửi thành công tới ${job.data.to}`);
});

emailWorker.on('failed', async (job: Job | undefined, error: Error) => {
  if (!job) return;

  console.warn(`[Job ${job.id}] Lần thử ${job.attemptsMade}/${job.opts.attempts} thất bại: ${error.message}`);

  // Chuyển hướng sang DLQ khi đã hết số lần thử lại
  if (job.attemptsMade >= (job.opts.attempts || 1)) {
    console.error(`[Job ${job.id}] Đã hết lượt thử lại. Đang chuyển hướng sang DLQ: ${EMAIL_DLQ_NAME}`);
    await deadLetterQueue.add('failed-email-poison-pill', {
      originalJobId: job.id,
      originalData: job.data,
      failedReason: error.message,
      failedAt: new Date().toISOString(),
      stacktrace: error.stack,
    });
  }
});

4. Graceful Shutdown & Xử lý tín hiệu hệ thống

Dừng worker đột ngột khi đang xử lý tác vụ có thể khiến job bị mồ côi (orphaned) hoặc làm sai lệch dữ liệu. Luôn lắng nghe và bắt các tín hiệu kết thúc từ hệ điều hành (SIGTERM, SIGINT) tại file khởi chạy chính (src/index.ts) để các job đang chạy có đủ thời gian hoàn thành:

import { emailWorker } from './workers/emailWorker';
import { emailQueue } from './queues/emailQueue';

async function shutdown(signal: string) {
  console.log(`Nhận tín hiệu ${signal}. Đang hoàn tất các task đang chạy và đóng kết nối...`);
  
  // Hoàn tất các job đang chạy và từ chối task mới
  await emailWorker.close();
  await emailQueue.close();
  
  console.log('Đã hoàn tất tắt task queue. Đang thoát an toàn.');
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));

Thực tiễn tốt nhất (Best Practices) cho hệ thống tải cao trong Production

  • Giữ Payload luôn gọn nhẹ: Chỉ truyền ID hoặc URL từ dịch vụ lưu trữ đối tượng (ví dụ: đường dẫn S3) thay vì nhúng trực tiếp buffer file 10MB vào dữ liệu của job.
  • Đảm bảo xử lý có tính Idempotent: Lỗi mạng tạm thời sẽ kích hoạt cơ chế thử lại, đồng nghĩa một job có thể được thực thi nhiều lần. Hãy sử dụng transaction ID của cơ sở dữ liệu hoặc các thao tác UPSERT để tránh trừ tiền hay gửi email trùng lặp.
  • Tự động dọn dẹp bộ nhớ Redis: Luôn cấu hình giới hạn cho removeOnComplete và removeOnFail. Thiết lập cảnh báo cho cụm cluster bất cứ khi nào dung lượng RAM của Redis vượt quá 75%.
  • Giám sát độ sâu hàng đợi (Queue Depth) và độ trễ: Theo dõi số lượng job đang chờ và đang chạy thông qua queue.getJobCounts() trên Prometheus hoặc Datadog. Tự động scale các worker pod khi thời gian chờ vượt quá 5 giây.
Share:
Tags: