Node.js Cluster ve Worker Threads: Çoklu İşlem Yönetimi

Node.js tek iş parçacıklı (single-threaded) bir runtime olmasına rağmen, modern sunucuların çok çekirdekli yapısından tam olarak yararlanmak için Cluster modülü ve Worker Threads API'si sunmaktadır. Cluster modülü, uygulamanızın birden fazla process olarak çalışmasını sağlayarak I/O yoğun işlemler için throughput'u artırır. Worker Threads ise CPU yoğun işlemleri ana iş parçacığını bloke etmeden paralel olarak çalıştırmanızı sağlar. Bu rehberde her iki yaklaşımı detaylı örneklerle inceleyeceğiz.

1. Cluster Modülü: Çok Çekirdekli Ölçekleme

Cluster modülü, master process'in birden fazla worker process fork etmesini sağlar. Her worker bağımsız bir Node.js instance'ıdır ve aynı port'u paylaşabilir. İşletim sistemi, gelen bağlantıları worker'lar arasında otomatik olarak dağıtır (round-robin).

const cluster = require('cluster');
const http = require('http');
const os = require('os');

const numCPUs = os.cpus().length;

if (cluster.isPrimary) {
  console.log(`Primary process ${process.pid} çalışıyor`);
  console.log(`${numCPUs} worker başlatılıyor...`);

  // Worker'ları fork et
  for (let i = 0; i < numCPUs; i++) {
    const worker = cluster.fork();

    worker.on('message', (msg) => {
      console.log(`Worker ${worker.process.pid}: ${JSON.stringify(msg)}`);
    });
  }

  // Worker çöktüğünde yenisini başlat
  cluster.on('exit', (worker, code, signal) => {
    console.warn(`Worker ${worker.process.pid} sonlandı (code: ${code}, signal: ${signal})`);

    if (code !== 0 && !worker.exitedAfterDisconnect) {
      console.log('Yeni worker başlatılıyor...');
      cluster.fork();
    }
  });

  // Graceful shutdown
  process.on('SIGTERM', () => {
    console.log('SIGTERM alındı, worker\'lar kapatılıyor...');
    for (const id in cluster.workers) {
      cluster.workers[id].process.kill('SIGTERM');
    }
  });

  // Worker durumu izleme
  setInterval(() => {
    const workerCount = Object.keys(cluster.workers).length;
    console.log(`Aktif worker sayısı: ${workerCount}`);
  }, 30000);

} else {
  // Worker process
  const server = http.createServer((req, res) => {
    // CPU-bound simülasyonu
    if (req.url === '/heavy') {
      let sum = 0;
      for (let i = 0; i < 1e7; i++) sum += i;
      res.end(`Worker ${process.pid} - Sonuç: ${sum}`);
    } else {
      res.end(`Worker ${process.pid} - Merhaba!`);
    }
  });

  server.listen(3000, () => {
    console.log(`Worker ${process.pid} başlatıldı`);
  });

  // Worker graceful shutdown
  process.on('SIGTERM', () => {
    server.close(() => {
      console.log(`Worker ${process.pid} kapatıldı`);
      process.exit(0);
    });
  });
}

Bilgi: Production ortamında cluster yönetimi için PM2 kullanmanız önerilir. PM2, otomatik restart, load balancing, log yönetimi, monitoring ve zero-downtime deployment gibi özellikler sunar. pm2 start app.js -i max komutu ile tüm CPU çekirdeklerini kullanabilirsiniz.

2. Worker Threads: CPU Yoğun İşlemler

Worker Threads, aynı process içinde birden fazla thread çalıştırmanızı sağlar. Cluster'dan farklı olarak, worker thread'ler SharedArrayBuffer ile bellek paylaşabilir ve MessageChannel ile verimli iletişim kurabilir. Görüntü işleme, veri analizi ve kriptografi gibi CPU yoğun işlemler için idealdir.

// main.js - Ana thread
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const os = require('os');

