Golang Concurrency

Go Concurrency Patterns

Goroutines and WaitGroup

func processItems(items []Item) {
    var wg sync.WaitGroup
    for _, item := range items {
        wg.Add(1)
        go func(it Item) {
            defer wg.Done()
            process(it)
        }(item)  // pass item as argument to avoid closure capture bug
    }
    wg.Wait()
}

Worker Pool

func workerPool(ctx context.Context, jobs <-chan Job, results chan<- Result, workers int) {
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case job, ok := <-jobs:
                    if !ok { return }
                    results <- process(job)
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    go func() {
        wg.Wait()
        close(results)
    }()
}

// Usage
jobs := make(chan Job, 100)
results := make(chan Result, 100)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

workerPool(ctx, jobs, results, 10)

for _, job := range allJobs { jobs <- job }
close(jobs)

for result := range results { handle(result) }

Context for Cancellation

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

    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("fetch %s: %w", url, err)
    }
    defer resp.Body.Close()
    return io.ReadAll(resp.Body)
}

// Propagate context through call stack
func (s *Service) Handle(ctx context.Context, req Request) (Response, error) {
    user, err := s.repo.FindUser(ctx, req.UserID)  // ctx passed down
    if err != nil { return Response{}, err }

    data, err := s.external.Fetch(ctx, user.ExternalID)
    return Response{Data: data}, err
}

Channels Patterns

// Pipeline
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums { out <- n }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in { out <- n * n }
    }()
    return out
}

// Fan-out + fan-in
func fanOut(in <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    for i := range channels {
        channels[i] = square(in)
    }
    return channels
}

func merge(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    output := func(c <-chan int) {
        defer wg.Done()
        for n := range c { out <- n }
    }
    wg.Add(len(channels))
    for _, c := range channels { go output(c) }
    go func() { wg.Wait(); close(out) }()
    return out
}

sync.Mutex and RWMutex

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

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

func (c *SafeCounter) Value(key string) int {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.v[key]
}

sync.Once for Lazy Initialization

type Config struct {
    once sync.Once
    data map[string]string
}

func (c *Config) Get(key string) string {
    c.once.Do(func() {
        c.data = loadConfigFromDisk()
    })
    return c.data[key]
}

Rate Limiting

// Token bucket using time.Ticker
func rateLimitedWorker(ctx context.Context, requests <-chan Request, rps int) {
    ticker := time.NewTicker(time.Second / time.Duration(rps))
    defer ticker.Stop()

    for req := range requests {
        select {
        case <-ticker.C:
            go handle(req)
        case <-ctx.Done():
            return
        }
    }
}

// golang.org/x/time/rate — production rate limiter
limiter := rate.NewLimiter(rate.Every(time.Second/10), 5)  // 10 rps, burst 5
if err := limiter.Wait(ctx); err != nil { return err }

Common Anti-Patterns

• Goroutine leak — always ensure goroutines can exit; use context cancellation
• Closure capturing loop variable — pass loop var as argument to goroutine
• Unbuffered channel + no receiver — deadlock; size channels or use select with default
• Mutex protecting wrong scope — lock/unlock around the exact shared state, not the whole function
• Using time.Sleep for synchronization — use channels, WaitGroup, or condition variables