Advanced Go Programming
Concurrency Patterns
Scaling with Concurrency Patterns
You already know how to launch a goroutine and send data over a channel. Now, let's combine those building blocks into powerful patterns that can make your applications faster and more efficient. We'll look at two common patterns for managing concurrent work: Fan-out/Fan-in and the Worker Pool.
The Fan-Out, Fan-In Pattern
Imagine you have a big, divisible job, like processing a thousand images. Doing them one by one is slow. The Fan-out/Fan-in pattern offers a better way.
Fan-out: A main goroutine (the producer) takes the large job and breaks it into smaller pieces. It then launches multiple new goroutines, distributing one piece of the job to each. This is like a manager handing out tasks to a team. Each new goroutine works on its small task in parallel.
Fan-in: Once the worker goroutines finish their tasks, they need to send their results back. The fan-in part involves collecting all these results from multiple channels into a single channel. This is like the manager collecting the finished reports from each team member.
Let's see this in action. We'll create a pipeline that generates numbers, squares them concurrently, and then prints the results.
package main
import (
"fmt"
"sync"
)
// producer generates numbers and sends them to a channel.
func producer(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
// square is our worker. It reads from an input channel,
// squares the number, and sends it to an output channel.
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
// merge (the fan-in part) collects results from multiple
// channels into one.
func merge(cs ...<-chan int) <-chan int {
var wg sync.WaitGroup
out := make(chan int)
// Start a goroutine for each input channel.
output := func(c <-chan int) {
for n := range c {
out <- n
}
wg.Done()
}
wg.Add(len(cs))
for _, c := range cs {
go output(c)
}
// Start a goroutine to close 'out' once all the
// output goroutines are done.
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
// 1. Set up the pipeline.
in := producer(1, 2, 3, 4, 5, 6)
// 2. Fan-out: Distribute the work to two 'square' goroutines.
c1 := square(in)
c2 := square(in)
// 3. Fan-in: Merge the results from c1 and c2.
for n := range merge(c1, c2) {
fmt.Println(n)
}
}
In this example, the producer creates our initial work. The main function fans out the work from the in channel to two concurrent square goroutines. The merge function then fans in the results. Notice that we use a sync.WaitGroup to ensure we only close the final output channel after all the incoming channels have been fully processed.
The Worker Pool Pattern
Sometimes you don't want to spin up a new goroutine for every single task, especially if you have thousands of them. Creating and destroying goroutines has a small cost, and having too many can consume a lot of memory. A worker pool is the solution.
The pattern is simple: start a fixed number of long-lived worker goroutines. These workers all listen for tasks on the same input channel. When a task arrives, one of the available workers picks it up, processes it, and sends the result to an output channel.
A worker pool is ideal for controlling concurrency and managing resource usage. You limit the number of tasks running simultaneously, preventing your system from being overwhelmed.
Think of a coffee shop with a fixed number of baristas (the workers). Customers (the tasks) line up. As soon as a barista is free, they take the next order from the line. The number of baristas doesn't change, no matter how long the line gets. This keeps the workflow steady and predictable.
Here's how you might implement a worker pool in Go. We'll create a pool of workers to process a list of jobs.
package main
import (
"fmt"
"time"
)
// The worker function. It receives jobs from the 'jobs' channel
// and sends results to the 'results' channel.
func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs {
fmt.Println("worker", id, "started job", j)
time.Sleep(time.Second) // Simulate work
fmt.Println("worker", id, "finished job", j)
results <- j * 2
}
}
func main() {
const numJobs = 5
jobs := make(chan int, numJobs)
results := make(chan int, numJobs)
// Start up 3 workers. They will block initially because
// there are no jobs yet.
for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
}
// Send 5 jobs and then close the channel to indicate
// that's all the work we have.
for j := 1; j <= numJobs; j++ {
jobs <- j
}
close(jobs)
// Finally, we collect all the results.
// This also ensures that the program waits for all
// workers to finish.
for a := 1; a <= numJobs; a++ {
<-results
}
}
In this setup, we create three workers. Even though we have five jobs, a maximum of three will ever run at the same time. Once a worker finishes a job, it automatically becomes available to pick up the next one from the jobs channel. Closing the jobs channel is crucial; it signals to the workers (who are in a for...range loop) that there's no more work, allowing them to terminate gracefully.
Ready to check your understanding?
What is the primary goal of the Fan-out/Fan-in concurrency pattern?
What is a primary advantage of using a Worker Pool instead of launching a new goroutine for every single task?
These patterns are foundational for building robust, high-performance services in Go. By controlling how you distribute and process concurrent tasks, you can take full advantage of modern hardware.