go / intermediate
Snippet
Implementing a Worker Pool for Concurrent Task Processing
A worker pool manages a concurrent set of goroutines executing tasks from a shared queue. This pattern prevents uncontrolled goroutine creation and resource exhaustion. We use a sync.WaitGroup to track worker completion, closing the job channel when finished to signal workers to exit, and then safely closing the results channel once all workers are done.
snippet.go
go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package mainimport ("fmt""sync")func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {defer wg.Done()for job := range jobs {results <- job * 2}}func runWorkerPool(numWorkers int, numJobs int) []int {jobs := make(chan int, numJobs)results := make(chan int, numJobs)var wg sync.WaitGroupfor w := 1; w <= numWorkers; w++ {wg.Add(1)go worker(w, jobs, results, &wg)}for j := 1; j <= numJobs; j++ {jobs <- j}close(jobs)wg.Wait()close(results)var output []intfor res := range results {output = append(output, res)}return output}
Breakdown
1
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup)
Defines a worker that reads from the receive-only jobs channel and writes to the send-only results channel.
2
wg.Add(1)
Increments the WaitGroup counter for each worker goroutine being spawned.
3
close(jobs)
Closes the jobs channel, indicating no more work will be sent; this lets the range jobs loop terminate.
4
wg.Wait()
Blocks until all worker goroutines call wg.Done(), ensuring all tasks are processed before concluding.