Queues (Hàng đợi)

Queue (hàng đợi) là một design pattern mạnh mẽ giúp bạn giải quyết các bài toán phổ biến về khả năng mở rộng (scaling) và hiệu năng của ứng dụng. Một vài ví dụ về vấn đề mà Queue có thể giúp bạn giải quyết:

• San đều các đợt cao điểm khi xử lý. Ví dụ, nếu người dùng có thể khởi tạo các tác vụ tốn nhiều tài nguyên vào bất kỳ thời điểm nào, bạn có thể đưa các tác vụ này vào một queue thay vì xử lý đồng bộ (synchronously) ngay lập tức. Sau đó bạn có thể để các worker process lấy tác vụ từ queue ra xử lý một cách có kiểm soát. Bạn có thể dễ dàng thêm các Queue consumer mới để mở rộng khả năng xử lý tác vụ ở back-end khi ứng dụng phát triển.
• Chia nhỏ các tác vụ monolithic (đơn khối) mà nếu không sẽ làm nghẽn Node.js event loop. Ví dụ, nếu một request của người dùng đòi hỏi công việc nặng về CPU như transcode âm thanh, bạn có thể giao tác vụ này cho các process khác, giải phóng process phục vụ người dùng để nó luôn phản hồi nhanh.
• Cung cấp một kênh giao tiếp đáng tin cậy giữa các service khác nhau. Ví dụ, bạn có thể đưa các tác vụ (job) vào queue ở một process hoặc service, rồi tiêu thụ (consume) chúng ở một process/service khác. Bạn có thể được thông báo (bằng cách lắng nghe các sự kiện trạng thái) khi hoàn thành, lỗi, hoặc có các thay đổi trạng thái khác trong vòng đời của job, từ bất kỳ process hay service nào. Khi Queue producer hoặc consumer gặp sự cố, trạng thái của chúng vẫn được lưu giữ và việc xử lý tác vụ có thể tự động khởi động lại khi node được khởi động lại.

Nest cung cấp package @nestjs/bullmq để tích hợp với BullMQ, và package @nestjs/bull để tích hợp với Bull. Cả hai package đều là lớp abstraction/wrapper trên các thư viện tương ứng, do cùng một team phát triển. Bull hiện đang ở chế độ bảo trì (maintenance mode), team chỉ tập trung sửa lỗi, trong khi BullMQ đang được phát triển tích cực, với cách triển khai TypeScript hiện đại và một số tính năng khác biệt. Nếu Bull vẫn đáp ứng đủ nhu cầu của bạn, đây vẫn là một lựa chọn đáng tin cậy và đã được kiểm chứng qua thực tế. Các package của Nest giúp bạn tích hợp cả BullMQ lẫn Bull Queue vào ứng dụng Nest một cách dễ dàng.

Cả BullMQ và Bull đều dùng Redis để lưu trữ (persist) dữ liệu job, vì vậy bạn cần cài Redis trên hệ thống. Vì dựa trên Redis, kiến trúc Queue của bạn có thể hoàn toàn phân tán (distributed) và độc lập nền tảng (platform-independent). Ví dụ, bạn có thể có một số Queue producer, consumer và listener chạy trong Nest trên một (hoặc nhiều) node, còn các producer, consumer và listener khác chạy trên các nền tảng Node.js khác trên các node mạng khác.

Chương này nói về hai package @nestjs/bullmq và @nestjs/bull. Chúng tôi cũng khuyến khích bạn đọc thêm tài liệu của BullMQ và Bull để hiểu rõ hơn về bối cảnh cũng như các chi tiết triển khai cụ thể.

Cài đặt BullMQ

Để bắt đầu dùng BullMQ, trước tiên hãy cài các dependency cần thiết.

$ npm install --save @nestjs/bullmq bullmq

Sau khi cài đặt xong, chúng ta có thể import BullModule vào AppModule gốc.

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: 'localhost',
        port: 6379,
      },
    }),
  ],
})
export class AppModule {}

Phương thức forRoot() dùng để đăng ký một object cấu hình cho package bullmq, được tất cả các queue đăng ký trong ứng dụng sử dụng (trừ khi được chỉ định khác). Dưới đây là một số thuộc tính tiêu biểu trong object cấu hình:

• connection: ConnectionOptions - Các tùy chọn để cấu hình kết nối Redis. Xem Connections để biết thêm. Không bắt buộc.
• prefix: string - Tiền tố cho tất cả các key của queue. Không bắt buộc.
• defaultJobOptions: JobOpts - Các tùy chọn để cấu hình mặc định cho job mới. Xem JobOpts để biết thêm. Không bắt buộc.
• settings: AdvancedSettings - Các cài đặt nâng cao cho Queue. Thông thường không nên thay đổi các cài đặt này. Xem AdvancedSettings để biết thêm. Không bắt buộc.
• extraOptions - Các tùy chọn thêm cho việc khởi tạo module. Xem Manual Registration

Tất cả các tùy chọn đều không bắt buộc, giúp bạn kiểm soát chi tiết hành vi của queue. Các tùy chọn này được truyền thẳng vào constructor của Queue trong BullMQ. Đọc thêm về các tùy chọn này và các tùy chọn khác tại đây.

Để đăng ký một queue, import dynamic module BullModule.registerQueue() như sau:

BullModule.registerQueue({
  name: 'audio',
});

