Merge Several Channels Into One Channel in Go Safely

Fan-in is the quiet half of fan-out. A pipeline that spreads work over several workers ends up with several result channels, and the consumer wants one stream: for result := range merged. The function that makes that stream is short, and each line of it exists to prevent a specific bug: closing a channel twice, closing it while a sender is still sending, or leaving goroutines blocked after the consumer walks away.

This article writes the generic merge, adds cancellation, shows a version for a number of channels that is not known until run time, and tests the properties that matter. It is a focused look at the fan-in step of fan-out and fan-in, uses channels and close rules, and belongs to the concurrency patterns cluster. Everything ran on Go 1.25.6.

The basic merge

One goroutine per input copies values into the output. The output must be closed exactly once, after the last forwarder finishes, so a WaitGroup counts the forwarders and one extra goroutine closes the channel when the count reaches zero:

package main

import (
	"context"
	"sync"
)

// Merge forwards every value from every input to the returned channel. The result is closed
// after all inputs have closed, or after ctx is cancelled and the forwarders have exited.
func Merge[T any](ctx context.Context, inputs ...<-chan T) <-chan T {
	out := make(chan T)
	var wg sync.WaitGroup
	for _, in := range inputs {
		wg.Go(func() {
			for v := range in {
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		})
	}
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

Four details deserve a sentence each. The output is created by Merge and returned receive-only, so the caller cannot close it; the closing goroutine is the only closer, which is the rule from closing channels. The closing goroutine waits on the WaitGroup rather than being one of the forwarders, because no single forwarder knows whether the others have finished. wg.Go (Go 1.25) replaces the Add and Done pair, which removes the classic bug of calling Add inside the goroutine. And every send selects on ctx.Done(): without it, a consumer that stops reading would leave each forwarder blocked on out <- v forever.

The receive side, for v := range in, does not select on the context, which is fine as long as the inputs close or send eventually. If an input could block indefinitely without closing, the forwarder would never notice cancellation while it waits on that input; a producer that you do not control calls for the receive to select on ctx.Done() as well.

Seeing it work

Three producers send different numbers of values at different speeds. The merged stream arrives interleaved, in no fixed order, so the program sorts before printing to keep the output stable:

package main

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"time"
)

func produce(name string, n int, every time.Duration) <-chan string {
	ch := make(chan string)
	go func() {
		defer close(ch)
		for i := 1; i <= n; i++ {
			time.Sleep(every)
			ch <- fmt.Sprintf("%s-%d", name, i)
		}
	}()
	return ch
}

func main() {
	ctx := context.Background()
	merged := Merge(ctx,
		produce("fast", 4, 5*time.Millisecond),
		produce("medium", 3, 9*time.Millisecond),
		produce("slow", 2, 14*time.Millisecond),
	)

	var all []string
	for v := range merged { // ends when all three inputs have closed
		all = append(all, v)
	}
	slices.Sort(all)
	fmt.Println(len(all), "values:", all)

	// Abandon a merge early: cancel the context and the forwarders exit.
	before := runtime.NumGoroutine()
	cctx, cancel := context.WithCancel(ctx)
	m := Merge(cctx,
		produce("a", 100, time.Millisecond),
		produce("b", 100, time.Millisecond),
	)
	<-m // read one value and walk away
	cancel()
	time.Sleep(50 * time.Millisecond)
	fmt.Println("goroutines left after cancel:", runtime.NumGoroutine()-before)
}

9 values: [fast-1 fast-2 fast-3 fast-4 medium-1 medium-2 medium-3 slow-1 slow-2]
goroutines left after cancel: 2

Nine values arrived from three sources and the loop ended on its own, because the output closed after the last input did. The second half leaves a merge with two producers that would run for another 100 ms each. After cancel() the forwarders return; the producers themselves, written without a context for brevity, are still sending into channels nobody reads and remain blocked, which the output shows as goroutines left. That is the other half of the same lesson: cancellation has to reach the producers too, and a producer that takes the context and selects on it in its own send does (generators). Merge can release its own goroutines, but it cannot unblock goroutines it did not start.

What order do you get

Within one input, values keep their order: a forwarder reads them in sequence and sends them in sequence. Across inputs, there is no defined order. Which forwarder's send is received next depends on scheduling, and when two senders are ready the runtime picks one at random (select fairness). If the consumer needs a global order, merging channels is the wrong tool, and the options are a sorted merge (as in a k-way merge of sorted streams, which needs one value from each input in hand), or tagging values with a sequence number or timestamp and sorting after collection.

A related question is fairness: if one input is always ready and another produces rarely, does the busy one crowd out the quiet one? Not in this design. Each input has its own forwarder, and competition happens only at the unbuffered output, where senders queue in order of arrival. A busy producer cannot starve a quiet one by more than the time it takes to hand over a value.

When the number of inputs is dynamic

Merge takes its inputs as a variadic list, which is fine when you have a slice of channels. It starts one goroutine per channel, which is cheap for tens and wasteful for tens of thousands. When the set of inputs is very large or changes while running, reflect.Select lets a single goroutine wait on a slice of channels:

package main

import (
	"fmt"
	"reflect"
	"slices"
)

// MergeReflect merges any number of channels with one goroutine and no forwarders.
func MergeReflect[T any](inputs []<-chan T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		cases := make([]reflect.SelectCase, len(inputs))
		for i, ch := range inputs {
			cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
		}
		for len(cases) > 0 {
			i, v, ok := reflect.Select(cases)
			if !ok {
				cases = slices.Delete(cases, i, i+1) // this input closed
				continue
			}
			out <- v.Interface().(T)
		}
	}()
	return out
}

func source(start, n int) <-chan int {
	ch := make(chan int)
	go func() {
		defer close(ch)
		for i := range n {
			ch <- start + i
		}
	}()
	return ch
}

func main() {
	inputs := []<-chan int{source(0, 3), source(100, 3), source(200, 3)}
	var got []int
	for v := range MergeReflect(inputs) {
		got = append(got, v)
	}
	slices.Sort(got)
	fmt.Println(got)
}

[0 1 2 100 101 102 200 201 202]

When an input closes, reflect.Select returns its index with ok false, and the code removes that case; when none are left the output closes. A single goroutine handles any number of inputs, and there is no WaitGroup. The costs are real: reflect.Select is much slower than a native select (it allocates and takes every channel lock on each call), and the v.Interface().(T) conversion trades the compiler's type checking for a run-time assertion. I use the goroutine-per-input version unless I have measured a problem with it, which for fewer than a few thousand inputs I never have. The dynamic version is for a set that is large or grows while running, and it needs more work to add inputs after starting.

Tests worth writing

Three properties cover the bugs. Every value from every input arrives exactly once, whatever the order. The output closes after all inputs close and not before. And cancelling releases the forwarders. The first two follow from collecting into a sorted slice and comparing, and from a for range loop that terminates; the third from comparing runtime.NumGoroutine before and after, with a short settle loop, or from a leak checker (goleak). Run under -race and repeat with -count=20, because a fan-in bug depends on how the goroutines interleave and a single run proves little. The same runs also catch a close that races with a send, the characteristic fan-in bug.

If the merge is the end of a pipeline, pipelines shows the stages before it. The Go blog's pipelines article has the original merge function this one descends from, and the reflect.Select documentation specifies the dynamic form.