04 - Fan-Out / Fan-In

🎁 You have 100 URLs to fetch. Doing them one at a time takes 100 seconds. Doing them all at once hammers the server. How do you find the sweet spot?

Fan-out means distributing work from one source across multiple concurrent workers. Fan-in means merging their results back into a single stream. How the work reaches the workers — a shared channel, per-worker channels, or a dispatcher — is an implementation detail; the shape is what matters. In this lesson we use the shared-channel approach because it's the simplest. Once you see this shape, you'll recognize it everywhere.

We touched on this in Go Essentials. Now we go deeper — dynamic workers, ordered results, and real-world usage.

Why Fan-Out / Fan-In

A single goroutine processing items one at a time is simple but slow. If each item takes 100ms and you have 100 items, that's 10 seconds. Fan-out to 10 workers and it's 1 second. The work is the same — you're just doing it in parallel.

But now you have 10 goroutines producing results on separate channels. The consumer shouldn't care how many workers there are. Fan-in merges everything back into one channel — the consumer reads from a single stream.

Fan-Out

Multiple goroutines reading from a single channel. The Go runtime handles distribution — whichever goroutine is ready gets the next value.

package main

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

    // Fan-out: 3 workers reading from the same channel
    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                fmt.Printf("worker %d got job %d\n", id, job)
                time.Sleep(100 * time.Millisecond) // simulate processing time
            }
        }(i)
    }

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

    wg.Wait()
}

Work is distributed automatically. When multiple goroutines range over the same channel, each value is received by exactly one goroutine — whichever is ready first grabs it. No duplication, no locking. The Go runtime handles the scheduling: if worker 0 is busy, workers 1 or 2 pick up the next job.

Fan-In

Merge multiple channels into one. The consumer reads from a single channel regardless of how many producers there are.

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for val := range c {
                out <- val
            }
        }(ch)
    }

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

    return out
}

This is a reusable function. Pass any number of channels, get one merged channel back.

The variadic signature (channels ...<-chan int) is convenient when you have a known set of channels: fanIn(ch1, ch2, ch3). When channels come from a dynamic loop, a slice parameter (channels []<-chan int) is more natural. Either works — it's just a calling convention.

Fan-Out / Fan-In Combined

Compute squares of 1 to n using 3 concurrent workers. The generator produces numbers, workers square them in parallel, and fanIn merges the results into a single channel.

package main

func generator(ctx context.Context, n int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= n; i++ {
            select {
            case out <- i:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func worker(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
}

func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

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

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

    return out
}

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

    // Generate numbers 1 to 10
    nums := generator(ctx, 10)

    // Fan-out: spawn 3 workers reading from the same input
    numWorkers := 3
    workers := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = worker(ctx, nums)
    }

    // Fan-in: merge results
    for val := range fanIn(ctx, workers...) {
        fmt.Println(val)
    }
}

Important: all three workers read from the same nums channel. Each value goes to exactly one worker — the channel handles distribution.

We use integer squaring here so the mechanics stay front and center, but mentally swap n * n for "fetch this URL" and it's exactly the 100-URLs problem from the top of the lesson: the generator emits URLs, the workers fetch them in parallel, and fan-in merges the responses into one stream. Lesson 05 builds that concrete URL fetcher as a worker pool.

Ordered Results

Fan-out/fan-in doesn't preserve order. If you need results in the same order as input, wrap values with an index.

type Job struct {
    Index int
    Value int
}

type Result struct {
    Index int
    Value int
}

func orderedWorker(ctx context.Context, in <-chan Job) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for job := range in {
            select {
            case out <- Result{Index: job.Index, Value: job.Value * job.Value}:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

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

    jobs := make(chan Job, 10)
    for i := 0; i < 10; i++ {
        jobs <- Job{Index: i, Value: i + 1}
    }
    close(jobs)

    // Fan-out
    r1 := orderedWorker(ctx, jobs)
    r2 := orderedWorker(ctx, jobs)
    r3 := orderedWorker(ctx, jobs)

    // Fan-in and collect
    merged := fanInResults(ctx, r1, r2, r3)

    // Collect all results, then sort by index
    var all []Result
    for r := range merged {
        all = append(all, r)
    }

    sort.Slice(all, func(i, j int) bool {
        return all[i].Index < all[j].Index
    })

    for _, r := range all {
        fmt.Printf("job %d: %d\n", r.Index, r.Value)
    }
}

The trade-off: you collect all results before sorting, which uses more memory. For streaming ordered results, you'd need a more complex reordering buffer.

Dynamic Worker Count

Sometimes you don't know how many workers you need upfront. Scale based on input size or system resources.

func fanOutDynamic(ctx context.Context, in <-chan int, numWorkers int) <-chan int {
    workers := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = worker(ctx, in)
    }
    return fanIn(ctx, workers...)
}

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

    nums := generator(ctx, 10)

    numWorkers := runtime.NumCPU()
    results := fanOutDynamic(ctx, nums, numWorkers)

    for val := range results {
        fmt.Println(val)
    }
}

