Xây dựng hệ thống xử lý ảnh hàng loạt bằng Node.js, BullMQ và Redis: Kiến trúc, tối ưu và triển khai thực tế

Giới thiệu

Trong môi trường thương mại điện tử, mạng xã hội hoặc các nền tảng nội dung, việc tải lên và xử lý hàng nghìn ảnh mỗi ngày là một thách thức lớn. Nếu không có kiến trúc phù hợp, server sẽ bị nghẽn, thời gian phản hồi kéo dài và trải nghiệm người dùng giảm sút. Bài viết này sẽ hướng dẫn chi tiết cách xây dựng một pipeline xử lý ảnh hàng loạt bằng Node.js, BullMQ làm hàng đợi công việc, và Redis để cache kết quả. Toàn bộ giải pháp sẽ được container hoá bằng Docker, cho phép mở rộng dễ dàng trên môi trường VPS hoặc cloud.

Kiến trúc tổng quan

Yêu cầu và mục tiêu

Hệ thống cần đáp ứng các tiêu chí sau:

• Khả năng nhận multipart/form-data từ client và lưu tạm thời.
• Tiến hành các bước xử lý ảnh: resize, watermark, chuyển đổi định dạng.
• Thực hiện các công việc này một cách bất đồng bộ, không làm block request thread.
• Lưu trữ kết quả cuối cùng trên S3 (hoặc bất kỳ storage nào) và trả về URL cho client.
• Sử dụng cache để tránh xử lý lại các ảnh đã được xử lý trước đó.
• Quản lý retry, timeout và logging chi tiết.

Thành phần chính

• API Gateway: Ứng dụng Express nhận yêu cầu upload và trả về ID công việc.
• Queue Server: BullMQ dựa trên Redis, chịu trách nhiệm lưu trữ và phân phối các job.
• Worker Process: Node.js process lắng nghe queue, thực hiện các bước xử lý ảnh.
• Cache Layer: Redis dùng để lưu hash của ảnh gốc và URL kết quả, giảm tải xử lý lại.
• Storage Service: Đối tượng lưu trữ (S3, MinIO) chứa ảnh đã xử lý.

Cài đặt môi trường

Docker Compose

Docker Compose giúp khởi tạo đồng thời các container Redis, MinIO (đại diện S3) và ứng dụng Node.js. Dưới đây là file docker-compose.yml mẫu:

version: '3.8'
services:
  redis:
    image: redis:7-alpine
    container_name: redis
    ports:
      - "6379:6379"
    restart: unless-stopped

  minio:
    image: minio/minio:latest
    container_name: minio
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: password123
    command: server /data
    ports:
      - "9000:9000"
    restart: unless-stopped

  api:
    build: ./api
    container_name: node_api
    environment:
      REDIS_URL: redis://redis:6379
      MINIO_ENDPOINT: minio:9000
      MINIO_ACCESS_KEY: admin
      MINIO_SECRET_KEY: password123
    ports:
      - "3000:3000"
    depends_on:
      - redis
      - minio
    restart: unless-stopped

Thư mục ./api chứa Dockerfile cho ứng dụng Node.js:

FROM node:18-alpine
WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY . .
EXPOSE 3000
CMD ["node", "src/index.js"]

Triển khai hàng đợi với BullMQ

Khởi tạo queue

Trong dự án Node.js, chúng ta tạo một module queue.js để cấu hình BullMQ. BullMQ yêu cầu một kết nối Redis, vì vậy chúng ta sẽ dùng ioredis làm client.

const { Queue, Worker, QueueScheduler } = require('bullmq');
const IORedis = require('ioredis');

const connection = new IORedis(process.env.REDIS_URL);

// QueueScheduler đảm bảo các job bị trì hoãn được xử lý đúng thời gian
new QueueScheduler('image-processing', { connection });

const imageQueue = new Queue('image-processing', { connection });

module.exports = { imageQueue, connection };

API nhận file và đưa vào queue

Endpoint POST /upload nhận file, tính hash SHA256 để kiểm tra cache, sau đó tạo job nếu chưa tồn tại.

const express = require('express');
const multer = require('multer');
const crypto = require('crypto');
const { imageQueue } = require('./queue');
const redis = require('ioredis');
const cache = new redis(process.env.REDIS_URL);

const router = express.Router();
const upload = multer({ dest: 'tmp/' });

router.post('/upload', upload.single('image'), async (req, res) => {
  const filePath = req.file.path;
  const fileBuffer = require('fs').readFileSync(filePath);
  const hash = crypto.createHash('sha256').update(fileBuffer).digest('hex');

  // Kiểm tra cache
  const cachedUrl = await cache.get(`image:${hash}`);
  if (cachedUrl) {
    return res.json({ url: cachedUrl, cached: true });
  }

  // Đưa job vào queue
  const job = await imageQueue.add('process', { filePath, hash }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 5000 },
    removeOnComplete: true,
    removeOnFail: false,
  });

  res.json({ jobId: job.id, cached: false });
});

module.exports = router;

Worker xử lý ảnh

Worker sẽ lắng nghe queue image-processing, thực hiện các bước: đọc file, resize bằng sharp, upload lên MinIO và lưu cache.

const { Worker } = require('bullmq');
const { connection } = require('./queue');
const sharp = require('sharp');
const Minio = require('minio');
const redis = require('ioredis');
const cache = new redis(process.env.REDIS_URL);

