Node.js worker_threads Kullanımı: Worker, postMessage ve SharedArrayBuffer

worker_threads ne zaman gerekir?

Node.js JavaScript kodunuzu tek bir olay döngüsünde çalıştırır. Bir istek işleyicisi yarım saniye süren bir döngüye girerse, o yarım saniye boyunca başka hiçbir isteğe cevap verilemez. Node.js Learn rehberindeki "Don't Block the Event Loop" yazısı, karmaşık bir işi olay döngüsünden çıkarmanın iki yolunu anlatır: işi küçük parçalara bölmek ya da ayrı bir hesaplama havuzuna devretmek. worker_threads ikinci yolun JavaScript ile yapılan en doğrudan biçimidir. Olay döngüsünün fazlarını nextTick ve setImmediate yazısında, dil düzeyindeki modeli JavaScript event loop yazısında anlattık.

Belge, worker'ların child_process ve cluster modüllerinden farklı olarak bellek paylaşabildiğini vurgular. Ayrı süreçlerle çalışmak için child_process yazısına, üretimde çok çekirdekli sunucu için cluster ve PM2 yazısına bakabilirsiniz.

İlk worker: workerData, parentPort ve olaylar

Bir worker'ı new Worker(dosya, seçenekler) ile başlatırsınız. Dosya yolu mutlak bir yol, ./ veya ../ ile başlayan ve çalışma klasörüne göre çözülen bir yol ya da file: veya data: protokollü bir URL olmalıdır. ESM dosyalarında en pratik yöntem, aynı dosyayı new URL(import.meta.url) ile worker olarak açmak ve hangi tarafta olduğunuzu isMainThread ile anlamaktır. Başlangıç verisi workerData seçeneğiyle gider; worker içinde aynı adlı dışa aktarımdan okunur. Worker sonucu parentPort.postMessage() ile yollar, ana thread bunu message olayıyla alır.

import { Worker, isMainThread, parentPort, workerData, threadId } from 'node:worker_threads';

function fib(n) {
  return n < 2 ? n : fib(n - 1) + fib(n - 2);
}

if (isMainThread) {
  const isci = new Worker(new URL(import.meta.url), { workerData: { n: 32 } });
  isci.on('online', () => console.log('worker çalışmaya başladı'));
  isci.on('message', (sonuc) => console.log('sonuç:', sonuc));
  isci.on('error', (err) => console.error('worker hatası:', err));
  isci.on('exit', (kod) => console.log('worker çıktı, kod:', kod));
  console.log('ana thread beklemiyor, threadId:', threadId);
} else {
  // Burası ayrı bir thread: ana thread'in olay döngüsünü bloklamaz
  parentPort.postMessage({ n: workerData.n, fib: fib(workerData.n), threadId });
}
// çıktı: ana thread beklemiyor, threadId: 0
// çıktı: worker çalışmaya başladı
// çıktı: sonuç: { n: 32, fib: 2178309, threadId: 1 }
// çıktı: worker çıktı, kod: 0

Çıktının sırası öğreticidir: ana thread worker'ı başlattıktan hemen sonra kendi satırını yazdı, yani hesaplamayı beklemedi. online olayı worker JavaScript çalıştırmaya başladığında gelir. Belgeye göre worker'ın gönderdiği bütün mesajlar exit olayından önce yayılır ve exit her worker'ın yaydığı son olaydır. Ana thread'in threadId değeri 0, ilk worker'ınki 1 oldu; belge bu değerin bir süreç içindeki her worker için benzersiz olduğunu söyler.

Belgedeki uyarıyı aklınızda tutun: örnekteki gibi her görev için yeni bir worker açmak, pratikte worker oluşturma maliyetinin kazancı aşmasına yol açabilir. Gerçek uygulamada bir worker havuzu kurulur ve görevler hazır worker'lara dağıtılır; aşağıda bunun basit bir biçimini göreceğiz.

Worker ortamı ana thread'den nasıl farklıdır?

Worker içinde Node.js API'lerinin çoğu kullanılabilir, ama belge bazı önemli farkları listeler. Bunları bilmemek, "ana thread'de çalışan kod worker'da neden farklı davranıyor" sorusunun en yaygın cevabıdır:

