go-concurrency-patterns

name	go-concurrency-patterns
description	Master Go concurrency with goroutines, channels, sync primitives, and context. Use when building concurrent Go applications, implementing worker pools, or debugging race conditions.

Go Concurrency Patterns

Production patterns for Go concurrency including goroutines, channels, synchronization primitives, and context management.

When to Use This Skill

• Building concurrent Go applications
• Implementing worker pools and pipelines
• Managing goroutine lifecycles
• Using channels for communication
• Debugging race conditions
• Implementing graceful shutdown

Core Concepts

1. Go Concurrency Primitives

Primitive	Purpose
goroutine	Lightweight concurrent execution
channel	Communication between goroutines
select	Multiplex channel operations
sync.Mutex	Mutual exclusion
sync.WaitGroup	Wait for goroutines to complete
context.Context	Cancellation and deadlines

2. Go Concurrency Mantra

Don't communicate by sharing memory;
share memory by communicating.

Quick Start

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    results := make(chan string, 10)
    var wg sync.WaitGroup

    // Spawn workers
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go worker(ctx, i, results, &wg)
    }

    // Close results when done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for result := range results {
        fmt.Println(result)
    }
}

func worker(ctx context.Context, id int, results chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()

    select {
    case <-ctx.Done():
        return
    case results <- fmt.Sprintf("Worker %d done", id):
    }
}

Patterns

Pattern 1: Worker Pool

package main

import (
    "context"
    "fmt"
    "sync"
)

type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID int
    Output string
    Err   error
}

func WorkerPool(ctx context.Context, numWorkers int, jobs <-chan Job) <-chan Result {
    results := make(chan Result, len(jobs))

    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for job := range jobs {
                select {
                case <-ctx.Done():
                    return
                default:
                    result := processJob(job)
                    results <- result
                }
            }
        }(i)
    }

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

    return results
}

func processJob(job Job) Result {
    
     Result{
        JobID:  job.ID,
        Output: fmt.Sprintf(, job.Data),
    }
}


 {
    ctx, cancel := context.WithCancel(context.Background())
     cancel()

    jobs := ( Job, )

    
      {
         i := ; i < ; i++ {
            jobs <- Job{ID: i, Data: fmt.Sprintf(, i)}
        }
        (jobs)
    }()

    
    results := WorkerPool(ctx, , jobs)

     result :=  results {
        fmt.Printf(, result)
    }
}

Pattern 2: Fan-Out/Fan-In Pipeline

package main

import (
    "context"
    "sync"
)

// Stage 1: Generate numbers
func generate(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case <-ctx.Done():
                return
            case out <- n:
            }
        }
    }()
    return out
}

// Stage 2: Square numbers (can run multiple instances)
func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
     out
}


 <-  {
     wg sync.WaitGroup
    out := ( )

    
    output :=  {
         wg.Done()
         n :=  c {
             {
             <-ctx.Done():
                
             out <- n:
            }
        }
    }

    wg.Add((cs))
     _, c :=  cs {
         output(c)
    }

    
      {
        wg.Wait()
        (out)
    }()

     out
}

 {
    ctx, cancel := context.WithCancel(context.Background())
     cancel()

    
    in := generate(ctx, , , , , , , , , , )

    
    c1 := square(ctx, in)
    c2 := square(ctx, in)
    c3 := square(ctx, in)

    
     result :=  merge(ctx, c1, c2, c3) {
        fmt.Println(result)
    }
}

Pattern 3: Bounded Concurrency with Semaphore

package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/semaphore"
    "sync"
)

type RateLimitedWorker struct {
    sem *semaphore.Weighted
}

func NewRateLimitedWorker(maxConcurrent int64) *RateLimitedWorker {
    return &RateLimitedWorker{
        sem: semaphore.NewWeighted(maxConcurrent),
    }
}

func (w *RateLimitedWorker) Do(ctx context.Context, tasks []func() error) []error {
    var (
        wg     sync.WaitGroup
        mu     sync.Mutex
        errors []error
    )

    for _, task := range tasks {
        // Acquire semaphore (blocks if at limit)
        if err := w.sem.Acquire(ctx, 1); err != nil {
            return []error{err}
        }

        wg.Add(1)
        go func(t func() error) {
            defer wg.Done()
            defer w.sem.Release(1)

            if err := t(); err != nil {
                mu.Lock()
                errors = append(errors, err)
                mu.Unlock()
            }
        }(task)
    }

    wg.Wait()
    return errors
}

// Alternative: Channel-based semaphore
 Semaphore  {}

 Semaphore {
     ( {}, n)
}

 Acquire() {
    s <- {}{}
}

 Release() {
    <-s
}

Pattern 4: Graceful Shutdown

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

type Server struct {
    shutdown chan struct{}
    wg       sync.WaitGroup
}

func NewServer() *Server {
    return &Server{
        shutdown: make(chan struct{}),
    }
}

func (s *Server) Start(ctx context.Context) {
    // Start workers
    for i := 0; i < 5; i++ {
        s.wg.Add(1)
        go s.worker(ctx, i)
    }
}