Phương thức registerQueue() dùng để khởi tạo và/hoặc đăng ký queue. Các queue được chia sẻ giữa các module và process kết nối tới cùng một database Redis với cùng thông tin đăng nhập (credentials). Mỗi queue là duy nhất theo thuộc tính name của nó. Tên queue được dùng vừa làm injection token (để inject queue vào controller/provider), vừa làm tham số cho decorator để gắn kết các class consumer và listener với queue.

Bạn cũng có thể ghi đè một số tùy chọn đã cấu hình sẵn cho một queue cụ thể, như sau:

BullModule.registerQueue({
  name: 'audio',
  connection: {
    port: 6380,
  },
});

BullMQ cũng hỗ trợ quan hệ cha - con (parent - child) giữa các job. Tính năng này cho phép tạo ra các flow, trong đó job là các node của những cây (tree) có độ sâu tùy ý. Đọc thêm về chúng tại đây.

Để thêm một flow, bạn có thể làm như sau:

BullModule.registerFlowProducer({
  name: 'flowProducerName',
});

Vì job được lưu trữ trong Redis, mỗi lần một queue có tên cụ thể được khởi tạo (ví dụ, khi ứng dụng khởi động/khởi động lại), nó sẽ cố gắng xử lý các job cũ còn sót lại từ phiên làm việc chưa hoàn tất trước đó.

Mỗi queue có thể có một hoặc nhiều producer, consumer và listener. Consumer lấy job từ queue theo một thứ tự cụ thể: FIFO (mặc định), LIFO, hoặc theo mức độ ưu tiên (priority). Cách kiểm soát thứ tự xử lý queue được bàn ở đây.

Cấu hình có tên (Named configurations)

Nếu các queue của bạn kết nối tới nhiều instance Redis khác nhau, bạn có thể dùng kỹ thuật gọi là named configurations (cấu hình có tên). Tính năng này cho phép bạn đăng ký nhiều cấu hình dưới các key được chỉ định, sau đó bạn có thể tham chiếu tới chúng trong các tùy chọn của queue.

Ví dụ, giả sử bạn có thêm một instance Redis khác (ngoài instance mặc định) được một vài queue trong ứng dụng sử dụng, bạn có thể đăng ký cấu hình cho nó như sau:

BullModule.forRoot('alternative-config', {
  connection: {
    port: 6381,
  },
});

Trong ví dụ trên, 'alternative-config' chỉ đơn giản là một key cấu hình (có thể là bất kỳ chuỗi tùy ý nào).

Với thiết lập này, giờ đây bạn có thể trỏ tới cấu hình này trong object tùy chọn của registerQueue():

BullModule.registerQueue({
  configKey: 'alternative-config',
  name: 'video',
});

Producers (Bên tạo job)

Job producer là bên đưa job vào queue. Producer thường là các service của ứng dụng (Nest provider). Để thêm job vào queue, trước tiên hãy inject queue vào service như sau:

import { Injectable } from '@nestjs/common';
import { Queue } from 'bullmq';
import { InjectQueue } from '@nestjs/bullmq';

@Injectable()
export class AudioService {
  constructor(@InjectQueue('audio') private audioQueue: Queue) {}
}

Bây giờ, thêm một job bằng cách gọi phương thức add() của queue, truyền vào một object job do bạn tự định nghĩa. Job được biểu diễn dưới dạng object JavaScript có thể serialize (vì đó là cách chúng được lưu trong database Redis). Hình dạng (shape) của job bạn truyền vào là tùy ý; hãy dùng nó để biểu diễn ngữ nghĩa của object job. Bạn cũng cần đặt tên cho nó. Điều này cho phép bạn tạo các consumer chuyên biệt, chỉ xử lý các job có tên nhất định.

const job = await this.audioQueue.add('transcode', {
  foo: 'bar',
});

Job options (Tùy chọn job)

Job có thể có thêm các tùy chọn đi kèm. Truyền một object tùy chọn sau tham số job trong phương thức Queue.add(). Một số thuộc tính của job options gồm:

• priority: number - Giá trị độ ưu tiên (không bắt buộc). Nằm trong khoảng từ 1 (ưu tiên cao nhất) đến MAX_INT (ưu tiên thấp nhất). Lưu ý rằng việc dùng priority có ảnh hưởng nhẹ tới hiệu năng, nên hãy cân nhắc khi sử dụng.
• delay: number - Khoảng thời gian (mili-giây) cần chờ trước khi job này có thể được xử lý. Lưu ý để có độ trễ chính xác, cả server và client cần đồng bộ đồng hồ.
• attempts: number - Tổng số lần thử lại job cho tới khi hoàn thành.
• repeat: RepeatOpts - Lặp lại job theo một cron specification. Xem RepeatOpts.
• backoff: number | BackoffOpts - Cấu hình backoff cho việc tự động thử lại khi job thất bại. Xem BackoffOpts.
• lifo: boolean - Nếu true, thêm job vào cuối bên phải của queue thay vì bên trái (mặc định false).
• jobId: number | string - Ghi đè ID của job - mặc định, ID job là một số nguyên duy nhất, nhưng bạn có thể dùng tùy chọn này để ghi đè nó. Nếu dùng tùy chọn này, bạn phải tự đảm bảo jobId là duy nhất. Nếu bạn cố thêm một job với id đã tồn tại, nó sẽ không được thêm vào.
• removeOnComplete: boolean | number - Nếu true, xóa job khi hoàn thành thành công. Một số cụ thể quy định số lượng job được giữ lại. Hành vi mặc định là giữ job trong tập hợp completed.
• removeOnFail: boolean | number - Nếu true, xóa job khi job thất bại sau tất cả các lần thử. Một số cụ thể quy định số lượng job được giữ lại. Hành vi mặc định là giữ job trong tập hợp failed.
• stackTraceLimit: number - Giới hạn số dòng stack trace được ghi lại.

