بناء Worker Pool

بناء Worker Pool — Build a Worker Pool

الـ worker pool نمط عملي عندما لديك مهام كثيرة، ولا تريد تشغيل goroutine غير محدود لكل مهمة. تحدد عدد العمال، وترسل لهم المهام عبر channel.

تخيل أن لديك ألف صورة تحتاج معالجة، أو ألف بريد تحتاج إرساله. تشغيل goroutine لكل مهمة قد يبدو سهلاً، لكنه قد يضغط الذاكرة أو الاتصال الخارجي أو قاعدة البيانات. worker pool يعطيك حداً واضحاً: عندي عدد محدد من العمال، وكل عامل يأخذ مهمة من القناة وينفذها. بهذه الطريقة تتحكم في التوازي بدلاً من تركه يكبر بلا حدود.

في هذا الدرس سنركز على الشكل الأساسي للنمط: قناة للمهام، دالة worker، وsync.WaitGroup للانتظار. لا نضيف نتائج أو أخطاء حتى لا تختلط الأفكار. عندما تفهم هذا الأساس، تستطيع لاحقاً إضافة قناة نتائج، سياق إلغاء context، أو حد زمني حسب الحاجة.

الخطوة 1: مهمة وعامل واحد

main.go

package main

import "fmt"

type Job struct {
    ID int
}

func worker(jobs <-chan Job) {
    for job := range jobs {
        fmt.Printf("معالجة مهمة %d\n", job.ID)
    }
}

func main() {
    jobs := make(chan Job)

    go worker(jobs)

    jobs <- Job{ID: 1}
    jobs <- Job{ID: 2}
    close(jobs)
}

Output:

العامل يقرأ من القناة باستخدام for job := range jobs. هذه الحلقة تستمر حتى تغلق القناة. لذلك المسؤولية المهمة على المرسل: عندما تنتهي من إرسال كل المهام، أغلق القناة حتى يعرف العامل أن العمل انتهى. لا يغلق المستقبل القناة عادة؛ من يرسل هو من يعرف متى لا توجد قيم أخرى.

الخطوة 2: أضف WaitGroup

بدون انتظار، قد ينتهي main قبل انتهاء العمل. sync.WaitGroup يجعل الإنهاء واضحاً.

main.go

package main

import (
    "fmt"
    "sync"
)

type Job struct {
    ID int
}

func worker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("عامل %d أنجز مهمة %d\n", id, job.ID)
    }
}

func main() {
    jobs := make(chan Job)
    var wg sync.WaitGroup

    wg.Add(1)
    go worker(1, jobs, &wg)

    for i := 1; i <= 3; i++ {
        jobs <- Job{ID: i}
    }
    close(jobs)

    wg.Wait()
    fmt.Println("كل المهام انتهت")
}

Output:

استخدمنا defer wg.Done() في بداية العامل حتى يتم إنقاص العداد عند خروج الدالة مهما كان سبب الخروج. القاعدة العملية: كل wg.Add(1) قبل goroutine يجب أن يقابله wg.Done() داخلها. إذا نسيت Done سيبقى Wait ينتظر للأبد. وإذا استدعيت Done أكثر من اللازم سيحدث panic.

الخطوة 3: أكثر من عامل

الآن نبدأ ثلاثة عمال يستقبلون من نفس القناة.

main.go

package main

import (
    "fmt"
    "sync"
)

type Job struct {
    ID int
}

func worker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("عامل %d أنجز مهمة %d\n", id, job.ID)
    }
}

func main() {
    jobs := make(chan Job)
    var wg sync.WaitGroup

    for w := 1; w <= 3; w++ {
        wg.Add(1)
        go worker(w, jobs, &wg)
    }

    for i := 1; i <= 6; i++ {
        jobs <- Job{ID: i}
    }
    close(jobs)

    wg.Wait()
    fmt.Println("كل المهام انتهت")
}

Output:

قد تلاحظ أن ترتيب الرسائل يختلف بين تشغيل وآخر. هذا طبيعي في التزامن. لا تبنِ صحة البرنامج على أن العامل 1 سيأخذ المهمة 1 دائماً. ما يهم هو أن كل مهمة تُرسل مرة واحدة، وأن العمال ينهون عملهم، وأن main ينتظر النهاية. لذلك في التحدي سنطبع ملخصاً ثابتاً بدلاً من الاعتماد على ترتيب تنفيذ العمال.

قواعد أمان للنمط

ابدأ العمال قبل إرسال المهام إذا كانت القناة غير buffered، لأن الإرسال سينتظر مستقبلاً جاهزاً. أغلق قناة المهام بعد آخر إرسال، وليس قبل ذلك. لا تستدع wg.Wait() قبل إغلاق القناة، لأن العمال سيبقون ينتظرون قيماً جديدة. ولا تجعل العامل يغلق قناة المهام؛ العامل لا يعرف هل يوجد مرسل آخر.

عندما ترى deadlock في مثل هذا الكود، اسأل ثلاثة أسئلة: هل يوجد goroutine يستقبل من القناة؟ هل أغلقت القناة بعد الإرسال؟ هل عدد Add يطابق عدد العمال؟ غالباً ستجد الخلل في واحد من هذه المواضع.

تمرين موجه

أكمل worker pool بمعالجة 5 مهام. لا تعتمد على ترتيب العمال في الإخراج؛ في التزامن قد يختلف الترتيب. هذا التحدي يطلب ملخصاً مستقراً فقط.

تحدي — Challenge

package main

import (
    "fmt"
    "sync"
)

type Job struct {
    ID int
}

func worker(jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for range jobs {
        // محاكاة معالجة المهمة
    }
}

func main() {
    jobs := make(chan Job)
    var wg sync.WaitGroup

    // ابدأ عاملين هنا

    for i := 1; i <= 5; i++ {
        jobs <- Job{ID: i}
    }
    fmt.Println("تم إرسال 5 مهام")
    close(jobs)

    wg.Wait()
    fmt.Println("كل المهام انتهت")
}

خلاصة

worker pool ليس مجرد تمرين على goroutine وchannel. هو قرار تصميم يحمي البرنامج من التوازي غير المحدود. القناة تمثل قائمة العمل، العمال يمثلون القدرة المتاحة، وWaitGroup يمثل وعداً واضحاً بأن main لن ينتهي قبل اكتمال العمال. عندما تفهم هذه الحدود، تستطيع بناء معالجة خلفية، queues بسيطة، أو import jobs بثقة أكبر.