Go Concurrency Patterns

Go's concurrency model, built on goroutines and channels, makes concurrent programming approachable. This guide covers essential patterns for production Go code.

Goroutines and Channels

// Basic goroutine
go func() {
    fmt.Println("Running in background")
}()

// Channel communication
ch := make(chan int, 10) // Buffered channel

go func() {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch) // Signal no more values
}()

for v := range ch {
    fmt.Println(v)
}

Worker Pool Pattern

func workerPool(jobs []Job, numWorkers int) []Result {
    jobCh := make(chan Job, len(jobs))
    resultCh := make(chan Result, len(jobs))

    // Start workers
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobCh {
                result := processJob(job)
                resultCh <- result
            }
        }()
    }

    // Send jobs
    for _, job := range jobs {
        jobCh <- job
    }
    close(jobCh)

    // Wait and close results
    go func() {
        wg.Wait()
        close(resultCh)
    }()

    // Collect results
    var results []Result
    for r := range resultCh {
        results = append(results, r)
    }
    return results
}

Fan-Out / Fan-In

func fanOut(input <-chan int, numWorkers int) []<-chan int {
    outputs := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        outputs[i] = process(input)
    }
    return outputs
}

func fanIn(channels ...<-chan int) <-chan int {
    merged := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                merged <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(merged)
    }()

    return merged
}

Context for Cancellation

func fetchWithTimeout(url string) ([]byte, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    return io.ReadAll(resp.Body)
}

// Cancellable server handler
func handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    result := make(chan string, 1)
    go func() {
        // Simulate slow work
        time.Sleep(2 * time.Second)
        result <- "data"
    }()

    select {
    case data := <-result:
        fmt.Fprintln(w, data)
    case <-ctx.Done():
        http.Error(w, "Request cancelled", 499)
    }
}

Select Statement

func prioritizedSelect(highPriority, lowPriority <-chan Message) {
    for {
        // Check high priority first
        select {
        case msg := <-highPriority:
            handleUrgent(msg)
            continue
        default:
        }

        // Then handle any available
        select {
        case msg := <-highPriority:
            handleUrgent(msg)
        case msg := <-lowPriority:
            handleNormal(msg)
        case <-time.After(100 * time.Millisecond):
            // Timeout: do housekeeping
            doHousekeeping()
        }
    }
}

sync.Mutex and sync.RWMutex

type SafeCounter struct {
    mu    sync.RWMutex
    count map[string]int
}

func (c *SafeCounter) Increment(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count[key]++
}

func (c *SafeCounter) Get(key string) int {
    c.mu.RLock() // Multiple readers allowed
    defer c.mu.RUnlock()
    return c.count[key]
}

errgroup for Error Propagation

import "golang.org/x/sync/errgroup"

func fetchAll(urls []string) ([][]byte, error) {
    g, ctx := errgroup.WithContext(context.Background())
    results := make([][]byte, len(urls))

    for i, url := range urls {
        i, url := i, url // Capture loop vars
        g.Go(func() error {
            data, err := fetchURL(ctx, url)
            if err != nil {
                return fmt.Errorf("fetching %s: %w", url, err)
            }
            results[i] = data
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Summary

Go concurrency best practices:

• Use goroutines liberally, they're cheap (~2KB stack)
• Channels for communication, mutexes for shared state
• Always handle goroutine lifecycle with WaitGroup or context
• Use errgroup for error propagation in concurrent code
• Close channels from the sender side only