Building production systems in Go means thinking carefully about concurrency. Channels form the backbone of this approach, offering a powerful way to coordinate goroutines. Over time, I’ve found certain patterns emerge as particularly valuable for creating robust, maintainable systems.
Worker pools stand out as one of the most practical patterns. They allow you to control resource consumption while maximizing throughput. Here’s how I typically implement them:
type Job struct {
ID int
Processed bool
Error error
}
func processWorker(ctx context.Context, jobs <-chan Job, results chan<- Job) {
for {
select {
case job, ok := <-jobs:
if !ok {
return
}
// Simulate work
time.Sleep(50 * time.Millisecond)
job.Processed = true
results <- job
case <-ctx.Done():
return
}
}
}
func main() {
jobs := make(chan Job, 100)
results := make(chan Job, 100)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
// Start worker pool
for i := 0; i < 10; i++ {
go processWorker(ctx, jobs, results)
}
// Send jobs
go func() {
for i := 0; i < 50; i++ {
jobs <- Job{ID: i}
}
close(jobs)
}()
// Collect results
for i := 0; i < 50; i++ {
result := <-results
fmt.Printf("Processed job %d\n", result.ID)
}
}
The context integration here proves crucial. It ensures that if the operation takes too long, everything cleans up properly. I’ve learned through experience that without proper context handling, goroutines can linger and cause resource leaks.
Fan-out and fan-in patterns help manage multiple data streams. When dealing with several concurrent data sources, merging them into a single channel simplifies consumption:
func mergeChannels(ctx context.Context, channels ...<-chan int) <-chan int {
var wg sync.WaitGroup
merged := make(chan int)
output := func(c <-chan int) {
defer wg.Done()
for n := range c {
select {
case merged <- n:
case <-ctx.Done():
return
}
}
}
wg.Add(len(channels))
for _, c := range channels {
go output(c)
}
go func() {
wg.Wait()
close(merged)
}()
return merged
}
This pattern shines when processing data from multiple databases or APIs concurrently. The context ensures we can cancel the entire operation if needed.
Timeout handling deserves special attention. In production systems, operations must complete within reasonable timeframes:
func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
result := make(chan string, 1)
errChan := make(chan error, 1)
go func() {
// Simulate network call
time.Sleep(time.Millisecond * 200)
result <- "response data"
}()
select {
case res := <-result:
return res, nil
case err := <-errChan:
return "", err
case <-time.After(timeout):
return "", fmt.Errorf("request timed out after %v", timeout)
}
}
I’ve found that setting appropriate timeouts prevents cascading failures. The key is choosing timeout values that reflect actual service level objectives.
Channel ownership clarifies code structure. I follow a simple rule: the goroutine that creates a channel should be responsible for closing it. This prevents panic situations and makes code easier to reason about.
Buffered channels require careful sizing. Too small, and you lose throughput. Too large, and you risk excessive memory usage. I typically start with small buffers and increase based on monitoring:
func processWithBufferedChannel() {
// Buffer size based on expected throughput
dataChan := make(chan DataPoint, 1000)
processed := make(chan Result, 1000)
go processor(dataChan, processed)
// Monitor channel capacity
go func() {
for range time.Tick(time.Second) {
fmt.Printf("Buffer usage: %d/%d\n",
len(dataChan), cap(dataChan))
}
}()
}
Non-blocking operations prove useful for monitoring and fallback scenarios. The select statement with default cases enables this:
func trySend(ch chan<- Data, value Data) bool {
select {
case ch <- value:
return true
default:
// Channel full, implement fallback
log.Println("Channel full, dropping data")
return false
}
}
Error handling in concurrent code requires discipline. I prefer using result types that encapsulate both data and error information:
type Result struct {
Value int
Error error
}
func processConcurrently(inputs []int) []Result {
results := make(chan Result, len(inputs))
var wg sync.WaitGroup
for _, input := range inputs {
wg.Add(1)
go func(x int) {
defer wg.Done()
val, err := processItem(x)
results <- Result{Value: val, Error: err}
}(input)
}
wg.Wait()
close(results)
var output []Result
for res := range results {
output = append(output, res)
}
return output
}
This approach ensures errors don’t get lost in concurrent processing. Each result carries complete information about the operation.
Composing these patterns enables complex workflows. I often build pipelines that combine multiple patterns:
func dataPipeline(ctx context.Context, source <-chan Data) <-chan Result {
// Stage 1: Validation
validated := validateStage(ctx, source)
// Stage 2: Processing
processed := processStage(ctx, validated)
// Stage 3: Enrichment
enriched := enrichStage(ctx, processed)
return enriched
}
Each stage can use different concurrency patterns based on its requirements. This modular approach makes systems easier to understand and maintain.
Monitoring channel performance helps identify bottlenecks. I track metrics like channel utilization, processing latency, and error rates. This data informs optimization decisions and helps set appropriate buffer sizes.
The true power of these patterns emerges when they work together. They provide a toolkit for building concurrent systems that are both efficient and maintainable. Each pattern solves specific problems, and combining them creates robust production systems.
Through experience, I’ve learned that simplicity often beats complexity. The most effective solutions use the simplest patterns that solve the problem. Over-engineering concurrent systems can create more problems than it solves.
These patterns form a foundation for building reliable concurrent systems in Go. They address common production challenges while maintaining code clarity and performance. The key is understanding when to apply each pattern and how they work together.
