Go Concurrency Patterns — Pipeline, Fan-Out/In, Generator, and Error Handling

In this tutorial, you will learn about Go Concurrency Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.

Go concurrency patterns include pipelines for streaming data, fan-out/fan-in for parallelism, generators for lazy sequences, and errgroups for error handling.

What You'll Learn

• Pipeline pattern for data streaming
• Fan-out, fan-in for parallel stages
• Generator pattern for lazy sequences
• errgroup for error propagation

Why It Matters

Concurrency patterns solve real problems. Docker uses pipeline for image building. Kubernetes uses fan-out for parallel reconciliation. DodaZIP uses pipeline for streaming file processing.

Real-World Use

Data processing pipelines, parallel image processing, streaming file transformation, concurrent API aggregation.

flowchart LR
    A["Concurrency Patterns"] --> B["Pipeline"]
    A --> C["Fan-Out/Fan-In"]
    A --> D["Generator"]
    A --> E["ErrGroup"]
    A:::current --> B
    style A fill:#2563eb,stroke:#2563eb,color:#fff
    style B fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style C fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style D fill:#dbeafe,stroke:#2563eb,color:#1e40af
    style E fill:#f1f5f9,stroke:#94a3b8,color:#64748b

Pipeline Pattern

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

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

func main() {
    for n := range sq(sq(gen(1, 2, 3, 4))) {
        fmt.Println(n)
    }
}

Fan-Out, Fan-In

func fanOut(in <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan int)
        channels[i] = ch
        go func(out chan int) {
            for n := range in {
                out <- n * n
            }
            close(out)
        }(ch)
    }
    return channels
}

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

func main() {
    in := gen(1, 2, 3, 4, 5)
    channels := fanOut(in, 3)
    merged := fanIn(channels...)
    for n := range merged {
        fmt.Println("Result:", n)
    }
}

Generator Pattern

func fibonacci() <-chan int {
    ch := make(chan int)
    go func() {
        a, b := 0, 1
        for i := 0; i < 10; i++ {
            ch <- a
            a, b = b, a+b
        }
        close(ch)
    }()
    return ch
}

func main() {
    for n := range fibonacci() {
        fmt.Println(n)
    }
}

ErrGroup Pattern

func main() {
    g, ctx := errgroup.WithContext(context.Background())

    urls := []string{
        "https://example.com",
        "https://golang.org",
        "https://invalid-url",
    }

    results := make([]string, len(urls))
    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return fmt.Errorf("fetching %s: %w", url, err)
            }
            defer resp.Body.Close()
            results[i] = resp.Status
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Println("Error:", err)
        return
    }

    for i, url := range urls {
        fmt.Printf("%s: %s\n", url, results[i])
    }
}

Tee Pattern

func tee(in <-chan int) (_, _ <-chan int) {
    out1 := make(chan int)
    out2 := make(chan int)
    go func() {
        defer close(out1)
        defer close(out2)
        for n := range in {
            out1 <- n
            out2 <- n
        }
    }()
    return out1, out2
}

func main() {
    in := gen(1, 2, 3)
    out1, out2 := tee(in)
    go func() {
        for n := range out1 { fmt.Println("out1:", n) }
    }()
    for n := range out2 { fmt.Println("out2:", n) }
}

Common Mistakes

1. Goroutine Leaks

ch := make(chan int)
go func() { ch <- 42 }()  // Leaks if nobody reads

2. Pipeline Blocking

All stages must run concurrently. If one stage blocks, the whole pipeline stalls.

3. Not Handling Done Signals

Pipelines should accept context for cancellation. Otherwise goroutines leak on shutdown.

4. Deadlock in Fan-In

Closing the output channel must happen after all input channels are drained.

5. Race in Shared Results

Writing to a shared slice from multiple goroutines needs synchronization.

Practice Questions

1. What is the pipeline pattern? A series of stages connected by channels. Each stage is a goroutine that processes data from input channel and sends to output.

2. How does fan-out differ from worker pool? Fan-out distributes to fixed channels. Worker pool has workers competing for jobs from a single channel.

3. What does errgroup provide? Sync.WaitGroup with error propagation. If one goroutine errors, context is cancelled and other goroutines see cancellation.

4. What is the generator pattern? A function that returns a channel and produces values lazily in a goroutine. Used for infinite sequences.

Challenge: Implement a pipeline for processing log lines: read, parse, filter errors, write.

func readLines(filename string) <-chan string { /* read file, send lines */ }
func parseLine(in <-chan string) <-chan LogEntry { /* parse each line */ }
func filterErrors(in <-chan LogEntry) <-chan LogEntry {
    out := make(chan LogEntry)
    go func() {
        for entry := range in {
            if entry.Level == "ERROR" { out <- entry }
        }
        close(out)
    }()
    return out
}
func writeEntries(in <-chan LogEntry) { /* write to output */ }

FAQ

{{< faq question="When should I use pipeline over slice processing?" >}} Pipelines for streaming data (files, network) or when each stage is expensive. Slice processing for in-memory batch operations. {{< /faq >}}

{{< faq question="How do I limit goroutines in a pipeline?" >}} Use a Semaphore channel: sem := make(chan struct{}, 10). Acquire before launch, release after completion. {{< /faq >}}

{{< faq question="Can pipelines handle backpressure?" >}} Yes. If a stage processes slower, channels fill up and senders block. This naturally propagates backpressure upstream. {{< /faq >}}

{{< faq question="What is the difference between errgroup and WaitGroup?" >}} errgroup wraps WaitGroup with error collection and context cancellation. First error cancels all goroutines. {{< /faq >}}

{{< faq question="How do I test concurrent patterns?" >}} Use buffered channels for deterministic tests. Test each stage in isolation. Use race detector (-race) during tests. {{< /faq >}}

Try It Yourself

package main

import "fmt"

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

func main() {
    for n := range gen(1, 2, 3) {
        fmt.Println(n)
    }
}

Expected output:

1
2
3

What's Next

Now that you understand concurrency patterns, explore file I/O and JSON handling in Go.