runtime.NumCPU() is a common choice for CPU-bound work — squaring numbers, parsing, compression — because more goroutines than cores just adds scheduling overhead without doing more work in parallel.

For I/O-bound work (HTTP calls, database queries) the CPU is mostly idle, waiting on the network. Here the right count isn't tied to cores at all — it's bounded by whatever is downstream:

• The remote service's limits. If an API allows 100 concurrent requests, more than 100 workers just get rejected or throttled.
• Your connection pool. With a DB pool of 20, a 200-worker pool leaves 180 workers blocked waiting for a connection — you've spent memory to create a queue.
• Target latency vs throughput. More workers raise throughput until the downstream saturates, then latency climbs while throughput flatlines.

So "higher for I/O-bound" really means "as high as the slowest downstream resource can absorb, and no higher." Finding and enforcing that ceiling is exactly what the next two lessons cover — rate limiting (06) caps requests over time, and semaphores/bounded concurrency (07) cap how many run at once.

Fan-Out vs Worker Pool

They solve similar problems but differ in structure:

Both exit when the input channel is exhausted in batch scenarios. The lifetime difference only appears when a worker pool is used in a long-running service (e.g., background job queue that never closes) — in that case workers stay alive waiting for the next job.

Failure Mode: Forgetting to Close the Merged Channel

The fanIn function's most important line is easy to leave out. The merged out channel has multiple producers (one goroutine per input channel), so no single producer can close it. A separate goroutine must wg.Wait() for all of them, then close:

func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for val := range c {
                out <- val
            }
        }(ch)
    }

    // ❌ Forget this and out is never closed.
    // go func() {
    //     wg.Wait()
    //     close(out)
    // }()

    return out
}

If you drop that closing goroutine, for val := range fanIn(...) in the consumer never ends — range blocks forever waiting for a value or a close that never comes. That's a deadlock (if it's the only goroutine left, Go panics with all goroutines are asleep) or a goroutine leak (if other work keeps the program alive). The fix is always the same: whoever coordinates the producers closes the channel after wg.Wait().

When NOT to Use Fan-Out / Fan-In

Fan-out/fan-in isn't free. Skip it when:

• The work is trivial. If each item takes microseconds, goroutine creation, channel sends, and merging cost more than just looping. Parallelism only pays off when per-item work clearly exceeds coordination overhead.
• You need strict ordering with low latency. Fan-out scrambles order. You can restore it (index-wrapping above), but collect-then-sort buffers every result before emitting the first — bad for streaming or large result sets.
• A single bottleneck serializes everything anyway. If all workers hit one lock, one connection, or one rate-limited API, they queue behind it and you get the coordination cost with none of the speedup.
• A plain worker pool is enough. If you don't need per-worker output streams or pipeline composition, the shared-channel worker pool in the next lesson is simpler.

Key Takeaways

• Fan-out: multiple goroutines reading from one channel — work distributes automatically
• Fan-in: merge multiple channels into one — consumers see a single stream
• Always use context for cancellation in every goroutine
• Results arrive in arbitrary order — use index wrapping if order matters
• Scale worker count with runtime.NumCPU() for CPU-bound; for I/O-bound, bound it by the slowest downstream resource (rate limit, connection pool) — not by cores
• Fan-out/fan-in is a dataflow shape (route work out, merge results in); a worker pool is one way to implement it — both can exit on input exhaustion in batch mode

🎁 The naive approach — one goroutine per item — works until you have 100,000 items and 100,000 goroutines competing for resources. Worker pools give you the same parallelism with a fixed, controlled number of goroutines.