Akademi Manajemen Informatika

Membangun Queue Job Nestjs Menggunakan Bull

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

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

Yuk kita gas… 🚀

Menyiapkan Project

Membangun 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

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

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 Mau 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 Dapat 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 Teladan ini, dashboard akan diakses melalui ‘/queue-monitor’.
  • adapter: Ini Kepada menentukan jenis adapter yang digunakan oleh BullBoard. Dalam kasus ini, kita menggunakan ExpressAdapter, yang menyesuaikan BullBoard dengan aplikasi Express.
READ  Apa itu Link? Definisi, Fungsi, Jenis dan 8 Istilahnya

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 Teladan ini, queue bernama ’network-request-queue’.
  • adapter: Ini Kepada menentukan jenis adapter yang digunakan Kepada 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 Kepada memasikan Segala konfigurasi sudah Betul.

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

Bull Dashboard

Membangun Queue Producer

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

Kepada Membangun queue producer, kita Dapat 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 Kepada menambahkan job ke dalam queue dengan data berupa alamat URL Kepada melakukan network request.

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

Membangun 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 Maju memantau antrian, eksekusi tugas, dan mengelola kesalahan. Skalabilitas dapat ditingkatkan dengan menambahkan lebih banyak instance.

Kepada Membangun 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: Jobany>, error: any) {
    Logger.error(`Failed job ${job.id} of type ${job.name}: ${error.message}`);
  }
}

  • @Process() – process(job: Job):

    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 Teladan ini, log dicetak Kepada 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 Teladan ini, log dicetak Kepada 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 Teladan ini, log dicetak Kepada 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 Teladan ini, log error dicetak Kepada 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 Kepada berbagai tahap dalam siklus hidup eksekusi job di dalam queue pekerjaan.

READ  #6: Variable - Belajar Golang Dari Dasar

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

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

Konfigurasi HTTP Module

Setelah Membangun queue consumer, berikutnya kita akan mengkonfigurasi HTTP module Kepada 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 Kepada 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 Kepada menambahkan queue job. Kepada menggunakannya kita Membangun 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 Kepada kemudian kita gunakan pada queue job.

READ  Pengertian, Jenis, Fungsi, dan Kelebihan

Pengujian Queue Job

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

  • Pastikan redis server Engkau telah diaktifkan

  • Jalankan aplikasi

  • Buka queue monitor (http://127.0.0.1:3000/queue-monitor)

  • Lakukan post request ke endpoint Kepada menambahkan job
    Pada bagian ini saya menggunakan Ekspansi vscode bernama rest-client, Engkau juga Dapat menggunakan Metode lain seperti menggunakan Postman atau Susah tidur

    Post Request Untuk Menambahkan Job

  • Periksa queue job melalui queue monitor dashboard
    Pastikan Engkau dapat Memperhatikan queue yang sebelumnya sudah ditambahkan.

    Queue Job Terlihat Pada Queue Monitor

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

Selamat Engkau sudah berhasil Membangun queue job menggunakan Bull. Tetap banyak konfigurasi queue job yang dapat diterapkan Kepada menyesuaikan dengan kebutuhan aplikasi Engkau. Selengkapnya, Engkau Dapat periksa langsung di sumber berikut:

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