// Worker thread oluşturma
function runWorker(workerFile, data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(workerFile, { workerData: data });

    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker ${code} koduyla sonlandı`));
      }
    });
  });
}

// CPU-yoğun işlemi worker'a delege et
async function processImage(imagePath) {
  const result = await runWorker('./workers/image-processor.js', {
    imagePath,
    operations: ['resize', 'compress', 'watermark']
  });
  return result;
}

// Worker Pool - Birden fazla worker yönetimi
class WorkerPool {
  constructor(workerFile, poolSize = os.cpus().length) {
    this.workerFile = workerFile;
    this.poolSize = poolSize;
    this.workers = [];
    this.queue = [];
    this.activeWorkers = 0;

    this._initPool();
  }

  _initPool() {
    for (let i = 0; i < this.poolSize; i++) {
      this._createWorker();
    }
  }

  _createWorker() {
    const worker = new Worker(this.workerFile);
    worker.busy = false;

    worker.on('message', (result) => {
      worker.busy = false;
      this.activeWorkers--;
      worker._resolve(result);
      this._processQueue();
    });

    worker.on('error', (error) => {
      worker.busy = false;
      this.activeWorkers--;
      worker._reject(error);
      this._processQueue();
    });

    this.workers.push(worker);
  }

  execute(data) {
    return new Promise((resolve, reject) => {
      this.queue.push({ data, resolve, reject });
      this._processQueue();
    });
  }

  _processQueue() {
    if (this.queue.length === 0) return;

    const availableWorker = this.workers.find(w => !w.busy);
    if (!availableWorker) return;

    const { data, resolve, reject } = this.queue.shift();
    availableWorker.busy = true;
    availableWorker._resolve = resolve;
    availableWorker._reject = reject;
    this.activeWorkers++;
    availableWorker.postMessage(data);
  }

  async destroy() {
    for (const worker of this.workers) {
      await worker.terminate();
    }
  }
}

// Kullanım
const pool = new WorkerPool('./workers/compute.js', 4);
const results = await Promise.all([
  pool.execute({ task: 'fibonacci', n: 45 }),
  pool.execute({ task: 'prime', n: 100000 }),
  pool.execute({ task: 'sort', data: largeArray })
]);

3. Worker Thread İçi: İşlem Mantığı

Worker thread dosyası, ana thread'den gelen mesajları dinler, işlemi gerçekleştirir ve sonucu geri gönderir. workerData ile başlangıç verisi, parentPort.postMessage ile sonuç iletilir.

// workers/compute.js
const { parentPort, workerData } = require('worker_threads');

// workerData ile başlatma verisi alan worker
if (workerData) {
  const result = processTask(workerData);
  parentPort.postMessage(result);
}

// Message bazlı worker (pool için)
parentPort.on('message', (task) => {
  try {
    let result;

    switch (task.task) {
      case 'fibonacci':
        result = fibonacci(task.n);
        break;
      case 'prime':
        result = findPrimes(task.n);
        break;
      case 'hash':
        result = hashData(task.data);
        break;
      default:
        throw new Error(`Bilinmeyen görev: ${task.task}`);
    }

    parentPort.postMessage({ success: true, result });
  } catch (error) {
    parentPort.postMessage({ success: false, error: error.message });
  }
});

function fibonacci(n) {
  if (n <= 1) return n;
  let a = 0, b = 1;
  for (let i = 2; i <= n; i++) {
    [a, b] = [b, a + b];
  }
  return b;
}

function findPrimes(limit) {
  const sieve = new Uint8Array(limit + 1);
  const primes = [];

  for (let i = 2; i <= limit; i++) {
    if (!sieve[i]) {
      primes.push(i);
      for (let j = i * i; j <= limit; j += i) {
        sieve[j] = 1;
      }
    }
  }

  return { count: primes.length, last: primes[primes.length - 1] };
}

Uyarı: Worker thread'ler arasında veri transferinde structuredClone kullanılır, bu da büyük objeler için maliyetli olabilir. Büyük veri setlerini paylaşmak için SharedArrayBuffer ve Atomics kullanın; ancak bu durumda thread safety'ye dikkat etmeniz gerekir.

İpucu: Cluster ve Worker Threads farklı amaçlara hizmet eder. I/O yoğun uygulamalarda (web sunucusu, API) Cluster kullanarak yatay ölçekleme yapın. CPU yoğun tekil işlemler (görüntü işleme, veri analizi) için Worker Threads tercih edin. İkisini birlikte de kullanabilirsiniz.

Sonuç

Node.js'in Cluster modülü ve Worker Threads API'si, tek iş parçacıklı yapının sınırlamalarını aşmanızı sağlar. Cluster ile çok çekirdekli sunucuların tam kapasitesinden yararlanabilir, Worker Threads ile CPU yoğun işlemleri ana thread'i bloke etmeden çalıştırabilirsiniz. Worker Pool pattern'i ile thread yönetimini optimize edebilir ve production ortamında PM2 gibi araçlarla profesyonel süreç yönetimi sağlayabilirsiniz.

Node.js Cluster ve Worker Threads: Çoklu İşlem Yönetimi