Dưới đây là một vài ví dụ về việc tùy biến job bằng job options.

Để trì hoãn thời điểm bắt đầu một job, dùng thuộc tính cấu hình delay.

const job = await this.audioQueue.add(
  'transcode',
  {
    foo: 'bar',
  },
  { delay: 3000 }, // trì hoãn 3 giây
);

Để thêm một job vào cuối bên phải của queue (xử lý job theo kiểu LIFO - Last In First Out, vào sau ra trước), đặt thuộc tính lifo trong object cấu hình thành true.

const job = await this.audioQueue.add(
  'transcode',
  {
    foo: 'bar',
  },
  { lifo: true },
);

Để ưu tiên một job, dùng thuộc tính priority.

const job = await this.audioQueue.add(
  'transcode',
  {
    foo: 'bar',
  },
  { priority: 2 },
);

Để xem danh sách đầy đủ các tùy chọn, tham khảo tài liệu API tại đây và tại đây.

Consumers (Bên xử lý job)

Consumer là một class định nghĩa các phương thức để xử lý job được thêm vào queue, hoặc lắng nghe sự kiện trên queue, hoặc cả hai. Khai báo một class consumer bằng decorator @Processor() như sau:

import { Processor } from '@nestjs/bullmq';

@Processor('audio')
export class AudioConsumer {}

Trong đó, tham số chuỗi của decorator (ví dụ, 'audio') là tên của queue được gắn kết với các phương thức trong class.

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('audio')
export class AudioConsumer extends WorkerHost {
  async process(job: Job<any, any, string>): Promise<any> {
    let progress = 0;
    for (let i = 0; i < 100; i++) {
      await doSomething(job.data);
      progress += 1;
      await job.updateProgress(progress);
    }
    return {};
  }
}

Phương thức process được gọi mỗi khi worker rảnh (idle) và có job cần xử lý trong queue. Phương thức xử lý này nhận object job làm tham số duy nhất. Giá trị trả về từ phương thức xử lý được lưu trong object job và có thể truy cập sau này, ví dụ trong một listener lắng nghe sự kiện completed.

Object Job có nhiều phương thức cho phép bạn tương tác với trạng thái của nó. Ví dụ, đoạn code trên dùng phương thức updateProgress() để cập nhật tiến độ job. Xem tại đây để có tài liệu API đầy đủ về object Job.

Ở phiên bản cũ hơn là Bull, bạn có thể chỉ định một phương thức xử lý job sẽ chỉ xử lý duy nhất các job thuộc một loại nhất định (job có name cụ thể) bằng cách truyền name đó vào decorator @Process() như dưới đây.

@Process('transcode')
async transcode(job: Job<unknown>) { ... }

Hành vi này không được hỗ trợ trong BullMQ do gây ra nhiều nhầm lẫn. Thay vào đó, bạn cần dùng switch case để gọi các service hoặc logic khác nhau cho từng tên job:

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('audio')
export class AudioConsumer extends WorkerHost {
  async process(job: Job<any, any, string>): Promise<any> {
    switch (job.name) {
      case 'transcode': {
        let progress = 0;
        for (i = 0; i < 100; i++) {
          await doSomething(job.data);
          progress += 1;
          await job.progress(progress);
        }
        return {};
      }
      case 'concatenate': {
        await doSomeLogic2();
        break;
      }
    }
  }
}

Nội dung này được đề cập trong phần named processor của tài liệu BullMQ.

Consumer theo phạm vi request (Request-scoped consumers)

Khi một consumer được đánh dấu là request-scoped (tìm hiểu thêm về các phạm vi inject tại đây), một instance mới của class sẽ được tạo riêng cho mỗi job. Instance này sẽ được garbage-collect sau khi job hoàn thành.

@Processor({
  name: 'audio',
  scope: Scope.REQUEST,
})

Vì các class consumer request-scoped được khởi tạo động và gắn với phạm vi một job cụ thể, bạn có thể inject JOB_REF qua constructor theo cách tiêu chuẩn.

constructor(@Inject(JOB_REF) jobRef: Job) {
  console.log(jobRef);
}

Event listeners (Bên lắng nghe sự kiện)

BullMQ phát sinh một tập hợp các sự kiện hữu ích khi có thay đổi trạng thái của queue và/hoặc job. Các sự kiện này có thể được lắng nghe ở cấp Worker bằng decorator @OnWorkerEvent(event), hoặc ở cấp Queue bằng một class listener riêng và decorator @OnQueueEvent(event).

Sự kiện của Worker phải được khai báo bên trong một class consumer (nghĩa là bên trong một class được decorate bằng @Processor()). Để lắng nghe một sự kiện, dùng decorator @OnWorkerEvent(event) với sự kiện bạn muốn xử lý. Ví dụ, để lắng nghe sự kiện phát sinh khi một job chuyển sang trạng thái active trong queue audio, dùng cấu trúc sau:

import { Processor, Process, OnWorkerEvent } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('audio')
export class AudioConsumer {
  @OnWorkerEvent('active')
  onActive(job: Job) {
    console.log(
      `Processing job ${job.id} of type ${job.name} with data ${job.data}...`,
    );
  }

  // ...
}

