Pola Worker Pool

Chapter Worker Threads di Section Asynchronous Runtime telah membuktikan bahwa komputasi berat bisa dipindahkan ke thread terpisah. Aplikasi nyata jarang punya satu tugas berat — biasanya ada sungai tugas yang datang terus dari pengguna: kompresi berkas unggahan, pembuatan thumbnail, hash password, pengolahan laporan. Membuat satu worker baru per tugas akan membayar biaya pembuatan thread berulang kali tanpa kendali berapa banyak thread yang hidup bersamaan. Pola worker pool menjawabnya dengan jumlah pekerja tetap dan satu antrean: pekerja yang menganggur mengambil tugas berikutnya, seperti kasir berjumlah tetap melayani antrian pelanggan. Chapter ini membangun pool semacam itu dari nol.

Anatomi Worker Pool #

Komponen sebuah pool selalu tiga:

Jumlah pekerja biasanya disetel mendekati jumlah core CPU — lebih banyak dari itu hanya menambah perpindahan konteks antar thread tanpa menambah daya hitung. Thread dari worker_threads lebih ringan daripada proses cluster karena sesama thread berbagi memori proses yang sama, tetapi tetap bukan gratis; itulah mengapa pool menahannya pada angka tetap.

Membangun Pool Sederhana #

Demo kita mengolah empat tugas checksum melalui dua pekerja. Berkas pertama adalah pekerjanya — sekadar menerima tugas, menghitung, dan melapor:

// Pekerja serbaguna: menerima tugas lewat pesan,
// menghitung checksum, lalu melaporkan hasil.
import { parentPort } from "node:worker_threads";
import { createHash } from "node:crypto";

parentPort.on("message", ({ id, data, iterasi }) => {
  const hash = createHash("sha256");
  for (let i = 0; i < iterasi; i++) {
    hash.update(data);
  }
  parentPort.postMessage({
    id,
    digest: hash.digest("hex").slice(0, 16),
  });
});

Berkas kedua adalah pool-nya beserta program utamanya:

// Kolam pekerja: antrean tugas + sekumpulan worker_threads.
// Tugas masuk antrean; pekerja yang menganggur mengambil satu.
import { Worker } from "node:worker_threads";

class KolamPekerja {
  constructor(jumlahPekerja, berkasPekerja) {
    this.antrean = [];
    this.anggota = [];
    for (let i = 0; i < jumlahPekerja; i++) {
      this.anggota.push({
        worker: new Worker(berkasPekerja),
        sibuk: false,
      });
    }
  }

  jalankan(tugas) {
    return new Promise((selesai, gagal) => {
      this.antrean.push({ tugas, selesai, gagal });
      this.isiTugas();
    });
  }

  isiTugas() {
    for (const anggota of this.anggota) {
      if (anggota.sibuk || this.antrean.length === 0) continue;
      const { tugas, selesai, gagal } = this.antrean.shift();
      anggota.sibuk = true;
      anggota.worker.once("message", (hasil) => {
        anggota.sibuk = false;
        selesai(hasil);
        this.isiTugas();
      });
      anggota.worker.once("error", (galat) => {
        anggota.sibuk = false;
        gagal(galat);
        this.isiTugas();
      });
      anggota.worker.postMessage(tugas);
    }
  }

  async tutup() {
    await Promise.all(
      this.anggota.map((anggota) => anggota.worker.terminate())
    );
  }
}

const berkasPekerja = new URL("./pekerja.mjs", import.meta.url);
const kolam = new KolamPekerja(2, berkasPekerja);

const tugas = [];
for (let id = 1; id <= 4; id++) {
  tugas.push(
    kolam.jalankan({ id, data: `potongan-${id}`, iterasi: 200000 })
  );
}

const hasil = await Promise.all(tugas);
await kolam.tutup();

// Urutan selesai tidak dijamin; urutkan sebelum mencetak.
hasil.sort((a, b) => a.id - b.id);
for (const { id, digest } of hasil) {
  console.log(`tugas ${id}: checksum ${digest}`);
}
console.log("total tugas selesai:", hasil.length);

