Advanced Golang Development
Concurrency Patterns
Orchestrating Goroutines
You know how to start goroutines. The real challenge, however, isn't just starting them—it's managing them. How do you distribute work efficiently among many goroutines and gather the results in an orderly way? How do you prevent them from tripping over each other? The answer lies in established concurrency patterns.
Concurrency is about dealing with many things at once, while parallelism is about doing many things at once.
These patterns are like blueprints for common concurrent programming problems. They provide a structured way to handle complex workflows, making your code more efficient, scalable, and easier to understand. Let's explore some of the most powerful ones.
Fan-Out, Fan-In
Imagine you have a large number of tasks to complete, like processing thousands of images. Doing them one by one would be slow. The fan-out, fan-in pattern offers a much faster approach.
Fan-Out: You take a single stream of tasks (an input channel) and distribute the work across multiple goroutines. Each goroutine acts as a worker, pulling a task from the channel and processing it. This is like a manager handing out assignments to a team.
Fan-In: Once the workers complete their tasks, they send their results to a single output channel. This step funnels all the individual results back into one coherent stream, like the manager collecting all the finished assignments.
This pattern is incredibly useful for parallelizing work. You're not just running tasks concurrently; you're creating a structured data flow that scales with the number of workers you deploy.
A key part of implementing this pattern is managing goroutine lifecycles. We use a sync.WaitGroup to ensure the main program doesn't exit before all workers have finished processing and sent their results.
package main
import (
"fmt"
"sync"
"time"
)
// Worker function that receives tasks and sends results
func worker(id int, tasks <-chan int, results chan<- int) {
for task := range tasks {
fmt.Printf("Worker %d started job %d\n", id, task)
time.Sleep(time.Second) // Simulate work
results <- task * 2
fmt.Printf("Worker %d finished job %d\n", id, task)
}
}
func main() {
numTasks := 10
numWorkers := 3
tasks := make(chan int, numTasks)
results := make(chan int, numTasks)
// Fan-out: Start the workers
for w := 1; w <= numWorkers; w++ {
go worker(w, tasks, results)
}
// Feed tasks to the workers
for j := 1; j <= numTasks; j++ {
tasks <- j
}
close(tasks) // Close tasks channel when all tasks are sent
// Fan-in: Collect the results
for a := 1; a <= numTasks; a++ {
<-results
}
fmt.Println("All results collected.")
}
Pipelines and Worker Pools
We can extend the fan-out pattern to build more complex workflows. A pipeline is a series of stages connected by channels, where the output of one stage becomes the input for the next. Each stage is a goroutine (or group of goroutines) that performs a specific piece of work. This is like an assembly line for data.
For example, a pipeline might have three stages:
- Stage 1: Reads data from a source.
- Stage 2: Processes the data (e.g., filters it).
- Stage 3: Stores the processed data.
This pattern allows you to break down a complex task into smaller, manageable, and concurrent steps.
A worker pool is a fixed number of goroutines that are available to handle tasks. Instead of creating a new goroutine for every single task, which can be inefficient, you create a pool of workers at the start. Tasks are sent to a channel, and the idle workers in the pool pick them up.
This pattern is excellent for controlling resource consumption. It limits the number of concurrent goroutines, preventing your application from becoming overwhelmed when dealing with a massive number of tasks.
Here's how you might set up a pipeline where one of the stages is a worker pool.
package main
import (
"fmt"
"sync"
)
// First stage: generates numbers
func generator(done <-chan struct{}, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case out <- n:
case <-done:
return
}
}
}()
return out
}
// Second stage: a worker pool that squares numbers
func square(done <-chan struct{}, in <-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
const numWorkers = 3
wg.Add(numWorkers)
for i := 0; i < numWorkers; i++ {
go func() {
defer wg.Done()
for n := range in {
select {
case out <- n * n:
case <-done:
return
}
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
// A channel to signal cancellation
done := make(chan struct{})
defer close(done)
// Set up the pipeline
in := generator(done, 2, 4, 6, 8)
ch := square(done, in)
// Consume the output
for res := range ch {
fmt.Println(res)
}
}
Notice the done channel passed through the pipeline. This is a crucial technique for graceful shutdowns. By closing the done channel, we can signal all running goroutines in the pipeline to stop their work and exit, preventing leaks.
What is the primary purpose of the fan-out, fan-in concurrency pattern?
In a Go concurrency pipeline, how are the different stages typically connected?
By mastering these patterns, you can build sophisticated, high-performance applications that take full advantage of modern multi-core processors.