Bạn có thể xem danh sách đầy đủ các sự kiện và tham số của chúng dưới dạng thuộc tính của WorkerListener tại đây.

QueueEvent listener phải dùng decorator @QueueEventsListener(queue) và kế thừa class QueueEventsHost được cung cấp bởi @nestjs/bullmq. Để lắng nghe một sự kiện, dùng decorator @OnQueueEvent(event) với sự kiện bạn muốn xử lý. Ví dụ, để lắng nghe sự kiện phát sinh khi một job chuyển sang trạng thái active trong queue audio, dùng cấu trúc sau:

import {
  QueueEventsHost,
  QueueEventsListener,
  OnQueueEvent,
} from '@nestjs/bullmq';

@QueueEventsListener('audio')
export class AudioEventsListener extends QueueEventsHost {
  @OnQueueEvent('active')
  onActive(job: { jobId: string; prev?: string }) {
    console.log(`Processing job ${job.jobId}...`);
  }

  // ...
}

Bạn có thể xem danh sách đầy đủ các sự kiện và tham số của chúng dưới dạng thuộc tính của QueueEventsListener tại đây.

Quản lý queue

Queue có một API cho phép bạn thực hiện các chức năng quản lý như tạm dừng và tiếp tục, lấy số lượng job theo các trạng thái khác nhau, và nhiều hơn nữa. Bạn có thể xem đầy đủ API của queue tại đây. Gọi trực tiếp bất kỳ phương thức nào trong số này trên object Queue, như minh họa dưới đây với ví dụ pause/resume.

Tạm dừng một queue bằng lệnh gọi phương thức pause(). Một queue bị tạm dừng sẽ không xử lý job mới cho tới khi được tiếp tục (resume), nhưng các job hiện đang được xử lý sẽ tiếp tục cho tới khi hoàn tất.

await audioQueue.pause();

Để tiếp tục một queue đang tạm dừng, dùng phương thức resume(), như sau:

await audioQueue.resume();

Quan sát queue trong production

Queue gặp sự cố theo cách khác với HTTP endpoint. Một job không trả về status code cho người dùng đang sốt ruột - nó âm thầm thử lại, ba lần, với backoff, và triệu chứng duy nhất là một điều gì đó ở downstream chưa từng xảy ra. Hai câu hỏi quan trọng nhất do đó là “queue này có đang xử lý kịp không?” và “job đó có thực sự chạy hay không?”, và không câu nào trả lời được chỉ bằng log của riêng consumer.

NestJS Observe tự động đo lường (instrument) các queue consumer, theo cùng cách nó đo lường controller - các class @Processor và handler của chúng được nhận diện như các job, nên bạn không cần tự nối span thủ công:

• Thời gian chờ trong queue được đo tách biệt với thời gian thực thi. Một job chạy mất 200 ms nhưng nằm chờ trong queue tới bốn phút là vấn đề về năng lực xử lý (capacity), không phải handler chậm, và hai con số này được báo cáo song song để bạn biết mình đang gặp vấn đề nào.
• Số lần thử và lý do thất bại được ghi lại cho từng lần chạy. Bạn thấy được job thành công ở lần thử thứ 3 thay vì chỉ thấy kết quả thành công, điều này thường là khác biệt giữa “ổn” và “đang âm thầm xuống cấp”.
• Sự im lặng cũng có thể cảnh báo được. Một quy tắc job silence (job im lặng) sẽ kích hoạt khi một job có tên cụ thể không báo cáo trong khoảng thời gian dài hơn ngưỡng bạn chọn - đây là cách bạn phát hiện ra consumer tính hóa đơn hằng đêm đã ngừng chạy, ngay trong đêm nó dừng, thay vì đợi tới cuối tháng.

Job thất bại mang theo error card giống hệt các request thất bại: stack trace đã được resolve kèm dòng mã nguồn, log được ghi lại trong quá trình chạy, và waterfall thể hiện những gì job đã làm trước khi nó throw lỗi. Xem chương Observability để biết cách thiết lập.

Process tách biệt

Handler xử lý job cũng có thể chạy trong một process tách biệt (forked) (nguồn). Cách này có một số lợi ích sau:

• Process được cách ly (sandbox), nên nếu nó crash cũng không ảnh hưởng tới worker.
• Bạn có thể chạy code blocking mà không ảnh hưởng tới queue (job sẽ không bị stall).
• Tận dụng tốt hơn nhiều CPU đa nhân.
• Ít kết nối tới Redis hơn.

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { join } from 'node:path';

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'audio',
      processors: [join(import.meta.dirname, 'processor.js')],
    }),
  ],
})
export class AppModule {}

Cấu hình bất đồng bộ (Async configuration)

Có thể bạn muốn truyền tùy chọn bullmq một cách bất đồng bộ thay vì tĩnh (static). Trong trường hợp này, dùng phương thức forRootAsync(), phương thức này cung cấp nhiều cách để xử lý cấu hình bất đồng bộ. Tương tự, nếu muốn truyền tùy chọn queue một cách bất đồng bộ, dùng phương thức registerQueueAsync().

Một cách tiếp cận là dùng factory function:

BullModule.forRootAsync({
  useFactory: () => ({
    connection: {
      host: 'localhost',
      port: 6379,
    },
  }),
});

Factory của chúng ta hoạt động giống như bất kỳ asynchronous provider nào khác (ví dụ, nó có thể là async và có thể inject dependency thông qua inject).

BullModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: async (configService: ConfigService) => ({
    connection: {
      host: configService.get('QUEUE_HOST'),
      port: configService.get('QUEUE_PORT'),
    },
  }),
  inject: [ConfigService],
});

