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 {
out := make(chan Metric)
var wg sync.WaitGroup
8 
drain := func(src <-chan Metric) {
defer wg.Done()
for m := range src {
	out <- m
}
}
15 
wg.Add(len(sources))
for _, src := range sources {
go drain(src)
}
20 
go func() {
wg.Wait()
close(out)
}()
25 
return out
27}
28 
29func collect(primary, fallback <-chan Metric, done <-chan struct{}) []Metric {
var collected []Metric
for {
select {
case m, ok := <-primary:
	if !ok {
		primary = nil
		break
	}
	collected = append(collected, m)
case m, ok := <-fallback:
	if !ok {
		fallback = nil
		break
	}
	collected = append(collected, m)
case <-done:
	return collected
}
if primary == nil && fallback == nil {
	return collected
}
}
52}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

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

Related explainers

package metrics
 
import (
	"sync"

A thread-safe sliding-window average in Go

concurrency sliding-window running-average

Intermediate 8 steps

package logging
 
import (
	"fmt"

Redacting secrets with Go reflection

reflection recursion struct-tags

Advanced 10 steps

java

public class RequestThrottler {
 
    private final Semaphore permits;
    private final long acquireTimeoutMillis;

Bounding concurrency with a Semaphore in Java

concurrency semaphore rate-limiting

Intermediate 6 steps

python

import time
import threading
from flask import Flask, request, jsonify, g

A token-bucket rate limiter in Flask

rate-limiting token-bucket middleware

Intermediate 7 steps

package middleware
 
import (
	"context"

Per-tenant daily rate limiting in Gin

rate-limiting middleware redis

Intermediate 8 steps

ruby

class Document < ApplicationRecord
  class StaleObjectError < StandardError
    def initialize(id)
      super("Document ##{id} was modified by another process")