[NODE JS] Worker Threads ile Çok İş Parçacıklı Çalışma

[Silverhand]

Türkiye'de ilk Mobil & PC Aynı Anda Metin2 Oyna. Triarchonline kalıcı ve uzun ömürlü yapısı ile 24 Temmuz'da açılıyor. | 1-99 Mobil Metin2 Triarch HEMEN TIKLA!

[NODE JS] Worker Threads ile Çok İş Parçacıklı Çalışma​

Node.js'in tek iş parçacıklı yapısı, I/O yoğun uygulamalar için mükemmeldir ama CPU yoğun işlemlerde darboğaz yaratır. İşte Worker Threads modülü tam bu sorunu çözmek için Node.js 10.5 ile tanıtıldı ve Node.js 12'den itibaren kararlı hale geldi. Ben de ilk kez bir görüntü işleme projesinde kullandım ve performans farkı inanılmazdı. Tek iş parçacığında 45 saniye süren bir işlem, dört worker thread ile 12 saniyeye düştü.

Worker Threads ile child_process.fork arasında temel bir fark vardır: worker thread'ler aynı süreç içinde çalışır ve belleği paylaşabilir. Fork ise tamamen ayrı bir süreç oluşturur. Bu, worker thread'lerin çok daha az bellek tüketmesi anlamına gelir. Bir fork yaklaşık 30-50 MB bellek harcarken, bir worker thread sadece 5-10 MB civarında bellek kullanır. Bu fark, özellikle çok sayıda paralel işlem çalıştırmanız gerektiğinde büyük önem kazanır.

Worker Threads modülünün en güçlü özelliklerinden biri SharedArrayBuffer desteğidir. Bu sayede ana thread ve worker thread'ler arasında veri kopyalamak yerine, aynı bellek bölgesini paylaşabilirsiniz. Büyük veri setleriyle çalışırken bu özellik hayat kurtarıcıdır. Ancak dikkatli olmanız gerekir çünkü paylaşılan belleğe eşzamanlı erişim yarış durumlarına yol açabilir. Bu sorunu Atomics API ile çözebilirsiniz.

Pratikte worker thread'leri şu senaryolarda kullanıyorum: büyük JSON dosyalarını ayrıştırma, kriptografik işlemler, veri sıkıştırma ve açma, matematiksel modelleme, makine öğrenmesi inference işlemleri. Tüm bunların ortak özelliği CPU'yu yoğun şekilde kullanmalarıdır.

Worker thread oluştururken iki yaklaşım vardır. Birincisi, ayrı bir dosya belirtmektir. İkincisi ise inline worker oluşturmaktır, yani worker kodunu doğrudan ana dosya içinde tanımlarsınız. Inline yaklaşım küçük işler için pratiktir ama büyük projelerde ayrı dosya kullanmanızı öneririm. Kod düzeni ve bakım kolaylığı açısından çok daha iyidir.

MessagePort ve MessageChannel kavramları da worker thread'lerin iletişim altyapısını oluşturur. Varsayılan olarak her worker, parentPort üzerinden ana thread ile konuşur. Ama isterseniz kendi MessageChannel'larınızı oluşturup worker'lar arası doğrudan iletişim kurabilirsiniz. Bu, karmaşık pipeline yapılarında oldukça kullanışlıdır. Örneğin bir worker veriyi okuyor, ikinci worker işliyor, üçüncü worker sonucu yazıyor gibi bir akış tasarlayabilirsiniz.

workerData özelliği ile worker oluşturulurken başlangıç verisini iletebilirsiniz. Bu veri structuredClone ile kopyalanır, yani orijinal veri ile worker'daki kopya birbirinden bağımsızdır. Eğer kopyalama yerine aktarma (transfer) istiyorsanız, transferList parametresini kullanabilirsiniz. Transfer edilen nesne orijinal thread'de artık kullanılamaz hale gelir ama kopyalama maliyeti sıfırdır.

Worker thread'lerin bir diğer avantajı, hata izolasyonudur. Bir worker thread çökse bile ana thread etkilenmez. Worker'daki yakalanmamış hatalar error event'i olarak ana thread'e iletilir ve siz buna uygun şekilde tepki verebilirsiniz. Bu, uygulamanızın genel kararlılığını artırır.

// ==============================
// Worker Threads Temel Kullanımı
// ==============================
const {
    Worker,
    isMainThread,
    parentPort,
    workerData,
    threadId
} = require('worker_threads');