Ngoài ra, bạn có thể dùng cú pháp useClass:

BullModule.forRootAsync({
  useClass: BullConfigService,
});

Cấu trúc trên sẽ khởi tạo BullConfigService bên trong BullModule và dùng nó để cung cấp một object tùy chọn bằng cách gọi createSharedConfiguration(). Lưu ý rằng điều này có nghĩa là BullConfigService phải implement interface SharedBullConfigurationFactory, như minh họa dưới đây:

@Injectable()
class BullConfigService implements SharedBullConfigurationFactory {
  createSharedConfiguration(): BullModuleOptions {
    return {
      connection: {
        host: 'localhost',
        port: 6379,
      },
    };
  }
}

Để tránh việc BullConfigService được tạo bên trong BullModule và thay vào đó dùng một provider được import từ module khác, bạn có thể dùng cú pháp useExisting.

BullModule.forRootAsync({
  imports: [ConfigModule],
  useExisting: ConfigService,
});

Cấu trúc này hoạt động giống useClass, chỉ khác một điểm quan trọng - BullModule sẽ tìm trong các module đã import để tái sử dụng một ConfigService đã có sẵn thay vì khởi tạo một cái mới.

Tương tự, nếu muốn truyền tùy chọn queue một cách bất đồng bộ, dùng phương thức registerQueueAsync(), chỉ cần nhớ khai báo thuộc tính name bên ngoài factory function.

BullModule.registerQueueAsync({
  name: 'audio',
  useFactory: () => ({
    redis: {
      host: 'localhost',
      port: 6379,
    },
  }),
});

Đăng ký thủ công (Manual registration)

Theo mặc định, BullModule tự động đăng ký các thành phần BullMQ (queue, processor, và các service listener sự kiện) trong lifecycle function onModuleInit. Tuy nhiên, trong một số trường hợp, hành vi này có thể không phù hợp. Để ngăn việc tự động đăng ký, hãy bật manualRegistration trong BullModule như sau:

BullModule.forRoot({
  extraOptions: {
    manualRegistration: true,
  },
});

Để đăng ký các thành phần này thủ công, inject BullRegistrar và gọi hàm register, lý tưởng nhất là trong OnModuleInit hoặc OnApplicationBootstrap.

import { Injectable, OnModuleInit } from '@nestjs/common';
import { BullRegistrar } from '@nestjs/bullmq';

@Injectable()
export class AudioService implements OnModuleInit {
  constructor(private bullRegistrar: BullRegistrar) {}

  onModuleInit() {
    if (yourConditionHere) {
      this.bullRegistrar.register();
    }
  }
}

Trừ khi bạn gọi hàm BullRegistrar#register, không có thành phần BullMQ nào hoạt động - nghĩa là sẽ không có job nào được xử lý.

Cài đặt Bull

Để bắt đầu dùng Bull, trước tiên hãy cài các dependency cần thiết.

$ npm install --save @nestjs/bull bull

Sau khi cài đặt xong, chúng ta có thể import BullModule vào AppModule gốc.

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bull';

@Module({
  imports: [
    BullModule.forRoot({
      redis: {
        host: 'localhost',
        port: 6379,
      },
    }),
  ],
})
export class AppModule {}

Phương thức forRoot() dùng để đăng ký một object cấu hình cho package bull, được tất cả các queue đăng ký trong ứng dụng sử dụng (trừ khi được chỉ định khác). Một object cấu hình bao gồm các thuộc tính sau:

• limiter: RateLimiter - Các tùy chọn để kiểm soát tốc độ xử lý job của queue. Xem RateLimiter để biết thêm. Không bắt buộc.
• redis: RedisOpts - Các tùy chọn để cấu hình kết nối Redis. Xem RedisOpts để biết thêm. Không bắt buộc.
• prefix: string - Tiền tố cho tất cả các key của queue. Không bắt buộc.
• defaultJobOptions: JobOpts - Các tùy chọn để cấu hình mặc định cho job mới. Xem JobOpts để biết thêm. Không bắt buộc. Lưu ý: các tùy chọn này không có tác dụng nếu bạn lập lịch job qua FlowProducer. Xem bullmq#1034 để biết giải thích.
• settings: AdvancedSettings - Các cài đặt nâng cao cho Queue. Thông thường không nên thay đổi các cài đặt này. Xem AdvancedSettings để biết thêm. Không bắt buộc.

Tất cả các tùy chọn đều không bắt buộc, giúp bạn kiểm soát chi tiết hành vi của queue. Các tùy chọn này được truyền thẳng vào constructor của Queue trong Bull. Đọc thêm về các tùy chọn này tại đây.

Để đăng ký một queue, import dynamic module BullModule.registerQueue() như sau:

BullModule.registerQueue({
  name: 'audio',
});

Phương thức registerQueue() dùng để khởi tạo và/hoặc đăng ký queue. Các queue được chia sẻ giữa các module và process kết nối tới cùng một database Redis với cùng thông tin đăng nhập. Mỗi queue là duy nhất theo thuộc tính name của nó. Tên queue được dùng vừa làm injection token (để inject queue vào controller/provider), vừa làm tham số cho decorator để gắn kết các class consumer và listener với queue.

Bạn cũng có thể ghi đè một số tùy chọn đã cấu hình sẵn cho một queue cụ thể, như sau:

BullModule.registerQueue({
  name: 'audio',
  redis: {
    port: 6380,
  },
});