• process.exit() bütün programı değil yalnızca o thread'i durdurur; process.abort() kullanılamaz.
• process.chdir() ve kullanıcı ya da grup kimliğini değiştiren process metotları yoktur.
• process.env, ana thread'deki ortam değişkenlerinin bir kopyasıdır; bir thread'deki değişiklik diğerinde görünmez. env: SHARE_ENV verirseniz paylaşılır. Windows'ta bu kopya, ana thread'den farklı olarak büyük küçük harfe duyarlı çalışır.
• process.title değiştirilemez, sinyaller process.on() üzerinden worker'a iletilmez.
• worker.terminate() çağrıldığında çalışma herhangi bir noktada durabilir.
• Ebeveyn süreçten gelen IPC kanalları erişilebilir değildir, trace_events modülü desteklenmez.
• Worker içinden yeni worker oluşturmak mümkündür.

Worker'ın stdout ve stderr çıktısı varsayılan olarak ana thread'in akışlarına aktarılır. Belgenin "Notes" bölümü bunun bir yan etkisini anlatır: bu aktarım mesajlaşma ile yapıldığından, ana thread senkron bir döngüyle bloklanmışsa worker'ın console.log çıktısı da ana thread serbest kalana kadar görünmez.

Veri nasıl taşınır: kopyalama, transfer ve klonlama sınırları

postMessage() ile gönderilen değer HTML yapılandırılmış klonlama algoritmasıyla kopyalanır. Belge JSON'a göre farkları sıralar: döngüsel referanslar, Map, Set, RegExp, BigInt ve typed array'ler desteklenir. Buna karşılık fonksiyonlar klonlanamaz; sınıf örnekleri prototiplerini, özel alanlarını ve erişimcilerini kaybederek düz nesneye dönüşür; Buffer karşı tarafta düz bir Uint8Array olarak okunur. URL gibi bazı yerleşik nesneler ise hiç klonlanamaz ve DataCloneError fırlatır.

Büyük ikili verilerde kopyalama maliyetinden kaçmak için ikinci parametre olan transferList kullanılır. Bir ArrayBuffer transfer edildiğinde bellek kopyalanmaz, alıcı tarafa taşınır ve gönderen tarafta kullanılamaz hale gelir. Aşağıdaki örnek bu davranışları bir MessageChannel üzerinde gösteriyor; aynı kurallar worker.postMessage() ve parentPort.postMessage() için de geçerlidir.

import { MessageChannel } from 'node:worker_threads';

const { port1, port2 } = new MessageChannel();

class Sepet {
  #gizli = 1;
  constructor() { this.urunler = ['kalem']; }
  get adet() { return this.urunler.length; }
}

port1.on('message', (veri) => {
  console.log(veri);
});

// 1) Yapılandırılmış klonlama: Map korunur, sınıf örneği düz nesneye döner
port2.postMessage({ harita: new Map([['a', 1]]), sepet: new Sepet() });

// 2) Kopya ve transfer farkı
const dizi = new Uint8Array([1, 2, 3, 4]);
port2.postMessage(dizi);                 // kopya gider
port2.postMessage(dizi, [dizi.buffer]);  // bellek taşınır
console.log('transferden sonra byteLength:', dizi.byteLength);

// 3) Fonksiyon klonlanamaz
try {
  port2.postMessage({ f() {} });
} catch (err) {
  console.log(err.name, '-', err.message);
}

setTimeout(() => port1.close(), 50);
// çıktı: transferden sonra byteLength: 0
// çıktı: DataCloneError - f() {} could not be cloned.
// çıktı: { harita: Map(1) { 'a' => 1 }, sepet: { urunler: [ 'kalem' ] } }
// çıktı: Uint8Array(4) [ 1, 2, 3, 4 ]
// çıktı: Uint8Array(4) [ 1, 2, 3, 4 ]

Senkron çıktılar (transferden sonra byteLength değerinin 0 olması ve DataCloneError) mesajlardan önce yazıldı, çünkü mesajlar olay döngüsünün sonraki turlarında teslim edilir. Sınıf örneğinden geriye yalnızca urunler alanı kaldı; #gizli alanı ve adet erişimcisi aktarılmadı. Belge mesaj nesnesinin gönderildiği anda klonlandığını, bu yüzden sonradan değiştirmenin karşı tarafı etkilemediğini de belirtir.

