Go Concurrency: Fan-Out, Pipeline, and Worker Pool

Go’s concurrency model built on goroutines and channels is one of its strongest features, but raw goroutines and channels are just building blocks. Real-world systems need structured patterns that handle backpressure, cancellation, and error propagation. This guide covers three essential patterns that every Go developer should know.

The Pipeline Pattern

A pipeline is a series of stages connected by channels, where each stage is a group of goroutines running the same function. Each stage takes values in from upstream via inbound channels, performs some function on that data, and sends values downstream via outbound channels.

package main

import (
	"context"
	"fmt"
	"math"
)

// generator produces integers from a list and sends them on a channel.
func generator(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// square reads integers from a channel, squares them, and sends them out.
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 out <- n * n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// isPrime checks each number and passes through only primes.
func filterPrimes(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			if isPrime(n) {
				select {
				case out <- n:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out
}

func isPrime(n int) bool {
	if n < 2 {
		return false
	}
	for i := 2; i <= int(math.Sqrt(float64(n))); i++ {
		if n%i == 0 {
			return false
		}
	}
	return true
}

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

	// Build the pipeline: generate -> square -> filter primes
	nums := generator(ctx, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13)
	squared := square(ctx, nums)
	primes := filterPrimes(ctx, squared)

	// Consume the output
	for p := range primes {
		fmt.Println(p)
	}
}

Every stage respects the context for cancellation. When the context is cancelled, goroutines stop sending and exit cleanly. This is critical for avoiding goroutine leaks in production systems.

Fan-Out and Fan-In

Fan-out means starting multiple goroutines to read from the same channel. Fan-in means multiplexing multiple channels onto a single channel. This combination is powerful when you have a CPU-bound or I/O-bound stage that benefits from parallelism.

package main

import (
	"context"
	"crypto/sha256"
	"fmt"
	"sync"
)

// hashJob represents a unit of work.
type hashJob struct {
	input  string
	result string
}

// producer sends strings to be hashed.
func producer(ctx context.Context, items []string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for _, item := range items {
			select {
			case out <- item:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// hasher reads strings and computes their SHA-256 hash.
func hasher(ctx context.Context, in <-chan string) <-chan hashJob {
	out := make(chan hashJob)
	go func() {
		defer close(out)
		for s := range in {
			hash := sha256.Sum256([]byte(s))
			result := hashJob{
				input:  s,
				result: fmt.Sprintf("%x", hash),
			}
			select {
			case out <- result:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// fanIn merges multiple hashJob channels into one.
func fanIn(ctx context.Context, channels ...<-chan hashJob) <-chan hashJob {
	out := make(chan hashJob)
	var wg sync.WaitGroup

	for _, ch := range channels {
		wg.Add(1)
		go func(c <-chan hashJob) {
			defer wg.Done()
			for job := range c {
				select {
				case out <- job:
				case <-ctx.Done():
					return
				}
			}
		}(ch)
	}

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

	return out
}

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

	items := []string{
		"hello", "world", "go", "concurrency",
		"fan-out", "fan-in", "pipeline", "pattern",
	}

	// Single producer
	source := producer(ctx, items)

	// Fan-out: start 3 hashers reading from the same channel
	const numWorkers = 3
	hashChannels := make([]<-chan hashJob, numWorkers)
	for i := 0; i < numWorkers; i++ {
		hashChannels[i] = hasher(ctx, source)
	}

	// Fan-in: merge results
	results := fanIn(ctx, hashChannels...)

	// Consume results
	for job := range results {
		fmt.Printf("%s -> %s\n", job.input, job.result[:16])
	}
}

The key insight here is that multiple goroutines can safely read from the same channel. Go guarantees that each value sent on a channel is received by exactly one goroutine. This makes fan-out natural and safe.

The Worker Pool Pattern

A worker pool limits concurrency to a fixed number of goroutines. This is essential when you need to control resource usage, such as database connections, HTTP clients, or CPU-intensive operations.

package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// Task represents a unit of work.
type Task struct {
	ID      int
	Payload string
}

// Result represents the outcome of processing a task.
type Result struct {
	TaskID   int
	Output   string
	Duration time.Duration
	Worker   int
}

// WorkerPool manages a fixed number of worker goroutines.
type WorkerPool struct {
	numWorkers int
	tasks      chan Task
	results    chan Result
	wg         sync.WaitGroup
}

// NewWorkerPool creates a new pool with the given size and buffer.
func NewWorkerPool(numWorkers, bufferSize int) *WorkerPool {
	return &WorkerPool{
		numWorkers: numWorkers,
		tasks:      make(chan Task, bufferSize),
		results:    make(chan Result, bufferSize),
	}
}

// Start launches the worker goroutines.
func (wp *WorkerPool) Start(ctx context.Context) {
	for i := 0; i < wp.numWorkers; i++ {
		wp.wg.Add(1)
		go wp.worker(ctx, i)
	}

	// Close results channel when all workers are done
	go func() {
		wp.wg.Wait()
		close(wp.results)
	}()
}

// worker processes tasks from the tasks channel.
func (wp *WorkerPool) worker(ctx context.Context, id int) {
	defer wp.wg.Done()
	for {
		select {
		case task, ok := <-wp.tasks:
			if !ok {
				return
			}
			start := time.Now()

			// Simulate processing
			processingTime := time.Duration(rand.Intn(100)) * time.Millisecond
			time.Sleep(processingTime)

			result := Result{
				TaskID:   task.ID,
				Output:   fmt.Sprintf("processed: %s", task.Payload),
				Duration: time.Since(start),
				Worker:   id,
			}

			select {
			case wp.results <- result:
			case <-ctx.Done():
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

// Submit adds a task to the pool.
func (wp *WorkerPool) Submit(task Task) {
	wp.tasks <- task
}

// Results returns the results channel for reading.
func (wp *WorkerPool) Results() <-chan Result {
	return wp.results
}

// Close signals that no more tasks will be submitted.
func (wp *WorkerPool) Close() {
	close(wp.tasks)
}

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

	pool := NewWorkerPool(4, 10)
	pool.Start(ctx)

	// Submit tasks
	go func() {
		for i := 0; i < 20; i++ {
			pool.Submit(Task{
				ID:      i,
				Payload: fmt.Sprintf("task-%d", i),
			})
		}
		pool.Close()
	}()

	// Collect results
	for result := range pool.Results() {
		fmt.Printf("Task %d completed by worker %d in %v: %s\n",
			result.TaskID, result.Worker, result.Duration, result.Output)
	}
}

This worker pool implementation provides backpressure through the buffered channels. When the buffer is full, Submit blocks until a worker picks up a task. This prevents unbounded memory growth.

Combining Patterns: A Processing Pipeline

In real applications, you often combine these patterns. Here is an example that reads URLs, fetches them concurrently with a worker pool, and then processes the results through a pipeline.

package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"
)

type FetchResult struct {
	URL        string
	StatusCode int
	BodySize   int
	Error      error
	Duration   time.Duration
}

// urlGenerator sends URLs down the pipeline.
func urlGenerator(ctx context.Context, urls []string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for _, u := range urls {
			select {
			case out <- u:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// fetchWorkers fans out HTTP fetches across n workers.
func fetchWorkers(ctx context.Context, n int, urls <-chan string) <-chan FetchResult {
	out := make(chan FetchResult)
	var wg sync.WaitGroup

	client := &http.Client{Timeout: 10 * time.Second}

	for i := 0; i < n; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for url := range urls {
				start := time.Now()
				result := FetchResult{URL: url}

				req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
				if err != nil {
					result.Error = err
					result.Duration = time.Since(start)
					out <- result
					continue
				}

				resp, err := client.Do(req)
				if err != nil {
					result.Error = err
					result.Duration = time.Since(start)
					out <- result
					continue
				}

				body, _ := io.ReadAll(resp.Body)
				resp.Body.Close()

				result.StatusCode = resp.StatusCode
				result.BodySize = len(body)
				result.Duration = time.Since(start)

				select {
				case out <- result:
				case <-ctx.Done():
					return
				}
			}
		}()
	}

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

	return out
}

// filterSuccessful only passes through results with 200 status.
func filterSuccessful(ctx context.Context, in <-chan FetchResult) <-chan FetchResult {
	out := make(chan FetchResult)
	go func() {
		defer close(out)
		for r := range in {
			if r.Error == nil && r.StatusCode == 200 {
				select {
				case out <- r:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out
}

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

	urls := []string{
		"https://go.dev",
		"https://pkg.go.dev",
		"https://go.dev/blog",
		"https://go.dev/doc",
		"https://go.dev/play",
	}

	// Pipeline: generate -> fetch (fan-out) -> filter
	source := urlGenerator(ctx, urls)
	fetched := fetchWorkers(ctx, 3, source)
	successful := filterSuccessful(ctx, fetched)

	for result := range successful {
		fmt.Printf("OK: %s (%d bytes, %v)\n",
			result.URL, result.BodySize, result.Duration)
	}
}

Error Handling in Concurrent Pipelines

One challenge with pipelines is propagating errors. The errgroup package from golang.org/x/sync simplifies this significantly.

package main

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

func processItem(ctx context.Context, item int) (int, error) {
	if item == 5 {
		return 0, fmt.Errorf("item %d caused an error", item)
	}
	return item * 2, nil
}

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

	items := make(chan int, 10)
	results := make(chan int, 10)

	// Producer
	g.Go(func() error {
		defer close(items)
		for i := 0; i < 10; i++ {
			select {
			case items <- i:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		return nil
	})

	// Workers with bounded concurrency
	const numWorkers = 3
	for i := 0; i < numWorkers; i++ {
		g.Go(func() error {
			for item := range items {
				result, err := processItem(ctx, item)
				if err != nil {
					return err // cancels context for all goroutines
				}
				select {
				case results <- result:
				case <-ctx.Done():
					return ctx.Err()
				}
			}
			return nil
		})
	}

	// Close results when workers are done
	go func() {
		g.Wait()
		close(results)
	}()

	// Consume
	for r := range results {
		fmt.Println(r)
	}

	if err := g.Wait(); err != nil {
		fmt.Printf("Pipeline error: %v\n", err)
	}
}

When any goroutine in the group returns an error, the context is cancelled, which signals all other goroutines to stop. This gives you clean, coordinated shutdown across the entire pipeline.

Guidelines for Choosing a Pattern

Use a pipeline when your processing has clear sequential stages. Data flows through transformations one after another. Each stage can run concurrently with others because they communicate only through channels.

Use fan-out/fan-in when one stage of your pipeline is a bottleneck. If one stage takes significantly longer than others, running multiple instances of that stage in parallel can improve throughput dramatically.

Use a worker pool when you need to limit concurrency. This is common when interacting with external services that have rate limits, when you want to cap memory usage, or when you need to control the number of open file handles or database connections.

Combine patterns when your system has multiple concerns. A pipeline with a fan-out stage that uses a bounded worker pool gives you both structured data flow and controlled concurrency.

Wrapping Up

These three concurrency patterns form the backbone of most concurrent Go programs. Pipelines give you composable stages. Fan-out/fan-in gives you parallelism. Worker pools give you bounded concurrency. Together with context for cancellation and errgroup for error handling, you have everything you need to build robust concurrent systems in Go. Start simple, measure your bottlenecks, and apply these patterns where they provide real benefit.