Vì job được lưu trữ trong Redis, mỗi lần một queue có tên cụ thể được khởi tạo (ví dụ, khi ứng dụng khởi động/khởi động lại), nó sẽ cố gắng xử lý các job cũ còn sót lại từ phiên làm việc chưa hoàn tất trước đó.

Mỗi queue có thể có một hoặc nhiều producer, consumer và listener. Consumer lấy job từ queue theo một thứ tự cụ thể: FIFO (mặc định), LIFO, hoặc theo mức độ ưu tiên. Cách kiểm soát thứ tự xử lý queue được bàn ở đây.

Cấu hình có tên (Named configurations)

Nếu các queue của bạn kết nối tới nhiều instance Redis, bạn có thể dùng kỹ thuật gọi là named configurations. Tính năng này cho phép bạn đăng ký nhiều cấu hình dưới các key được chỉ định, sau đó bạn có thể tham chiếu tới chúng trong các tùy chọn của queue.

Ví dụ, giả sử bạn có thêm một instance Redis khác (ngoài instance mặc định) được một vài queue trong ứng dụng sử dụng, bạn có thể đăng ký cấu hình cho nó như sau:

BullModule.forRoot('alternative-config', {
  redis: {
    port: 6381,
  },
});

Trong ví dụ trên, 'alternative-config' chỉ đơn giản là một key cấu hình (có thể là bất kỳ chuỗi tùy ý nào).

Với thiết lập này, giờ đây bạn có thể trỏ tới cấu hình này trong object tùy chọn của registerQueue():

BullModule.registerQueue({
  configKey: 'alternative-config',
  name: 'video',
});

Producers (Bên tạo job)

Job producer là bên đưa job vào queue. Producer thường là các service của ứng dụng (Nest provider). Để thêm job vào queue, trước tiên hãy inject queue vào service như sau:

import { Injectable } from '@nestjs/common';
import { Queue } from 'bull';
import { InjectQueue } from '@nestjs/bull';

@Injectable()
export class AudioService {
  constructor(@InjectQueue('audio') private audioQueue: Queue) {}
}

Bây giờ, thêm một job bằng cách gọi phương thức add() của queue, truyền vào một object job do bạn tự định nghĩa. Job được biểu diễn dưới dạng object JavaScript có thể serialize (vì đó là cách chúng được lưu trong database Redis). Hình dạng của job bạn truyền vào là tùy ý; hãy dùng nó để biểu diễn ngữ nghĩa của object job.

const job = await this.audioQueue.add({
  foo: 'bar',
});

Named jobs (Job có tên)

Job có thể có tên riêng biệt. Điều này cho phép bạn tạo các consumer chuyên biệt, chỉ xử lý các job có tên nhất định.

const job = await this.audioQueue.add('transcode', {
  foo: 'bar',
});

Warning Cảnh báo Khi dùng named job, bạn phải tạo processor cho từng tên riêng biệt được thêm vào queue, nếu không queue sẽ báo lỗi rằng bạn đang thiếu processor cho job đó. Xem tại đây để biết thêm về cách tiêu thụ named job.

Job options (Tùy chọn job)

Job có thể có thêm các tùy chọn đi kèm. Truyền một object tùy chọn sau tham số job trong phương thức Queue.add(). Các thuộc tính của job options gồm:

• priority: number - Giá trị độ ưu tiên (không bắt buộc). Nằm trong khoảng từ 1 (ưu tiên cao nhất) đến MAX_INT (ưu tiên thấp nhất). Lưu ý rằng việc dùng priority có ảnh hưởng nhẹ tới hiệu năng, nên hãy cân nhắc khi sử dụng.
• delay: number - Khoảng thời gian (mili-giây) cần chờ trước khi job này có thể được xử lý. Lưu ý để có độ trễ chính xác, cả server và client cần đồng bộ đồng hồ.
• attempts: number - Tổng số lần thử lại job cho tới khi hoàn thành.
• repeat: RepeatOpts - Lặp lại job theo một cron specification. Xem RepeatOpts.
• backoff: number | BackoffOpts - Cấu hình backoff cho việc tự động thử lại khi job thất bại. Xem BackoffOpts.
• lifo: boolean - Nếu true, thêm job vào cuối bên phải của queue thay vì bên trái (mặc định false).
• timeout: number - Số mili-giây sau đó job sẽ thất bại với lỗi timeout.
• jobId: number | string - Ghi đè ID của job - mặc định, ID job là một số nguyên duy nhất, nhưng bạn có thể dùng tùy chọn này để ghi đè nó. Nếu dùng tùy chọn này, bạn phải tự đảm bảo jobId là duy nhất. Nếu bạn cố thêm một job với id đã tồn tại, nó sẽ không được thêm vào.
• removeOnComplete: boolean | number - Nếu true, xóa job khi hoàn thành thành công. Một số cụ thể quy định số lượng job được giữ lại. Hành vi mặc định là giữ job trong tập hợp completed.
• removeOnFail: boolean | number - Nếu true, xóa job khi job thất bại sau tất cả các lần thử. Một số cụ thể quy định số lượng job được giữ lại. Hành vi mặc định là giữ job trong tập hợp failed.
• stackTraceLimit: number - Giới hạn số dòng stack trace được ghi lại.

Dưới đây là một vài ví dụ về việc tùy biến job bằng job options.

Để trì hoãn thời điểm bắt đầu một job, dùng thuộc tính cấu hình delay.

