Membikin Queue Job Nestjs Menggunakan Bull

RUANGDEVELOPER.com – Queue job (tugas antrean) adalah konsep dalam pemrograman dan pengembangan aplikasi yang digunakan Demi menangani tugas atau pekerjaan secara asynchronous. Pada dasarnya, konsep ini melibatkan pengiriman tugas atau pekerjaan ke dalam antrian (queue) Demi dieksekusi nanti, tanpa memblokir eksekusi program Penting.

NestJS adalah salah satu framework yang menyediakan Metode yang simpel bagi kita Demi Membikin queue job, Yakni dengan menggunakan package Bull. Bull menyediakan Metode Demi Membikin, mengelola, dan mengeksekusi antrian pekerjaan secara asynchronous di lingkungan Node.js. NestJs telah mendukung penuh penggunaan Bull Demi mengelola antrean yang telah didokumentasikan di sini.

Yuk kita gas… 🚀

Menyiapkan Project

Membikin project NestJS

Install dependency

yarn add @nestjs/bull bull @bull-board/nestjs @bull-board/api @bull-board/express @nestjs/axios axios

Konfigurasi Bull & Bull Dashboard

Demi menggunakan Bull Queue kita perlu Mempunyai redis server sebagai database Demi menyimpan job yang kita buat. Maka dari itu, pastikan Anda sudah Mempunyai server redis yang jalan di local / server hosting Anda.

Buka file app.module.ts dan import BullModule serta sesuaikan koneksi redis server.

