Fan-Out Fan-In in Go: Parallel Work with Channels
You have a batch of independent items and one slow function to run over each. Fan-out fan-in runs that function on many at once and reassembles the answers. This guide builds it correctly: parallel, ordered, cancellable, and error-carrying. Every example was run on Go 1.24.7. Works with Go 1.21+.
If starting a goroutine or reading a channel is still shaky, the goroutines concurrency guide and Go channels explained cover the mechanics, and the complete Go tutorial covers the rest of the language.
What fan-out and fan-in actually mean
Two halves of one pattern. Fan-out is starting several goroutines that all read from a single input channel, so one queue of work is spread across many workers running in parallel. Fan-in is the reverse: merging several channels into one, so a single collector can read every result from one place without knowing which worker produced it. Put them together and you get the shape most Go concurrency ends up in: split a slice of work across N workers, let them run at once, and gather the results as they finish. The tutorials that rank for this pattern nail the shape and stop there. The shape is the easy part. Order, cancellation, and errors are where real code lives, and that is most of this article.
Fan-out: many workers reading one jobs channel
Start with the payoff. You have 40 frames to transcode, each taking 20 ms, and you want them done in a fraction of 800 ms. Fan out to 8 workers, all ranging over the same jobs channel. Because a range over a channel is safe for concurrent receivers, the runtime hands each queued value to whichever worker is free.
package main
import (
"fmt"
"sync"
"time"
)
// transcode simulates a slow per-item transform: a thumbnail resize, a checksum, an API call.
func transcode(frame int) int {
time.Sleep(20 * time.Millisecond)
return frame * frame
}
func main() {
const frameCount = 40
frames := make([]int, frameCount)
for i := range frames {
frames[i] = i + 1
}
// Sequential baseline.
seqStart := time.Now()
for _, f := range frames {
_ = transcode(f)
}
seq := time.Since(seqStart)
// Fan-out: workers all pull from one jobs channel.
const workers = 8
jobs := make(chan int)
results := make(chan int)
fanStart := time.Now()
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for f := range jobs { // every worker reads from the same channel
results <- transcode(f)
}
}()
}
go func() { // feeder
for _, f := range frames {
jobs <- f
}
close(jobs)
}()
go func() { // closer: close results only after all workers return
wg.Wait()
close(results)
}()
total := 0
for r := range results {
total += r
}
fan := time.Since(fanStart)
fmt.Printf("frames: %d\n", frameCount)
fmt.Printf("checksum: %d\n", total)
fmt.Printf("sequential: %v\n", seq.Round(time.Millisecond))
fmt.Printf("fan-out (8): %v\n", fan.Round(time.Millisecond))
fmt.Printf("speedup: %.1fx\n", float64(seq)/float64(fan))
}Output:
frames: 40
checksum: 22140
sequential: 810ms
fan-out (8): 102ms
speedup: 8.0xA clean 8x from 8 workers, because the work is I/O-like (a sleep): 40 jobs in 5 waves of 8. Three separate goroutines run the show and it matters which does what. The workers send results; the feeder sends jobs and closes jobs when the input is exhausted; a dedicated closer waits on the sync.WaitGroup and closes results only after every worker has returned. That last part is not a style choice. Whoever sends on a channel must never be the one to close it out from under a peer, and the close must happen after the last send. Get that ordering wrong and you get a panic or a hang, both shown at the end.
Fan-in: merging channels with a WaitGroup-closed output
The version above already fans in through a shared results channel, which is the common case. The classic fan-in is more general: you have several separate channels, each produced independently, and you want to read them as one stream. Think of three regional price feeds, each its own channel, that you want to consume without caring which region a price came from. A merge function starts one forwarding goroutine per input and closes the output once all inputs are drained.
package main
import (
"fmt"
"sync"
)
// merge fans in: one goroutine drains each input channel and forwards to a single output.
func merge(inputs ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(len(inputs))
for _, in := range inputs {
go func(c <-chan int) {
defer wg.Done()
for v := range c { // drain this input until it closes
out <- v
}
}(in)
}
go func() {
wg.Wait() // every forwarder has finished
close(out) // only now is it safe to close the merged channel
}()
return out
}
// shardPrices is one producer: it emits its values then closes its own channel.
func shardPrices(values ...int) <-chan int {
out := make(chan int, len(values))
for _, v := range values {
out <- v
}
close(out)
return out
}
func main() {
europe := shardPrices(101, 102, 103)
usEast := shardPrices(201, 202, 203)
asia := shardPrices(301, 302, 303)
total, count := 0, 0
for price := range merge(europe, usEast, asia) {
total += price
count++
}
fmt.Printf("merged %d prices from 3 shards\n", count)
fmt.Printf("sum: %d\n", total)
}Output:
merged 9 prices from 3 shards
sum: 1818The wg.Add(len(inputs)) counts one forwarder per input, and the separate goroutine that calls close(out) after wg.Wait() is the same close-after-Wait discipline as before, just applied to N sources instead of N workers. This merge is the reusable half of fan-in; it appears almost verbatim in the Go pipelines pattern and in Rob Pike’s Go concurrency patterns writeup, where it composes stages of a pipeline.
Fan-out loses order, and how to get it back
Here is the thing every “just merge the channels” tutorial skips: fan-out destroys input order. Once eight workers run in parallel, the one that finishes first sends first, and completion order depends on scheduling and how long each item takes, not on input position. If you need results lined up with inputs, you have to restore order yourself. The fix is cheap: tag each job with its index, and have the collector slot each result back into that position instead of appending.
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
// Tagged carries the item's original position so the collector can restore order.
type Tagged struct {
Index int
Value int
}
func main() {
inputs := []int{10, 20, 30, 40, 50, 60, 70, 80}
jobs := make(chan Tagged)
results := make(chan Tagged)
const workers = 4
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for job := range jobs {
// Uneven work time scrambles the order results come back.
time.Sleep(time.Duration(rand.Intn(15)) * time.Millisecond)
results <- Tagged{Index: job.Index, Value: job.Value * 2}
}
}()
}
go func() {
for i, v := range inputs {
jobs <- Tagged{Index: i, Value: v}
}
close(jobs)
}()
go func() {
wg.Wait()
close(results)
}()
ordered := make([]int, len(inputs))
var arrival []int
for r := range results {
ordered[r.Index] = r.Value // slot each result back into its input position
arrival = append(arrival, r.Value)
}
fmt.Println("arrival order: ", arrival)
fmt.Println("restored order:", ordered)
}Output (arrival order changes every run; restored order never does):
arrival order: [80 20 40 140 100 160 60 120]
restored order: [20 40 60 80 100 120 140 160]Run it twice and the arrival line is different each time, while the restored line is identical. Writing into ordered[r.Index] needs no lock because every worker writes a distinct index, so there is no shared slot. This is the frank answer to “why is my output shuffled”: fan-out is unordered by construction, and a one-field index tag plus a pre-sized slice is how you buy order back when you need it. When you do not need it, skip the tag and save the allocation.
Cancellation: stopping every worker on failure or timeout
A batch that runs to completion is the easy mode. An interactive request is not: if one item fails or the caller’s deadline passes, you want every worker to stop now, not grind through 400 more results nobody will read. That is what context.Context is for. Workers check ctx.Err() before taking new work and select on ctx.Done() inside any blocking send, the feeder stops feeding, and the collector calls cancel() on the first error.
package main
import (
"context"
"errors"
"fmt"
"sync"
"time"
)
func validate(ctx context.Context, id int) (int, error) {
select {
case <-time.After(30 * time.Millisecond): // simulate the work
if id == 5 {
return 0, fmt.Errorf("item %d: validation failed", id)
}
return id * id, nil
case <-ctx.Done(): // cancelled mid-flight: abandon this call
return 0, ctx.Err()
}
}
type Result struct {
Value int
Err error
}
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
jobs := make(chan int)
results := make(chan Result)
const workers = 4
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for id := range jobs {
if ctx.Err() != nil { // stop taking new work once cancelled
return
}
value, err := validate(ctx, id)
select {
case results <- Result{Value: value, Err: err}:
case <-ctx.Done():
return
}
}
}()
}
go func() { // feeder stops feeding the moment anyone cancels
defer close(jobs)
for id := 1; id <= 500; id++ {
select {
case jobs <- id:
case <-ctx.Done():
return
}
}
}()
go func() {
wg.Wait()
close(results)
}()
start := time.Now()
var firstErr error
succeeded := 0
for r := range results {
if r.Err != nil {
if firstErr == nil {
firstErr = r.Err
cancel() // one failure stops the whole fan-out
}
continue
}
succeeded++
}
fmt.Printf("succeeded before stop: %d (varies per run)\n", succeeded)
fmt.Printf("stopped after: %v\n", time.Since(start).Round(10*time.Millisecond))
fmt.Println("first error: ", firstErr)
fmt.Println("cancelled: ", errors.Is(ctx.Err(), context.Canceled))
}Output (the success count varies; the stop time and error do not):
succeeded before stop: 4 (varies per run)
stopped after: 60ms
first error: item 5: validation failed
cancelled: trueFive hundred items through four workers at 30 ms each would take almost four seconds. It stopped in 60 ms because item 5 failed in the second wave, cancel() fired, and every worker saw ctx.Err() != nil and returned instead of pulling more work. Two escape hatches make this reliable: the ctx.Done() case inside validate aborts a call already in flight, and the ctx.Err() check at the top of the loop stops a worker from taking new work. The select on the results send matters too. Without it, a worker could block forever trying to send into results after the collector has already stopped reading, which is a classic leak. Swap WithCancel for WithTimeout and the same machinery gives you a deadline for free. defer cancel() is not optional either: a context you never cancel leaks its internal goroutine.
Per-item errors: results that carry a value or an error
Cancellation stops on the first failure, which is right for a request. A batch job usually wants the opposite: process everything, then report which items failed. The clean way is to make the result a struct that carries either a value or an error, exactly the shape a function’s (T, error) return has. Workers never log or panic on their own; they package the outcome and send it, and the collector decides.
package main
import (
"errors"
"fmt"
"sort"
"sync"
"time"
)
// Result carries either the parsed value or the error for one record. Never both matter at once.
type Result struct {
RecordID int
Amount int
Err error
}
func parseRecord(id int) (int, error) {
time.Sleep(10 * time.Millisecond)
if id%9 == 0 { // pretend every 9th record is malformed
return 0, fmt.Errorf("record %d: malformed amount", id)
}
return id * 100, nil
}
func main() {
const recordCount = 30
jobs := make(chan int)
results := make(chan Result)
const workers = 6
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for id := range jobs {
amount, err := parseRecord(id)
results <- Result{RecordID: id, Amount: amount, Err: err}
}
}()
}
go func() {
for id := 1; id <= recordCount; id++ {
jobs <- id
}
close(jobs)
}()
go func() {
wg.Wait()
close(results)
}()
var failures []error
sum, ok := 0, 0
for r := range results {
if r.Err != nil {
failures = append(failures, r.Err)
continue // keep going: one bad record does not sink the batch
}
sum += r.Amount
ok++
}
sort.Slice(failures, func(i, j int) bool { return failures[i].Error() < failures[j].Error() })
fmt.Printf("records: %d\n", recordCount)
fmt.Printf("succeeded: %d (sum %d)\n", ok, sum)
fmt.Printf("failed: %d\n", len(failures))
for _, err := range failures {
fmt.Println(" -", err)
}
fmt.Println("joined into one error:", errors.Join(failures...) != nil)
}Output (failures are string-sorted here, so 18 and 27 precede 9):
records: 30
succeeded: 27 (sum 41100)
failed: 3
- record 18: malformed amount
- record 27: malformed amount
- record 9: malformed amount
joined into one error: trueThe collector keeps every failure instead of stopping at the first, then errors.Join (Go 1.20+) bundles them into a single error you can return upward. If you need to inspect specific failures with errors.Is or errors.As, wrap them with %w, covered in Go error handling. The one rule that keeps this from becoming a debugging nightmare: a worker must never silently drop an error. Every job produces exactly one Result, error or not, so the counts always add up.
Bounding the fan-out width
Notice that every example above already bounds concurrency, because “N workers reading one channel” caps in-flight work at N by construction. That is worth stating plainly, because the tempting mistake is to fan out with a goroutine per item:
for _, item := range items {
go process(item) // unbounded fan-out: 50,000 items means 50,000 goroutines
}With 50,000 items that is 50,000 concurrent calls hitting the same database or downstream API at once, which exhausts connection pools and trips rate limiters. The fixed-worker version has no such failure mode: the jobs channel provides backpressure, and feeding it blocks once all workers are busy. So the worker count is your concurrency limit, and choosing it is the same decision covered in depth in the worker pool guide: use runtime.GOMAXPROCS(0) for CPU-bound work, and a larger number tuned under load for I/O-bound work.
Which raises the frank question of how fan-out fan-in differs from a worker pool, because they overlap heavily. A worker pool is a long-lived set of workers you feed continuously, often for the life of a service. Fan-out fan-in is usually a one-shot: take this batch, spread it, collect it, done. Mechanically they share the same core (N workers, a jobs channel, a results channel, close-after-Wait), and you will see the terms used interchangeably. The useful distinction is lifetime and intent: reach for the pool vocabulary when workers persist and pull a stream, and for fan-out fan-in when you are parallelizing one bounded batch and reassembling its output. Both, and the other shapes like pipelines, are surveyed in Go concurrency patterns.
A real parallel transform: bounded, ordered, cancellable
Here is everything in one function you could ship: fingerprint a batch of documents with an expensive iterated hash, fan out to one worker per core, restore input order in the output, and respect a timeout. It times itself against the sequential version so the payoff is concrete, and the hashes are deterministic so you can verify the ordering held.
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"runtime"
"sync"
"time"
)
// Document is one unit of work: bytes to fingerprint, plus its position in the batch.
type Document struct {
Index int
Body []byte
}
// Fingerprint is the tagged result: the hash for a document, or an error, at its index.
type Fingerprint struct {
Index int
Hash string
Err error
}
// stretch is deliberate CPU work: iterated SHA-256, like a slow key-derivation step.
func stretch(ctx context.Context, body []byte) (string, error) {
sum := sha256.Sum256(body)
for i := 0; i < 150_000; i++ {
if i%20_000 == 0 && ctx.Err() != nil { // check cancellation periodically
return "", ctx.Err()
}
sum = sha256.Sum256(sum[:])
}
return hex.EncodeToString(sum[:8]), nil
}
// FingerprintBatch fans out over workerCount workers and fans results back in, ordered.
func FingerprintBatch(ctx context.Context, docs []Document, workerCount int) ([]string, error) {
jobs := make(chan Document)
results := make(chan Fingerprint)
go func() { // feeder
defer close(jobs)
for _, doc := range docs {
select {
case jobs <- doc:
case <-ctx.Done():
return
}
}
}()
var wg sync.WaitGroup
for w := 0; w < workerCount; w++ { // bounded fan-out: exactly workerCount workers
wg.Add(1)
go func() {
defer wg.Done()
for doc := range jobs {
hash, err := stretch(ctx, doc.Body)
select {
case results <- Fingerprint{Index: doc.Index, Hash: hash, Err: err}:
case <-ctx.Done():
return
}
}
}()
}
go func() { // closer
wg.Wait()
close(results)
}()
hashes := make([]string, len(docs))
var firstErr error
for r := range results { // fan-in, restoring order by index
if r.Err != nil && firstErr == nil {
firstErr = r.Err
}
hashes[r.Index] = r.Hash
}
if firstErr != nil {
return nil, firstErr
}
return hashes, nil
}
func main() {
docs := make([]Document, 24)
for i := range docs {
docs[i] = Document{Index: i, Body: []byte(fmt.Sprintf("document-payload-%d", i))}
}
// Sequential baseline.
seqStart := time.Now()
for _, doc := range docs {
_, _ = stretch(context.Background(), doc.Body)
}
seq := time.Since(seqStart)
workers := runtime.GOMAXPROCS(0) // CPU-bound: match the core count
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
fanStart := time.Now()
hashes, err := FingerprintBatch(ctx, docs, workers)
fan := time.Since(fanStart)
if err != nil {
fmt.Println("batch failed:", err)
return
}
fmt.Printf("documents: %d\n", len(docs))
fmt.Printf("GOMAXPROCS: %d\n", workers)
fmt.Printf("sequential: %v\n", seq.Round(time.Millisecond))
fmt.Printf("fan-out: %v\n", fan.Round(time.Millisecond))
fmt.Printf("speedup: %.1fx\n", float64(seq)/float64(fan))
fmt.Printf("first 3 hashes (in input order): %s %s %s\n", hashes[0], hashes[1], hashes[2])
fmt.Printf("last hash: %s\n", hashes[len(hashes)-1])
}Output (on a 2-core machine; hashes are identical every run):
documents: 24
GOMAXPROCS: 2
sequential: 343ms
fan-out: 173ms
speedup: 2.0xA clean 2.0x on two cores, which is the ceiling for CPU-bound work: unlike the I/O example that hit 8x, hashing keeps a core busy the whole time, so more workers than cores buys nothing. The speedup is real but capped by hardware, and that is the frank expectation for compute-heavy fan-out. The hashes print in input order despite four goroutines finishing in whatever order, because each writes hashes[r.Index]. Swap stretch for a real transform, pass the request’s context in, and FingerprintBatch drops into a service unchanged: the caller owns the timeout, order is preserved, and nothing leaks because every goroutine’s exit is guaranteed by a closed channel or a done context.
Common fan-out fan-in mistakes
Never closing the merged channel. Forget the wg.Wait() plus close(out) goroutine in merge and the collector’s range blocks forever. In a short program the runtime catches it:
fatal error: all goroutines are asleep - deadlock!In a long-running server there is no deadlock message: the collector goroutine just leaks, parked on a receive that will never complete, and it stacks up one per request until you run out of memory. This is one of the most common goroutine leaks in Go. The merged channel must always be closed after its last sender finishes.
Closing the shared channel from a worker. If each worker closes results after its send, the second worker to finish sends on an already-closed channel:
panic: send on closed channelOne goroutine owns the close, and it runs only after wg.Wait(). No worker ever closes a channel it sends on.
Losing errors. A results chan string has nowhere to put a failure, so workers end up logging and moving on, and the caller thinks the batch succeeded. Make the result a struct carrying a value or an error, and emit exactly one per job.
Deadlock from the wrong close order. Send every job before receiving any result over unbuffered channels, and the workers block sending results while the feeder blocks sending jobs. Feed from one goroutine and collect in another so the two never wait on each other.
Assuming ordered output. Fan-out is unordered, full stop. If your assertions expect input order, tag results with an index and reassemble into a pre-sized slice, as the ordering and capstone examples do.
What next
You can now fan out a batch across workers, merge the results, restore order when it matters, cancel every worker on failure or timeout, and carry per-item errors without dropping any. Go deeper on the pieces:
- Go concurrency patterns puts fan-out fan-in next to pipelines, generators, and bounded parallelism so you know which to reach for.
- The Go pipeline pattern chains fan-out fan-in stages into a stream, reusing the exact
mergefrom this article. - The worker pool guide covers sizing the worker count with measured CPU-bound and I/O-bound curves.
- The context package and select statement are the cancellation machinery every correct fan-out depends on.
Run the ordering example a few times and watch the arrival line change while the restored line holds. That difference is the whole reason this pattern needs more than a merge function.