Output yang Dihasilkan:

tugas 1: checksum a3b00e1ad3a05366
tugas 2: checksum 9b84edecd58c8bce
tugas 3: checksum 48e2b9bce6203b3c
tugas 4: checksum ad64552cf90d7aaf
total tugas selesai: 4

Mari kita bedah kedua berkas tersebut:

1. pekerja.mjs baris 6-14: worker mendengarkan pesan dari parentPort . Setiap tugas berisi id, data, dan iterasi — checksum SHA-256 dihitung dengan hash.update() berulang untuk memberi CPU pekerjaan yang sesungguhnya, lalu hasilnya dikirim balik. Pekerja sengaja tidak menutup portnya: berbeda dengan demo Chapter Worker Threads, pekerja pool harus siap menerima tugas berikutnya.
2. utama.mjs baris 6-15 (constructor): pool lahir dengan membuat sejumlah new Worker(berkas) — dua thread di contoh ini. Penanda sibuk per anggota menjadi kunci penugasan; pola new URL(..., import.meta.url) menjaga lokasi berkas pekerja tetap benar dari direktori mana pun.
3. Baris 17-22 (jalankan): pemanggil mendapat Promise yang diselesaikan nanti oleh pekerja — tugas beserta promise-nya dimasukkan ke antrean lalu isiTugas() dipanggil segera.
4. Baris 24-41 (isiTugas): inti pool. Setiap anggota yang tidak sibuk dan setiap tugas yang masih mengantre dipertemukan: once("message") dipasang untuk satu tugas ini saja, tugas dikirim lewat postMessage(), dan saat hasil tiba penanda sibuk dilepas lalu isiTugas() dipanggil lagi — berantai seperti itu, antrean selalu terkuras.
5. Baris 34-38 (once("error")): kegagalan di pekerja menolak promise pemanggil alih-alih dibiarkan menggantung; pekerja dianggap siap menerima tugas berikutnya.
6. Baris 43-47 (tutup): terminate() menghentikan semua pekerja dan Promise.all menunggu semuanya benar-benar keluar — tanpa ini, thread yang hidup menahan proses utama tetap terjaga.
7. Baris 53-58 dan 63-68: empat tugas dijadwalkan hampir bersamaan; dua pekerja membaginya dua-dua. Hasil di-sort berdasarkan id sebelum dicetak karena siapa yang selesai lebih dulu bergantung penjadwalan OS — satu-satunya bagian yang tidak deterministik sengaja tidak dicetak mentah-mentah.

Pool vs Cluster, Sekali Lagi #

Kedua pola ini sering dipertukarkan padahal menjawab kebutuhan berbeda, dan tabel Chapter Clustering telah merangkumnya. Aturan praktisnya: kalau tujuannya memperbanyak kapasitas melayani request, gunakan cluster — tiap worker adalah aplikasi utuh. Kalau tujuannya mempercepat atau mengisolasi pekerjaan tertentu (antrean kompresi, hash massal, parsing besar) di dalam satu aplikasi, gunakan pool — pekerjanya bukan aplikasi, melainkan karyawan spesialis yang dilengkapi satu jenis tugas.

Kombinasi dua-duanya pun lazim: aplikasi berjalan sebagai cluster, dan setiap prosesnya memiliki pool kecil untuk pekerjaan CPU berat. Dengan begitu beban I/O dibagi antar proses dan beban CPU dibagi antar thread — dua sumbu skalabilitas yang saling melengkapi.

Ada satu langkah pengujian terakhir sebelum semua rekayasa ini berlayar ke produksi: seberapa banyak beban yang sesungguhnya mampu ditangani aplikasi kita, dan pada titik berapa ia mulai menolak layanan? Untuk itu kita perlu menghujani aplikasi dengan permintaan secara terkendali — topik penutup Section ini, Chapter Load Testing.