const job = await this.audioQueue.add(
  {
    foo: 'bar',
  },
  { delay: 3000 }, // trì hoãn 3 giây
);

Để thêm một job vào cuối bên phải của queue (xử lý job theo kiểu LIFO - Last In First Out), đặt thuộc tính lifo trong object cấu hình thành true.

const job = await this.audioQueue.add(
  {
    foo: 'bar',
  },
  { lifo: true },
);

Để ưu tiên một job, dùng thuộc tính priority.

const job = await this.audioQueue.add(
  {
    foo: 'bar',
  },
  { priority: 2 },
);

Consumers (Bên xử lý job)

Consumer là một class định nghĩa các phương thức để xử lý job được thêm vào queue, hoặc lắng nghe sự kiện trên queue, hoặc cả hai. Khai báo một class consumer bằng decorator @Processor() như sau:

import { Processor } from '@nestjs/bull';

@Processor('audio')
export class AudioConsumer {}

Trong đó, tham số chuỗi của decorator (ví dụ, 'audio') là tên của queue được gắn kết với các phương thức trong class.

Bên trong một class consumer, khai báo job handler bằng cách decorate các phương thức xử lý với decorator @Process().

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('audio')
export class AudioConsumer {
  @Process()
  async transcode(job: Job<unknown>) {
    let progress = 0;
    for (let i = 0; i < 100; i++) {
      await doSomething(job.data);
      progress += 1;
      await job.progress(progress);
    }
    return {};
  }
}

Phương thức được decorate (ví dụ, transcode()) được gọi mỗi khi worker rảnh và có job cần xử lý trong queue. Phương thức xử lý này nhận object job làm tham số duy nhất. Giá trị trả về từ phương thức xử lý được lưu trong object job và có thể truy cập sau này, ví dụ trong một listener lắng nghe sự kiện completed.

Object Job có nhiều phương thức cho phép bạn tương tác với trạng thái của nó. Ví dụ, đoạn code trên dùng phương thức progress() để cập nhật tiến độ job. Xem tại đây để có tài liệu API đầy đủ về object Job.

Bạn có thể chỉ định một phương thức xử lý job sẽ chỉ xử lý duy nhất các job thuộc một loại nhất định (job có name cụ thể) bằng cách truyền name đó vào decorator @Process() như dưới đây. Bạn có thể có nhiều handler @Process() trong một class consumer, tương ứng với mỗi loại job (name). Khi dùng named job, hãy đảm bảo có một handler tương ứng cho mỗi tên.

@Process('transcode')
async transcode(job: Job<unknown>) { ... }

Consumer theo phạm vi request (Request-scoped consumers)

Khi một consumer được đánh dấu là request-scoped (tìm hiểu thêm về các phạm vi inject tại đây), một instance mới của class sẽ được tạo riêng cho mỗi job. Instance này sẽ được garbage-collect sau khi job hoàn thành.

@Processor({
  name: 'audio',
  scope: Scope.REQUEST,
})

Vì các class consumer request-scoped được khởi tạo động và gắn với phạm vi một job cụ thể, bạn có thể inject JOB_REF qua constructor theo cách tiêu chuẩn.

constructor(@Inject(JOB_REF) jobRef: Job) {
  console.log(jobRef);
}

Event listeners (Bên lắng nghe sự kiện)

Bull phát sinh một tập hợp các sự kiện hữu ích khi có thay đổi trạng thái của queue và/hoặc job. Nest cung cấp một tập hợp decorator cho phép bạn subscribe vào các sự kiện chuẩn cốt lõi. Chúng được export từ package @nestjs/bull.

Event listener phải được khai báo bên trong một class consumer (nghĩa là bên trong một class được decorate bằng @Processor()). Để lắng nghe một sự kiện, dùng một trong các decorator ở bảng dưới đây để khai báo handler cho sự kiện đó. Ví dụ, để lắng nghe sự kiện phát sinh khi một job chuyển sang trạng thái active trong queue audio, dùng cấu trúc sau:

import { Processor, Process, OnQueueActive } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('audio')
export class AudioConsumer {