@Module({
  imports: [
    // Import BullModule
    BullModule.forRoot({
      redis: {
        host: "localhost",
        port: 6379,
      },
    }),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Tetap pada file app.module.ts, registrasikan queue yang Ingin kita buat dengan menggunakan method BullModule.registerQueue()

@Module({
  imports: [
    // Import BullModule
    BullModule.forRoot({
      redis: {
        host: "localhost",
        port: 6379,
      },
    }),
    BullModule.registerQueue(
      // Registrasikan queue dengan nama network-request-queue
      {
        name: "network-request-queue",
      }
    ),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Agar kita Bisa memonitor queue yang telah diregistrasikan, kita import juga konfigurasi BullBoard.

@Module({
  imports: [
    BullModule.forRoot({
      redis: {
        host: "localhost",
        port: 6379,
      },
    }),
    BullModule.registerQueue({
      name: "network-request-queue",
    }),
    // Konfigurasi BullBoard
    BullBoardModule.forRoot({
      route: "/queue-monitor",
      adapter: ExpressAdapter,
    }),
    // Registrasikan queue ke BullBoard
    BullBoardModule.forFeature({
      name: "network-request-queue",
      adapter: BullAdapter,
    }),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Pada bagian konfigurasi BullBoard, beberapa konfigurasi yang kita atur adalah sebagai berikut:

route: Ini adalah konfigurasi rute URL tempat BullBoard akan diakses. Dalam Misalnya ini, dashboard akan diakses melalui ‘/queue-monitor’.
adapter: Ini Demi menentukan jenis adapter yang digunakan oleh BullBoard. Dalam kasus ini, kita menggunakan ExpressAdapter, yang menyesuaikan BullBoard dengan aplikasi Express.

READ: Apa Itu Webhook? | Digimensia

Dan pada bagian registrasi queue ke BullBoard, beberapa konfigurasi yang kita atur adalah sebagai berikut:

name: Ini adalah konfigurasi nama queue yang akan didaftarkan ke BullBoard. Dalam Misalnya ini, queue bernama ’network-request-queue’.
adapter: Ini Demi menentukan jenis adapter yang digunakan Demi queue yang didaftarkan. Dalam kasus ini, kita menggunakan BullAdapter, yang memungkinkan BullBoard berkomunikasi dengan queue yang digunakan oleh Bull.

Setelah Segala beres, mari kita coba menjalankan aplikasi Demi memasikan Segala konfigurasi sudah Betul.

Akses aplikasi melalui url http://127.0.0.1:3000/queue-monitor, pastikan Anda dapat Menyaksikan queue dengan nama sesuai dengan konfigurasi kita.

Membikin Queue Producer

Queue producer adalah komponen yang menempatkan tugas atau pesan ke dalam queue. Komponen ini Membikin atau menghasilkan pekerjaan yang nantinya akan dieksekusi secara asinkron oleh konsumen queue (queue consumer). Produsen menangani penempatan tugas, penjadwalan, dan komunikasi antar service.

Demi Membikin queue producer, kita Bisa menggunakan file app.service.ts. Mari kita buka file tersebut, kemudian ganti kodenya menjadi seperti berikut:

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

@Injectable()
export class AppService {
  constructor(
    @InjectQueue("network-request-queue")
    private readonly networkRequestQueue: Queue
  ) {}

  async addNetworkRequestJob(data: { url: string }) {
    await this.networkRequestQueue.add(data);
  }
}

Pada class service tersebut, kita inject queue yang sudah kita registrasikan sebelumnya. Kemudian kita juga menambahkan satu method Demi menambahkan job ke dalam queue dengan data berupa alamat URL Demi melakukan network request.

Sederhananya kita akan Membikin queue job yang akan melakukan network request ke URL yang dikirimkan pada data queue.

Membikin Queue Consumer

Queue consumer adalah komponen dalam sistem yang bertugas mengambil, memproses, dan mengeksekusi tugas atau pesan dari antrian. Ini memungkinkan eksekusi tugas secara asinkron, meningkatkan responsivitas sistem. Consumer Lalu memantau antrian, eksekusi tugas, dan mengelola kesalahan. Skalabilitas dapat ditingkatkan dengan menambahkan lebih banyak instance.

Demi Membikin queue consumer, silahkan buat sebuah file baru bernama app.consumer.ts dan tambahkan kode seperti berikut:

import {
  OnQueueActive,
  OnQueueCompleted,
  OnQueueFailed,
  Process,
  Processor,
} from "@nestjs/bull";
import { Logger } from "@nestjs/common";
import { Job } from "bull";

@Processor("network-request-queue")
export class AppConsumer {
  @Process()
  async process(job: Job<{ url: string }>) {
    Logger.log(`Processing job ${job.id}`);
    Logger.log(job.data);
  }

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

  @OnQueueCompleted()
  onCompleted(job: Job, result: any) {
    Logger.log(`Completed job ${job.id} of type ${job.name}`);
    Logger.log(result);
  }

  @OnQueueFailed()
  onError(job: Job<any>, error: any) {
    Logger.error(`Failed job ${job.id} of type ${job.name}: ${error.message}`);
  }
}

@Process() – process(job: Job<{ url: string }>):

Method ini menandakan bahwa fungsi process akan menangani eksekusi job ketika diambil dari queue.
Fungsi ini menerima parameter job, yang merupakan representasi dari job yang akan dieksekusi.
Pada Misalnya ini, log dicetak Demi menunjukkan bahwa job sedang diproses Berbarengan dengan mencetak data dari job.
@OnQueueActive() – onActive(job: Job):

Method ini akan dipanggil ketika suatu job aktif atau sedang dieksekusi oleh queue consumer.
Menerima parameter job yang menyediakan informasi tentang job yang aktif.
Pada Misalnya ini, log dicetak Demi menunjukkan bahwa job sedang aktif atau dalam proses eksekusi.
@OnQueueCompleted() – onCompleted(job: Job, result: any):

Method ini akan dipanggil setelah job berhasil diselesaikan (tanpa kesalahan).
Menerima dua parameter: job yang menyediakan informasi tentang job yang selesai dan result yang mewakili hasil dari eksekusi job.
Pada Misalnya ini, log dicetak Demi menunjukkan bahwa job telah diselesaikan Berbarengan dengan mencetak hasil eksekusi job.
@OnQueueFailed() – onError(job: Job, error: any):

Method ini akan dipanggil Apabila job gagal dieksekusi (terjadi kesalahan).
Menerima dua parameter: job yang menyediakan informasi tentang job yang gagal dan error yang merupakan objek kesalahan.
Pada Misalnya ini, log error dicetak Demi menunjukkan bahwa job gagal dan mencetak pesan kesalahan.
Dengan menggunakan decorator seperti @Process, @OnQueueActive, @OnQueueCompleted, dan @OnQueueFailed, Anda dapat dengan mudah menentukan logika yang berbeda Demi berbagai tahap dalam siklus hidup eksekusi job di dalam queue pekerjaan.

READ: Apa Itu Load Balancer? Pengertian, Jenis, dan Manfaat Demi Website

Setelah queue consumer dibuat, kita perlu menambahkannya ke dalam array providers pada kelas AppModule. Ini bertujuan agar consumer kita Bisa digunakan Demi mengeksekusi job yang kita tambahkan.

@Module({
  /**
   * Kode lainnya...
   */
  providers: [AppService, AppConsumer],
})
export class AppModule {}

Konfigurasi HTTP Module

Setelah Membikin queue consumer, berikutnya kita akan mengkonfigurasi HTTP module Demi melakukan network request. Silahkan buka kembali file app.module.ts dan import HttpModule ke dalam AppModule.

@Module({
  imports: [
    /**
     * Kode import lainnya
     */
    HttpModule,
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Modifikasi Method process Pada Queue Consumer

Setelah HttpModule diimport, kita dapat menggunakan HttpService pada queue consumer Demi melakukan network request. Buka kembali file app.consumer.ts kemudian Inject HttpService ke kelas AppConsumer.

@Injectable()
@Processor("network-request-queue")
export class AppConsumer {
  constructor(private readonly httpService: HttpService) {}
  /**
   * Kode lainnya...
   */
}

Setelah itu modifikasi method process menjadi seperti ini:

@Process()
async process(job: Job<{ url: string }>) {
  Logger.log(`Processing job ${job.id}`);
  Logger.log(job.data);
  const { data } = await firstValueFrom(this.httpService.get(job.data.url).pipe(
    catchError((error) => {
      Logger.error(error);
      throw error;
    })
  ));

  return data;
}

Menambahkan Queue Job

Kita telah Mempunyai kelas AppService yang berisi method addNetworkRequestJob Demi menambahkan queue job. Demi menggunakannya kita Membikin sebuah endpoint baru di controller dan mengeksekusi method addNetworkRequestJob. Silahkan buka file app.controller.ts dan modifikasi menjadi seperti berikut ini:

import { Body, Controller, Post } from "@nestjs/common";
import { AppService } from "./app.service";

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Post("/add-job")
  async addJob(@Body() data: { url: string }) {
    await this.appService.addNetworkRequestJob({
      url: data.url,
    });

    return {
      message: "Job added",
    };
  }
}

Pada contoller tersebut, kita menambahkan endpoint /add-job yang menerima request body berisi url Demi kemudian kita gunakan pada queue job.

READ: #1: Setup Project - Laravel URL Shorterner

Pengujian Queue Job

Segala konfigurasi dan kode sudah selesai dibuat, saatnya kita mencoba aplikasi yang kita buat.

Pastikan redis server Anda telah diaktifkan
Jalankan aplikasi
Buka queue monitor (http://127.0.0.1:3000/queue-monitor)
Lakukan post request ke endpoint Demi menambahkan job
Pada bagian ini saya menggunakan Ekspansi vscode bernama rest-client, Anda juga Bisa menggunakan Metode lain seperti menggunakan Postman atau Enggak bisa tidur
Periksa queue job melalui queue monitor dashboard
Pastikan Anda dapat Menyaksikan queue yang sebelumnya sudah ditambahkan.

Pada bagian ini, queue job saya terdapat pada tab Completed yang berarti sudah selesai dieksekusi. Anda juga mungkin akan Menyaksikan queue job yang Anda tambahkan berada pada tab Active, yang berarti proses sedang berlangsung.

Selamat Anda sudah berhasil Membikin queue job menggunakan Bull. Tetap banyak konfigurasi queue job yang dapat diterapkan Demi menyesuaikan dengan kebutuhan aplikasi Anda. Selengkapnya, Anda Bisa periksa langsung di sumber berikut:

Source code project pada tutorial ini dapat Anda lihat di link berikut:

Latest News