Worker Pool с graceful shutdown

Сложная

Concurrency

Условие задачи

Реализуй пул из 5 воркеров, которые обрабатывают задачи из канала. Каждая задача — строка. По сигналу завершения (например, через context или закрытие канала) воркеры должны завершиться корректно. Все задачи, поступившие до shutdown, должны быть обработаны.

package main

import (
    "fmt"
)

func worker(id int, jobs <-chan string, done chan<- int) {
    // TODO: получай задачи из jobs, завершись при его закрытии
}

func main() {
    jobs := make(chan string)
    done := make(chan int)

    // TODO: запусти 5 воркеров

    // TODO: отправь 20 задач в канал jobs

    // TODO: дождись завершения всех воркеров
}

package main

import (
    "fmt"
)

func worker(id int, jobs <-chan string, done chan<- int) {
    // TODO: получай задачи из jobs, завершись при его закрытии
}

func main() {
    jobs := make(chan string)
    done := make(chan int)

    // TODO: запусти 5 воркеров

    // TODO: отправь 20 задач в канал jobs

    // TODO: дождись завершения всех воркеров
}

Подсказка

- Когда канал jobs закрывается, range по нему завершается. - После завершения каждый воркер может отправлять в канал done или можно использовать WaitGroup.

Решение

Worker Pool — классический шаблон для масштабируемой обработки задач. Важно корректно завершать горутины и не терять задачи.

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, jobs <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d processing: %s
", id, job)
        time.Sleep(100 * time.Millisecond)
    }
}

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

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

    for j := 1; j <= 20; j++ {
        jobs <- fmt.Sprintf("job-%d", j)
    }
    close(jobs)

    wg.Wait()
    fmt.Println("All jobs processed.")
}

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, jobs <-chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d processing: %s
", id, job)
        time.Sleep(100 * time.Millisecond)
    }
}

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

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

    for j := 1; j <= 20; j++ {
        jobs <- fmt.Sprintf("job-%d", j)
    }
    close(jobs)

    wg.Wait()
    fmt.Println("All jobs processed.")
}