Concurrency Patterns

🎭 The Restaurant Kitchen: Mastering Go Concurrency Patterns

Imagine you’re running a busy restaurant kitchen. Orders flood in. Cooks work together. Food flows through stations. This is exactly how Go handles concurrent work!

🍳 The Big Picture: Why Concurrency Patterns Matter

Think of your Go program as a restaurant kitchen:

• Goroutines = Kitchen workers
• Channels = The counter where dishes are passed
• Patterns = The systems that keep everything organized

Without good patterns, your kitchen becomes chaos. Orders get lost. Workers bump into each other. Customers wait forever.

Let’s learn the patterns that make kitchens (and Go programs) run smoothly!

👷 Worker Pools: Your Team of Cooks

What’s a Worker Pool?

Imagine you have 100 orders but only 5 cooks. You don’t hire 100 cooks! Instead, your 5 cooks take orders from a queue, one at a time.

That’s a Worker Pool.

func worker(id int, jobs <-chan int,
    results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing %d\n",
            id, job)
        results <- job * 2
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    // Start 3 workers
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // Send 9 jobs
    for j := 1; j <= 9; j++ {
        jobs <- j
    }
    close(jobs)

    // Collect results
    for a := 1; a <= 9; a++ {
        <-results
    }
}

Why Use Worker Pools?

Simple rule: Fixed workers, unlimited jobs!

🌊 Fan-in and Fan-out: Splitting and Merging Work

Fan-Out: One Source, Many Workers

Picture a chef cutting vegetables. One pile of veggies goes to multiple prep cooks. Each cook works on a portion.

func fanOut(input <-chan int,
    workers int) []<-chan int {

    channels := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        channels[i] = processWork(input)
    }
    return channels
}

func processWork(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n // Square the number
        }
        close(out)
    }()
    return out
}

Fan-In: Many Sources, One Destination

Now those prep cooks finish their work. All the chopped veggies go into ONE big bowl.

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
}

graph TD
    A["📥 Input"] --> B["Worker 1"]
    A --> C["Worker 2"]
    A --> D["Worker 3"]
    B --> E["📤 Output"]
    C --> E
    D --> E

🔗 Pipeline Pattern: The Assembly Line

What’s a Pipeline?

Think of making a burger:

1. Station 1: Toast the bun
2. Station 2: Grill the patty
3. Station 3: Add toppings
4. Station 4: Wrap it up

Each station does ONE job, then passes it forward!

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

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

// Stage 3: Add 10
func addTen(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n + 10
        }
        close(out)
    }()
    return out
}

func main() {
    // Connect the pipeline!
    nums := generate(2, 3, 4)
    squared := square(nums)
    result := addTen(squared)

    for v := range result {
        fmt.Println(v) // 14, 19, 26
    }
}

Pipeline Power:

• Each stage runs concurrently
• Data flows like water through pipes
• Easy to add or remove stages

🚿 Goroutine Leaks: The Dripping Faucet

What’s a Goroutine Leak?

Imagine a faucet that won’t turn off. Water keeps dripping. Your water bill goes up. Eventually, something floods.

A goroutine leak is a goroutine that never stops. It sits there, using memory, doing nothing useful.

Common Causes

1. Blocked Channel (No Reader)

// 🚨 LEAK! Nobody reads from this channel
func leak() {
    ch := make(chan int)
    go func() {
        ch <- 42 // Blocks forever!
    }()
    // Function returns, goroutine stuck
}

2. Infinite Loop Without Exit

// 🚨 LEAK! No way to stop
func leak() {
    go func() {
        for {
            doSomething()
        }
    }()
}

The Fix: Use Context or Done Channel

func noLeak(ctx context.Context) {
    go func() {
        for {
            select {
            case <-ctx.Done():
                return // Clean exit! ✅
            default:
                doSomething()
            }
        }
    }()
}

Golden Rule: Every goroutine needs an exit door! 🚪

🚦 Rate Limiting: Don’t Overwhelm the Kitchen

What’s Rate Limiting?

Your restaurant can only serve 10 customers per minute. More than that? The kitchen can’t keep up. Quality drops.

Rate limiting controls HOW FAST work happens.

Simple Rate Limiter with Ticker

func rateLimited() {
    // Allow 1 request per 200ms
    limiter := time.NewTicker(200 * time.Millisecond)
    defer limiter.Stop()

    requests := []int{1, 2, 3, 4, 5}

    for _, req := range requests {
        <-limiter.C // Wait for permission
        fmt.Printf("Processing request %d\n", req)
    }
}

Burst Rate Limiter

Sometimes you allow a small burst, then slow down:

func burstLimiter() {
    // Allow burst of 3
    burst := make(chan time.Time, 3)
    for i := 0; i < 3; i++ {
        burst <- time.Now()
    }

    // Refill 1 per 200ms
    go func() {
        for t := range time.Tick(200 * time.Millisecond) {
            burst <- t
        }
    }()

    for i := 1; i <= 5; i++ {
        <-burst
        fmt.Printf("Request %d at %s\n",
            i, time.Now().Format("15:04:05.000"))
    }
}

🛑 Graceful Shutdown: Closing Time Done Right

What’s Graceful Shutdown?

It’s 10 PM. Restaurant closes. Do you:

• A) Kick everyone out mid-meal? 😱
• B) Stop new orders, let current diners finish, then close? 😊

Option B is graceful shutdown.

The Pattern

func gracefulServer() {
    jobs := make(chan int, 10)
    done := make(chan bool)

    // Worker
    go func() {
        for job := range jobs {
            fmt.Printf("Processing %d\n", job)
            time.Sleep(500 * time.Millisecond)
        }
        done <- true
    }()

    // Send some jobs
    for i := 1; i <= 5; i++ {
        jobs <- i
    }

    // Graceful shutdown
    close(jobs)  // Stop accepting new jobs
    <-done       // Wait for worker to finish
    fmt.Println("Shutdown complete!")
}

With WaitGroup for Multiple Workers

func gracefulMultiWorker() {
    var wg sync.WaitGroup
    jobs := make(chan int)

    // Start 3 workers
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                fmt.Printf("Worker %d: job %d\n",
                    id, job)
            }
        }(i)
    }

    // Send jobs
    for j := 1; j <= 9; j++ {
        jobs <- j
    }

    close(jobs)  // Signal: no more jobs
    wg.Wait()    // Wait for all workers
    fmt.Println("All workers finished!")
}

📡 Signal Handling: Listening for the Fire Alarm

What’s Signal Handling?

Your restaurant has fire alarms, closing bells, emergency buttons. When they ring, you need to respond!

In Go, signals are messages from the operating system:

• SIGINT = Someone pressed Ctrl+C
• SIGTERM = System asking you to stop
• SIGHUP = Terminal closed

Catching Signals

func signalHandler() {
    // Create signal channel
    sigs := make(chan os.Signal, 1)
    done := make(chan bool, 1)

    // Listen for these signals
    signal.Notify(sigs, syscall.SIGINT,
        syscall.SIGTERM)

    go func() {
        sig := <-sigs
        fmt.Printf("\nReceived: %s\n", sig)
        done <- true
    }()

    fmt.Println("Running... Press Ctrl+C")
    <-done
    fmt.Println("Shutting down gracefully")
}

Complete Example: Server with Signal Handling

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

    // Handle signals
    sigs := make(chan os.Signal, 1)
    signal.Notify(sigs, syscall.SIGINT,
        syscall.SIGTERM)

    go func() {
        <-sigs
        fmt.Println("\nShutdown signal received")
        cancel() // Cancel context
    }()

    // Start worker with context
    go worker(ctx)

    // Wait for context cancellation
    <-ctx.Done()

    // Cleanup time
    time.Sleep(time.Second)
    fmt.Println("Cleanup complete. Goodbye!")
}

func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("Worker stopping...")
            return
        default:
            fmt.Println("Working...")
            time.Sleep(500 * time.Millisecond)
        }
    }
}

🎯 Pattern Quick Reference

🌟 The Secret Recipe

graph TD
    A["🍳 Worker Pools"] --> B["Control Resources"]
    C["🌊 Fan-in/Fan-out"] --> D["Distribute Work"]
    E["🔗 Pipeline"] --> F["Sequential Processing"]
    G["🚿 Leak Prevention"] --> H["Clean Goroutines"]
    I["🚦 Rate Limiting"] --> J["Control Speed"]
    K["🛑 Graceful Shutdown"] --> L["Clean Exit"]
    M["📡 Signal Handling"] --> N["Respond to OS"]

Remember: A well-organized kitchen serves great food. A well-organized Go program handles concurrency beautifully!

🎉 You Did It!

You now understand the 7 essential concurrency patterns in Go:

1. ✅ Worker Pools - Fixed workers, unlimited jobs
2. ✅ Fan-Out - Spread work to many
3. ✅ Fan-In - Gather results from many
4. ✅ Pipeline - Chain processing stages
5. ✅ Goroutine Leaks - Always have an exit
6. ✅ Rate Limiting - Control the flow
7. ✅ Graceful Shutdown - Exit cleanly
8. ✅ Signal Handling - Listen to the system

Your Go programs are now ready to handle real-world concurrent workloads like a well-run kitchen! 🍳👨‍🍳

Now go build something amazing!