SharedArrayBuffer ve Atomics ile ortak bellek

Transfer belleği taşır; SharedArrayBuffer ise aynı belleği birden fazla thread'in aynı anda görmesini sağlar. Belgeye göre bir değer SharedArrayBuffer içeriyorsa bu bellek iki taraftan da erişilebilir olur ve transferList içine konamaz. Ortak belleğe birden fazla thread yazıyorsa sıradan a[0]++ işlemi yarış durumuna açıktır; bunun yerine Atomics nesnesinin bölünmez işlemlerini kullanırsınız.

import { Worker, isMainThread, workerData } from 'node:worker_threads';

if (isMainThread) {
  // SharedArrayBuffer: tüm thread'ler aynı belleği görür
  const ortak = new Int32Array(new SharedArrayBuffer(4));
  const isciler = [];
  for (let i = 0; i < 4; i++) {
    isciler.push(new Worker(new URL(import.meta.url), { workerData: ortak }));
  }
  await Promise.all(isciler.map((w) => new Promise((r) => w.on('exit', r))));
  console.log('sayaç:', Atomics.load(ortak, 0));
} else {
  for (let i = 0; i < 1000; i++) {
    Atomics.add(workerData, 0, 1); // yarış durumu olmadan artırır
  }
}
// çıktı: sayaç: 4000

Dört worker her biri 1000 kez artırma yaptı ve sonuç tam olarak 4000 çıktı. Ortak bellek güçlüdür ama tasarımı zorlaştırır; çoğu uygulamada mesajlaşma ile sonuç döndürmek daha anlaşılır bir yoldur. Ortak belleği gerçekten büyük, sık güncellenen sayısal verilerde düşünün.

Hatalar, sonlandırma ve bellek sınırları

Worker içinde yakalanmayan bir istisna oluşursa belgeye göre worker sonlandırılır ve ana thread'de error olayı yayılır; örnekte ardından gelen exit olayının kodu 1 oldu. Bir worker'ı dışarıdan durdurmak için worker.terminate() çağrılır; bu fonksiyon v12.5.0'dan beri bir Promise döndürür ve exit olayı yayıldığında çıkış koduyla çözülür. Belge, sonlandırılan bir worker için çıkış kodunun 1 olduğunu söyler.

import { Worker } from 'node:worker_threads';

// 1) Worker içinde yakalanmayan hata: 'error' olayı, ardından exit kodu 1
const hatali = new Worker("throw new TypeError('geçersiz girdi')", { eval: true });
hatali.on('error', (err) => console.log('error olayı:', err.name, err.message));
hatali.on('exit', (kod) => console.log('hatalı worker exit:', kod));

// 2) Sonsuz döngüdeki worker'ı terminate() ile durdurmak
const sonsuz = new Worker('for (;;) {}', { eval: true });
setTimeout(async () => {
  const kod = await sonsuz.terminate();
  console.log('terminate sonucu:', kod);
}, 200);
// çıktı: error olayı: TypeError geçersiz girdi
// çıktı: hatalı worker exit: 1
// çıktı: terminate sonucu: 1

Örnekteki eval: true seçeneği, ilk argümanı dosya yolu yerine çalıştırılacak kod olarak yorumlatır; kısa denemeler için kullanışlıdır. Worker'ın kullanabileceği belleği resourceLimits ile sınırlayabilirsiniz: maxOldGenerationSizeMb, maxYoungGenerationSizeMb, codeRangeSizeMb ve stackSizeMb (varsayılan 4). Belgeye göre bu sınırlara ulaşan worker sonlandırılır, sınırlar yalnızca JS motorunu kapsar (ArrayBuffer gibi harici veriler sayılmaz) ve küresel bir bellek tükenmesinde süreç yine de çökebilir. Aşağıdaki örnek ayrıca iki yaygın kurulum hatasını da gösteriyor.

import { Worker } from 'node:worker_threads';

// Göreli yol ./ veya ../ ile başlamalı
try {
  new Worker('isci.mjs');
} catch (err) {
  console.log(err.code);
}

// workerData klonlanamayan bir değer içeremez
try {
  new Worker('./isci.mjs', { workerData: { hesapla: () => 1 } });
} catch (err) {
  console.log(err.name);
}