func (s *Server) worker(ctx context.Context, id int) {
    defer s.wg.Done()
    defer fmt.Printf("Worker %d stopped\n", id)

    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            // Cleanup
            fmt.Printf("Worker %d cleaning up...\n", id)
            time.Sleep(500 * time.Millisecond) // Simulated cleanup
            return
        case <-ticker.C:
            fmt.Printf("Worker %d working...\n", id)
        }
    }
}

 Shutdown(timeout time.Duration) {
    
    (s.shutdown)

    
    done := ( {})
      {
        s.wg.Wait()
        (done)
    }()

     {
     <-done:
        fmt.Println()
     <-time.After(timeout):
        fmt.Println()
    }
}

 {
    
    ctx, cancel := context.WithCancel(context.Background())

    sigCh := ( os.Signal, )
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

    server := NewServer()
    server.Start(ctx)

    
    sig := <-sigCh
    fmt.Printf(, sig)

    
    cancel()

    
    server.Shutdown( * time.Second)
}

Pattern 5: Error Group with Cancellation

package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
    "net/http"
)

func fetchAllURLs(ctx context.Context, urls []string) ([]string, error) {
    g, ctx := errgroup.WithContext(ctx)

    results := make([]string, len(urls))

    for i, url := range urls {
        i, url := i, url // Capture loop variables

        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
            if err != nil {
                return fmt.Errorf("creating request for %s: %w", url, err)
            }

            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return fmt.Errorf("fetching %s: %w", url, err)
            }
            defer resp.Body.Close()

            results[i] = fmt.Sprintf("%s: %d", url, resp.StatusCode)
            return nil
        })
    }

    // Wait for all goroutines to complete or one to fail
    if err := g.Wait(); err != nil {
        return nil, err // First error cancels all others
    }

    return results, nil
}


 ([], ) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(limit) 

    results := ([], (urls))
     mu sync.Mutex

     i, url :=  urls {
        i, url := i, url

        g.Go(  {
            result, err := fetchURL(ctx, url)
             err !=  {
                 err
            }

            mu.Lock()
            results[i] = result
            mu.Unlock()
             
        })
    }

     err := g.Wait(); err !=  {
         , err
    }

     results, 
}

Pattern 6: Concurrent Map with sync.Map

package main

import (
    "sync"
)

// For frequent reads, infrequent writes
type Cache struct {
    m sync.Map
}

func (c *Cache) Get(key string) (interface{}, bool) {
    return c.m.Load(key)
}

func (c *Cache) Set(key string, value interface{}) {
    c.m.Store(key, value)
}

func (c *Cache) GetOrSet(key string, value interface{}) (interface{}, bool) {
    return c.m.LoadOrStore(key, value)
}

func (c *Cache) Delete(key string) {
    c.m.Delete(key)
}

// For write-heavy workloads, use sharded map
type ShardedMap struct {
    shards    []*shard
    numShards int
}

type shard struct {
    sync.RWMutex
    data map[string]interface{}
}

func NewShardedMap(numShards int) *ShardedMap {
    m := &ShardedMap{
        shards:    make([]*shard, numShards),
        numShards: numShards,
    }
    for i := range m.shards {
        m.shards[i] = &shard{data: make(map[]{})}
    }
     m
}

 getShard(key ) *shard {
    
    h := 
     _, c :=  key {
        h = *h + (c)
    }
     m.shards[h%m.numShards]
}

 Get(key ) ({}, ) {
    shard := m.getShard(key)
    shard.RLock()
     shard.RUnlock()
    v, ok := shard.data[key]
     v, ok
}

 Set(key , value {}) {
    shard := m.getShard(key)
    shard.Lock()
     shard.Unlock()
    shard.data[key] = value
}

Pattern 7: Select with Timeout and Default

func selectPatterns() {
    ch := make(chan int)

    // Timeout pattern
    select {
    case v := <-ch:
        fmt.Println("Received:", v)
    case <-time.After(time.Second):
        fmt.Println("Timeout!")
    }

    // Non-blocking send/receive
    select {
    case ch <- 42:
        fmt.Println("Sent")
    default:
        fmt.Println("Channel full, skipping")
    }

    // Priority select (check high priority first)
    highPriority := make(chan int)
    lowPriority := make(chan int)

    for {
        select {
        case msg := <-highPriority:
            fmt.Println("High priority:", msg)
        default:
            select {
            case msg := <-highPriority:
                fmt.Println("High priority:", msg)
            case msg := <-lowPriority:
                fmt.Println("Low priority:", msg)
            }
        }
    }
}

Race Detection

# Run tests with race detector
go test -race ./...

# Build with race detector
go build -race .

# Run with race detector
go run -race main.go

Best Practices

Do's

• Use context - For cancellation and deadlines
• Close channels - From sender side only
• Use errgroup - For concurrent operations with errors
• Buffer channels - When you know the count
• Prefer channels - Over mutexes when possible

Don'ts

• Don't leak goroutines - Always have exit path
• Don't close from receiver - Causes panic
• Don't use shared memory - Unless necessary
• Don't ignore context cancellation - Check ctx.Done()
• Don't use time.Sleep for sync - Use proper primitives

Resources

• Go Concurrency Patterns
• Effective Go - Concurrency
• Go by Example - Goroutines