if (isMainThread) {
    // Ana thread kodu
    console.log('Ana thread çalışıyor, Thread ID:', threadId);

    // Worker oluşturma
    const worker = new Worker(__filename, {
        workerData: {
            dizi: [5, 12, 8, 130, 44, 97, 3, 21, 56, 78],
            islem: 'sirala'
        }
    });

    worker.on('message', (sonuc) => {
        console.log('Worker sonucu:', sonuc);
    });

    worker.on('error', (hata) => {
        console.error('Worker hatası:', hata.message);
    });

    worker.on('exit', (kod) => {
        console.log('Worker çıkış kodu:', kod);
    });
} else {
    // Worker thread kodu
    console.log('Worker thread çalışıyor, Thread ID:', threadId);

    const { dizi, islem } = workerData;

    if (islem === 'sirala') {
        // Bubble sort ile sıralama (CPU yoğun simülasyonu)
        const siralanmis = [...dizi];
        for (let i = 0; i < siralanmis.length; i++) {
            for (let j = 0; j < siralanmis.length - i - 1; j++) {
                if (siralanmis[j] > siralanmis[j + 1]) {
                    const gecici = siralanmis[j];
                    siralanmis[j] = siralanmis[j + 1];
                    siralanmis[j + 1] = gecici;
                }
            }
        }
        parentPort.postMessage({
            durum: 'tamamlandı',
            siralanmisDizi: siralanmis
        });
    }
}

// ==============================
// Worker Havuzu (Thread Pool)
// ==============================
class WorkerHavuzu {
    constructor(workerDosyasi, havuzBoyutu) {
        this.workerDosyasi = workerDosyasi;
        this.havuzBoyutu = havuzBoyutu;
        this.calışanlar = [];
        this.bosCalısanlar = [];
        this.gorevKuyrugu = [];

        // Havuzu başlat
        for (let i = 0; i < havuzBoyutu; i++) {
            this.calısanEkle();
        }
    }

    calısanEkle() {
        const worker = new Worker(this.workerDosyasi);

        worker.on('message', (sonuc) => {
            // İşi bitince worker'ı serbest bırak
            const gorevCallback = worker._aktifGorevCallback;
            worker._aktifGorevCallback = null;
            this.bosCalısanlar.push(worker);

            if (gorevCallback) {
                gorevCallback(null, sonuc);
            }

            // Kuyrukta bekleyen görev varsa çalıştır
            this.siradakiGorevi();
        });

        worker.on('error', (hata) => {
            const gorevCallback = worker._aktifGorevCallback;
            if (gorevCallback) {
                gorevCallback(hata, null);
            }
        });

        this.calışanlar.push(worker);
        this.bosCalısanlar.push(worker);
    }

    gorevCalistir(veri) {
        return new Promise((resolve, reject) => {
            const gorev = {
                veri,
                callback: (hata, sonuc) => {
                    if (hata) reject(hata);
                    else resolve(sonuc);
                }
            };

            if (this.bosCalısanlar.length > 0) {
                const worker = this.bosCalısanlar.pop();
                worker._aktifGorevCallback = gorev.callback;
                worker.postMessage(gorev.veri);
            } else {
                this.gorevKuyrugu.push(gorev);
            }
        });
    }

    siradakiGorevi() {
        if (this.gorevKuyrugu.length > 0 && this.bosCalısanlar.length > 0) {
            const gorev = this.gorevKuyrugu.shift();
            const worker = this.bosCalısanlar.pop();
            worker._aktifGorevCallback = gorev.callback;
            worker.postMessage(gorev.veri);
        }
    }

    async kapat() {
        const kapatmaSözleri = this.calışanlar.map(
            (worker) => worker.terminate()
        );
        await Promise.all(kapatmaSözleri);
        console.log('Tüm worker threadler kapatıldı.');
    }
}

// Kullanım örneği
async function calistir() {
    const havuz = new WorkerHavuzu('./hesaplama-worker.js', 4);

    const gorevler = [
        havuz.gorevCalistir({ sayi: 100000 }),
        havuz.gorevCalistir({ sayi: 200000 }),
        havuz.gorevCalistir({ sayi: 150000 }),
        havuz.gorevCalistir({ sayi: 300000 }),
        havuz.gorevCalistir({ sayi: 250000 })
    ];

    const sonuclar = await Promise.all(gorevler);
    console.log('Tüm görevler tamamlandı:', sonuclar);

    await havuz.kapat();
}

calistir().catch(console.error);

Worker Threads, Node.js'te CPU yoğun işlemleri yönetmenin en verimli yoludur. fork'a kıyasla çok daha az bellek tüketir ve SharedArrayBuffer sayesinde veri paylaşımı kolaylaşır. Ancak her durumda worker thread kullanmak doğru değildir; I/O yoğun işlemler için Node.js'in doğal asenkron yapısı zaten yeterlidir. Worker thread'leri sadece gerçekten CPU'yu tıkayan işlemler için tercih edin.

 

[Best Studio]

Emeğiniz ve katkınız için teşekkür ederiz!
Konunuz gerekli incelemelerden geçerek onaylandı. Topluluğumuza katkılarınızın devamını bekliyoruz.
İyi forumlar!