// Bellek sınırı: sınıra ulaşan worker sonlandırılır
const w = new Worker('const a = []; for (;;) a.push(new Array(1e5).fill(1));', {
  eval: true,
  resourceLimits: { maxOldGenerationSizeMb: 32 },
});
w.on('error', (err) => console.log('error:', err.code));
w.on('exit', (kod) => console.log('exit:', kod));
// çıktı: ERR_WORKER_PATH
// çıktı: DataCloneError
// çıktı: error: ERR_WORKER_OUT_OF_MEMORY
// çıktı: exit: 1

ERR_WORKER_PATH, göreli yolun ./ ile başlamadığını söyler. DataCloneError, workerData içine bir fonksiyon koymaya çalıştığımız için geldi; belge klonlanamayan bir değer verildiğinde hata fırlatıldığını açıkça yazar. Bellek sınırına çarpan worker ERR_WORKER_OUT_OF_MEMORY koduyla error olayı yaydı ve 1 koduyla çıktı. Hata kodlarının genel mantığını ERR_* hata kodları yazısında anlattık.

Kaynak temizliği için iki yardımcı daha vardır. worker.unref(), worker olay sistemindeki tek etkin tutamaç olduğunda programın çıkabilmesine izin verir. v24.2.0 ve v22.18.0 ile eklenen worker[Symbol.asyncDispose]() ise await using kapsamından çıkıldığında worker.terminate() çağırır.

Mini örnek: işi çekirdeklere bölen küçük bir worker havuzu

Şimdi parçaları birleştirelim. 2 milyonun altındaki asal sayıları sayan bir işi, makinedeki paralellik kadar (en fazla 8) parçaya bölüyor, her parçayı bir worker'a veriyor ve sonuçları Promise.all ile topluyoruz. os.availableParallelism() fonksiyonunu os modülü yazısında anlattık. Her worker için message, error ve sıfırdan farklı exit durumlarını tek bir Promise'e bağlayan yapı, belgenin giriş örneğindeki desenin aynısıdır.

import { Worker, isMainThread, parentPort, workerData } from 'node:worker_threads';
import { availableParallelism } from 'node:os';

function asalSay(bas, son) {
  let adet = 0;
  for (let n = Math.max(2, bas); n < son; n++) {
    let asal = true;
    for (let d = 2; d * d <= n; d++) {
      if (n % d === 0) { asal = false; break; }
    }
    if (asal) adet++;
  }
  return adet;
}

function isciCalistir(bas, son) {
  return new Promise((resolve, reject) => {
    const w = new Worker(new URL(import.meta.url), { workerData: { bas, son } });
    w.once('message', resolve);
    w.once('error', reject);
    w.once('exit', (kod) => {
      if (kod !== 0) reject(new Error(`worker ${kod} koduyla durdu`));
    });
  });
}

if (isMainThread) {
  const UST = 2_000_000;
  const parca = Math.min(availableParallelism(), 8); // çekirdek kadar, en fazla 8 worker
  const boy = Math.ceil(UST / parca);
  const isler = [];
  for (let i = 0; i < parca; i++) {
    isler.push(isciCalistir(i * boy, Math.min(UST, (i + 1) * boy)));
  }
  const sonuclar = await Promise.all(isler);
  console.log('2 milyonun altındaki asal sayısı:', sonuclar.reduce((a, b) => a + b, 0));
} else {
  parentPort.postMessage(asalSay(workerData.bas, workerData.son));
}
// çıktı: 2 milyonun altındaki asal sayısı: 148933

Bu basit yapı her çağrıda yeni worker açtığı için tek seferlik toplu işlerde yeterlidir. Sürekli görev alan bir sunucuda ise worker'ları bir kez başlatıp boşta olanlara görev veren kalıcı bir havuz kurmak gerekir. Belge bu durumda görevler ile sonuçları arasındaki bağı tanı araçlarına bildirmek için AsyncResource API'sini kullanmanızı önerir ve async_hooks belgesindeki örnek havuz uygulamasına yönlendirir; bu API'yi AsyncLocalStorage yazısında ele aldık.

MessageChannel ve diğer iletişim araçları