  @OnQueueActive()
  onActive(job: Job) {
    console.log(
      `Processing job ${job.id} of type ${job.name} with data ${job.data}...`,
    );
  }
  ...

Vì Bull hoạt động trong môi trường phân tán (đa node), nó định nghĩa khái niệm về tính cục bộ của sự kiện (event locality). Khái niệm này công nhận rằng sự kiện có thể được kích hoạt hoàn toàn trong một process duy nhất, hoặc trên các queue được chia sẻ từ nhiều process khác nhau. Sự kiện local (cục bộ) là sự kiện được sinh ra khi một hành động hoặc thay đổi trạng thái được kích hoạt trên một queue ở process cục bộ. Nói cách khác, khi event producer và consumer của bạn đều cục bộ trong cùng một process, mọi sự kiện xảy ra trên queue đều là local.

Khi một queue được chia sẻ giữa nhiều process, chúng ta gặp khả năng xuất hiện sự kiện global (toàn cục). Để một listener ở một process nhận được thông báo sự kiện do một process khác kích hoạt, nó phải đăng ký lắng nghe sự kiện global.

Event handler được gọi mỗi khi sự kiện tương ứng được phát sinh. Handler được gọi với chữ ký (signature) như bảng dưới đây, cung cấp quyền truy cập tới thông tin liên quan tới sự kiện. Chúng ta sẽ bàn về một khác biệt quan trọng giữa chữ ký handler của sự kiện local và global dưới đây.

Khi lắng nghe sự kiện global, chữ ký phương thức có thể hơi khác so với phiên bản local tương ứng. Cụ thể, bất kỳ chữ ký phương thức nào nhận object job trong phiên bản local, thì ở phiên bản global lại nhận một jobId (number) thay thế. Để lấy tham chiếu tới object job thực sự trong trường hợp này, dùng phương thức Queue#getJob. Lệnh gọi này cần được await, do đó handler cần được khai báo là async. Ví dụ:

@OnGlobalQueueCompleted()
async onGlobalCompleted(jobId: number, result: any) {
  const job = await this.immediateQueue.getJob(jobId);
  console.log('(Global) on completed: job ', job.id, ' -> result: ', result);
}

Ngoài các decorator listener sự kiện chuyên biệt, bạn cũng có thể dùng decorator chung @OnQueueEvent() kết hợp với enum BullQueueEvents hoặc BullQueueGlobalEvents. Đọc thêm về sự kiện tại đây.

Quản lý queue

Queue có một API cho phép bạn thực hiện các chức năng quản lý như tạm dừng và tiếp tục, lấy số lượng job theo các trạng thái khác nhau, và nhiều hơn nữa. Bạn có thể xem đầy đủ API của queue tại đây. Gọi trực tiếp bất kỳ phương thức nào trong số này trên object Queue, như minh họa dưới đây với ví dụ pause/resume.

Tạm dừng một queue bằng lệnh gọi phương thức pause(). Một queue bị tạm dừng sẽ không xử lý job mới cho tới khi được tiếp tục, nhưng các job hiện đang được xử lý sẽ tiếp tục cho tới khi hoàn tất.

await audioQueue.pause();

Để tiếp tục một queue đang tạm dừng, dùng phương thức resume(), như sau:

await audioQueue.resume();

Process tách biệt

Handler xử lý job cũng có thể chạy trong một process tách biệt (forked) (nguồn). Cách này có một số lợi ích sau:

• Process được cách ly (sandbox), nên nếu nó crash cũng không ảnh hưởng tới worker.
• Bạn có thể chạy code blocking mà không ảnh hưởng tới queue (job sẽ không bị stall).
• Tận dụng tốt hơn nhiều CPU đa nhân.
• Ít kết nối tới Redis hơn.

import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bull';
import { join } from 'node:path';

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'audio',
      processors: [join(import.meta.dirname, 'processor.js')],
    }),
  ],
})
export class AppModule {}

Lưu ý rằng vì hàm của bạn được thực thi trong một process đã fork, Dependency Injection (và IoC container) sẽ không khả dụng. Điều đó có nghĩa là hàm processor của bạn cần tự chứa (hoặc tự tạo) tất cả các instance dependency bên ngoài mà nó cần.

import { Job, DoneCallback } from 'bull';

export default function (job: Job, cb: DoneCallback) {
  console.log(`[${process.pid}] ${JSON.stringify(job.data)}`);
  cb(null, 'It works');
}

Cấu hình bất đồng bộ (Async configuration)

Có thể bạn muốn truyền tùy chọn bull một cách bất đồng bộ thay vì tĩnh. Trong trường hợp này, dùng phương thức forRootAsync(), phương thức này cung cấp nhiều cách để xử lý cấu hình bất đồng bộ.

Một cách tiếp cận là dùng factory function:

BullModule.forRootAsync({
  useFactory: () => ({
    redis: {
      host: 'localhost',
      port: 6379,
    },
  }),
});

Factory của chúng ta hoạt động giống như bất kỳ asynchronous provider nào khác (ví dụ, nó có thể là async và có thể inject dependency thông qua inject).

BullModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: async (configService: ConfigService) => ({
    redis: {
      host: configService.get('QUEUE_HOST'),
      port: configService.get('QUEUE_PORT'),
    },
  }),
  inject: [ConfigService],
});

Ngoài ra, bạn có thể dùng cú pháp useClass:

BullModule.forRootAsync({
  useClass: BullConfigService,
});

Cấu trúc trên sẽ khởi tạo BullConfigService bên trong BullModule và dùng nó để cung cấp một object tùy chọn bằng cách gọi createSharedConfiguration(). Lưu ý rằng điều này có nghĩa là BullConfigService phải implement interface SharedBullConfigurationFactory, như minh họa dưới đây:

@Injectable()
class BullConfigService implements SharedBullConfigurationFactory {
  createSharedConfiguration(): BullModuleOptions {
    return {
      redis: {
        host: 'localhost',
        port: 6379,
      },
    };
  }
}

Để tránh việc BullConfigService được tạo bên trong BullModule và thay vào đó dùng một provider được import từ module khác, bạn có thể dùng cú pháp useExisting.

BullModule.forRootAsync({
  imports: [ConfigModule],
  useExisting: ConfigService,
});

Cấu trúc này hoạt động giống useClass, chỉ khác một điểm quan trọng - BullModule sẽ tìm trong các module đã import để tái sử dụng một ConfigService đã có sẵn thay vì khởi tạo một cái mới.

Tương tự, nếu muốn truyền tùy chọn queue một cách bất đồng bộ, dùng phương thức registerQueueAsync(), chỉ cần nhớ khai báo thuộc tính name bên ngoài factory function.

BullModule.registerQueueAsync({
  name: 'audio',
  useFactory: () => ({
    redis: {
      host: 'localhost',
      port: 6379,
    },
  }),
});

Ví dụ

Một ví dụ hoạt động đầy đủ có tại đây.