go 52 lines · 8 steps

The fan-in pattern in Go channels

How to merge many channels into one and collect from several sources without leaks or deadlocks.

Explained by highlit
Advanced concurrency channels fan-in select waitgroup
1package pipeline
2 
3import "sync"
4 
5func mergeMetrics(sources ...<-chan Metric) <-chan Metric {
6 out := make(chan Metric)
7 var wg sync.WaitGroup
8 
9 drain := func(src <-chan Metric) {
10 defer wg.Done()
11 for m := range src {
12 out <- m
13 }
14 }
15 
16 wg.Add(len(sources))
17 for _, src := range sources {
18 go drain(src)
19 }
20 
21 go func() {
22 wg.Wait()
23 close(out)
24 }()
25 
26 return out
27}
28 
29func collect(primary, fallback <-chan Metric, done <-chan struct{}) []Metric {
30 var collected []Metric
31 for {
32 select {
33 case m, ok := <-primary:
34 if !ok {
35 primary = nil
36 break
37 }
38 collected = append(collected, m)
39 case m, ok := <-fallback:
40 if !ok {
41 fallback = nil
42 break
43 }
44 collected = append(collected, m)
45 case <-done:
46 return collected
47 }
48 if primary == nil && fallback == nil {
49 return collected
50 }
51 }
52}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A WaitGroup that closes the output channel once all producers finish is the safe way to signal completion in a fan-in.
  2. 2Setting a channel variable to nil in a select disables that case, letting you wind down partial sources cleanly.
  3. 3Combining a done channel with drained-source detection gives you two independent ways to stop collecting.

Related explainers

go 
package feed
 
import (
	"encoding/json"

Streaming a JSON array in Go

streaming json decoder
Intermediate 6 steps
python 
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
 

Parallel file downloads with a thread pool

concurrency thread-pool streaming-io
Intermediate 8 steps
go 
package validate
 
import (
	"fmt"

Struct validation with reflection in Go

reflection struct-tags validation
Intermediate 7 steps
java 
public class EventBus {
 
    private final Map<Class<?>, List<Subscriber>> subscribers = new ConcurrentHashMap<>();
    private final Executor executor;

Building an annotation-driven EventBus in Java

publish-subscribe reflection annotations
Intermediate 7 steps
go 
package middleware
 
import (
	"bytes"

Verifying webhook HMAC signatures in Gin

hmac middleware webhooks
Intermediate 7 steps
php 
final class TokenBucketLimiter
{
    private float $tokens;
    private float $lastRefill;

How a token bucket rate limiter works

rate-limiting token-bucket throttling
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

The fan-in pattern in Go channels — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code