Her worker, ana thread ile arasında hazır bir MessagePort çiftiyle gelir; worker.postMessage() ve parentPort bu çifti kullanır. Belge, ilgi alanlarını ayırmak için kendi kanallarınızı kurmanızı teşvik eder: bir MessageChannel oluşturur, portlardan birini mevcut kanal üzerinden transferList ile karşı tarafa gönderirsiniz. Böylece aynı ebeveynin iki worker'ı birbirleriyle doğrudan konuşabilir. Aynı ada sahip kanala bağlı herkese yayın yapmak için BroadcastChannel sınıfı da vardır.

Belgede worker_threads.postMessageToThread() fonksiyonu Stability: 1.1 - Active development, worker_threads.locks ise Stability: 1 - Experimental olarak işaretlidir. Bunlar değişebileceği için bu yazının kapsamı dışında bıraktık; üretim kodunda kullanmadan önce kendi sürümünüzün belgesine bakın.

Sık yapılan hatalar

• G/Ç işini worker'a taşımak: Dosya okumak ya da HTTP isteği yapmak zaten asenkrondur; worker eklemek yalnızca ek maliyet getirir.
• Her görev için yeni worker açmak: Kısa görevlerde oluşturma maliyeti kazancı geçebilir. Bir havuz kullanın.
• Göreli yolu ./ olmadan vermek: new Worker('isci.mjs') çağrısı ERR_WORKER_PATH hatası verir.
• Fonksiyon ya da sınıf örneği göndermek: Fonksiyon DataCloneError verir, sınıf örneği prototipsiz düz nesneye dönüşür.
• Transfer edilen veriyi tekrar kullanmak: Transferden sonra gönderen taraftaki görünüm boştur; aynı ArrayBuffer'ı paylaşan bütün görünümler etkilenir.
• error ve exit olaylarını dinlememek: Worker içinde çöken bir görevin sonucu hiç gelmez ve bekleyen Promise asılı kalır.
• Ön yükleme betiğinden worker başlatmak: Belgeye göre -r ile yüklenen betikler, execArgv açıkça verilmedikçe her worker'da da çalışır; koşulsuz worker açan bir ön yükleme betiği uygulama çökene kadar worker üretir.
• İzin modelinde execArgv vermek: Belge, açıkça verilen bir execArgv dizisinin (boş dizi dahil) ebeveynin bayraklarını devralmayı kestiğini ve bu yüzden --permission ile --allow-* izinlerini düşürebileceğini belirtir. Ayrıntı izin modeli yazısında.

Sık Sorulan Sorular

worker_threads kodumu otomatik olarak hızlandırır mı?

Hayır. Worker yalnızca ana thread'i bloklayan CPU yoğun JavaScript işini başka bir thread'e taşır. G/Ç bekleyen kodda kazanç beklemeyin; ayrıca veriyi kopyalamanın ve worker başlatmanın da bir maliyeti vardır. Kazancı, kendi işinizi ölçerek doğrulayın.

Worker içinde require ve import kullanabilir miyim?

Evet. Worker dosyası normal bir CommonJS ya da ES modülüdür ve Node.js API'lerinin çoğu kullanılabilir. Belgenin listelediği farklar (process.exit davranışı, env kopyası, sinyaller, chdir) dışında ana thread'deki gibi yazarsınız. Modül sistemleri için CommonJS ve ESM yazısına bakın.

Kaç worker açmalıyım?

Belge sabit bir sayı vermez. Yaygın yaklaşım, CPU yoğun işler için os.availableParallelism() değerini üst sınır olarak almak ve ana thread'in de çalıştığını hesaba katmaktır. Aynı sunucuda cluster ya da başka süreçler çalışıyorsa sayıyı buna göre azaltın.

Worker'lar arasında veritabanı bağlantısı paylaşabilir miyim?

Bağlantı nesneleri genellikle klonlanabilir değildir; her worker kendi bağlantısını açmalıdır. Belgenin listesine göre mesajla taşınabilen yerel nesneler MessagePort, FileHandle, KeyObject, TCP net.Socket ve net.Server gibi sınırlı türlerdir.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Node.js öğrenme rehberi · child_process exec ve spawn · Node.js event loop fazları · Node.js performans optimizasyonu · cluster ve PM2