const minioClient = new Minio.Client({
  endPoint: process.env.MINIO_ENDPOINT,
  port: 9000,
  useSSL: false,
  accessKey: process.env.MINIO_ACCESS_KEY,
  secretKey: process.env.MINIO_SECRET_KEY,
});

const worker = new Worker('image-processing', async job => {
  const { filePath, hash } = job.data;
  // Resize to 800x800, maintain aspect ratio
  const buffer = await sharp(filePath)
    .resize(800, 800, { fit: 'inside' })
    .toFormat('jpeg')
    .toBuffer();

  const objectName = `${hash}.jpg`;
  await minioClient.putObject('images', objectName, buffer, {
    'Content-Type': 'image/jpeg',
  });

  const url = `${process.env.MINIO_ENDPOINT}/images/${objectName}`;
  // Lưu vào cache trong 24h
  await cache.setex(`image:${hash}`, 86400, url);

  // Xóa file tạm
  require('fs').unlinkSync(filePath);
  return { url };
}, { connection, concurrency: 4 });

worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed:`, err);
});

module.exports = worker;

Tối ưu hiệu năng với Redis Cache

Lưu trữ kết quả tạm thời

Cache giúp tránh việc xử lý lại các ảnh đã tồn tại. Chúng ta sử dụng SETEX để đặt thời gian sống (TTL) cho mỗi khóa. Khi TTL hết, ảnh sẽ được xử lý lại nếu có yêu cầu mới, giúp giảm tải bộ nhớ.

// Đặt cache trong worker
await cache.setex(`image:${hash}`, 86400, url); // 24h

// Kiểm tra cache trong API
const cachedUrl = await cache.get(`image:${hash}`);
if (cachedUrl) {
  return res.json({ url: cachedUrl, cached: true });
}

Phân phối tải với multiple workers

BullMQ hỗ trợ thiết lập concurrency khi tạo Worker. Trong ví dụ trên, chúng ta đặt concurrency: 4, nghĩa là 4 tiến trình sẽ đồng thời xử lý các job. Đối với môi trường production, chúng ta có thể chạy nhiều container worker và để Redis làm trung tâm phối hợp.

Quản lý lỗi và retry

Trong môi trường thực tế, các lỗi như mất kết nối tới MinIO, lỗi đọc file hoặc timeout đều có thể xảy ra. BullMQ cung cấp cơ chế attempts và backoff để tự động retry. Ngoài ra, chúng ta nên ghi log chi tiết và đưa các job thất bại vào một queue phụ để phân tích.

const job = await imageQueue.add('process', { filePath, hash }, {
  attempts: 5,
  backoff: { type: 'fixed', delay: 10000 }, // 10s giữa các lần retry
});

worker.on('failed', (job, err) => {
  // Gửi thông tin tới hệ thống monitoring (ví dụ: Sentry)
  console.error(`Job ${job.id} failed after ${job.attemptsMade} attempts`);
});

Kiểm thử và giám sát

Unit test với Jest

Chúng ta có thể viết test cho hàm tính hash và hàm tạo job. Dưới đây là một ví dụ ngắn:

const crypto = require('crypto');

describe('Hash function', () => {
  test('should generate correct SHA256 hash', () => {
    const data = Buffer.from('test');
    const hash = crypto.createHash('sha256').update(data).digest('hex');
    expect(hash).toBe('9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08');
  });
});

Giám sát queue

BullMQ cung cấp UI Dashboard (bull-board) để quan sát trạng thái các job. Chúng ta có thể mount dashboard vào Express như sau:

const { createBullBoard } = require('bull-board');
const { ExpressAdapter } = require('bull-board/express');
const { Queue } = require('bullmq');

const serverAdapter = new ExpressAdapter();
const { addQueue, removeQueue, setQueues, replaceQueues } = createBullBoard({
  queues: [new Queue('image-processing', { connection })],
  serverAdapter,
});

app.use('/admin/queues', serverAdapter.getRouter());

Dashboard cho phép chúng ta xem số job đang chờ, đang chạy, đã hoàn thành và lỗi.

Triển khai thực tế và mở rộng

Khi hệ thống đã ổn định trên môi trường local, chúng ta có thể triển khai lên VPS hoặc Kubernetes. Các bước chính:

1. Đẩy image Docker lên registry nội bộ.
2. Tạo file docker-compose.prod.yml với cấu hình mạng riêng và biến môi trường bảo mật.
3. Sử dụng docker stack deploy (trên Swarm) hoặc Helm chart (trên K8s) để khởi chạy các service.
4. Thiết lập autoscaling cho worker dựa trên độ dài queue (số job pending).
5. Giám sát Redis, MinIO và Node.js bằng Prometheus + Grafana.

Với kiến trúc này, việc mở rộng chỉ cần tăng số replica của worker; Redis sẽ tự động cân bằng tải. Nếu cần xử lý các loại file khác (video, PDF), chúng ta chỉ cần tạo queue mới và viết worker tương ứng.

Kết luận

Việc xây dựng một pipeline xử lý ảnh hàng loạt bằng Node.js, BullMQ và Redis không chỉ giúp giảm thời gian phản hồi mà còn mang lại khả năng mở rộng linh hoạt, đáp ứng nhu cầu tăng trưởng dữ liệu. Khi đã nắm vững các thành phần trên, bạn có thể áp dụng tương tự cho các công việc batch khác như gửi email, tạo báo cáo hoặc xử lý dữ liệu thời gian thực. Để nâng cao kiến thức và thực hành sâu hơn, Tham khảo khóa học "Lập trình Back-End với NodeJS Express" tại đây.