Go Fan-Out and Fan-In
Fan-out and fan-in is a pattern that spreads jobs across workers and gathers their results into one place. One channel carries the jobs out, and results flow back to a single collector.
Three workers share one jobs channel
Each worker pulls job numbers and writes each square into the slot with the same index. Main waits for the workers and prints the slots in job order.
package main
import (
"fmt"
"sync"
)
func worker(jobs <-chan int, results []int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
results[job] = job * job
}
}
func main() {
jobs := make(chan int)
results := make([]int, 4)
var wg sync.WaitGroup
for w := 0; w < 3; w++ {
wg.Add(1)
go worker(jobs, results, &wg)
}
for job := 0; job < 4; job++ {
jobs <- job
}
close(jobs)
wg.Wait()
for _, result := range results {
fmt.Println(result)
}
}Two channels merge into one
One forwarder goroutine serves each producer channel, and a closer goroutine closes the merged channel at the end. Main collects the values and sorts them before printing.
package main
import (
"fmt"
"sort"
"sync"
)
func produce(values []int, out chan<- int) {
for _, v := range values {
out <- v
}
close(out)
}
func forward(in <-chan int, out chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for v := range in {
out <- v
}
}
func merge(a, b <-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(2)
go forward(a, out, &wg)
go forward(b, out, &wg)
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
a := make(chan int)
b := make(chan int)
go produce([]int{5, 1}, a)
go produce([]int{4, 2}, b)
merged := merge(a, b)
var got []int
for v := range merged {
got = append(got, v)
}
sort.Ints(got)
for _, v := range got {
fmt.Println(v)
}
}Split one source and sum the results
Two workers pull from one jobs channel and send their results into one channel. Main adds every result it receives and prints the total.
package main
import (
"fmt"
"sync"
)
func worker(jobs <-chan int, out chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
out <- job * 10
}
}
func main() {
jobs := make(chan int)
out := make(chan int)
var wg sync.WaitGroup
for w := 0; w < 2; w++ {
wg.Add(1)
go worker(jobs, out, &wg)
}
go func() {
for job := 1; job <= 4; job++ {
jobs <- job
}
close(jobs)
wg.Wait()
close(out)
}()
total := 0
for v := range out {
total += v
}
fmt.Println("Total:", total)
}Example
package main
import (
"fmt"
"sync"
)
type result struct {
job int
value int
}
func worker(jobs <-chan int, results chan<- result, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
results <- result{job: job, value: job * job}
}
}
func main() {
jobs := make(chan int)
results := make(chan result)
var wg sync.WaitGroup
for w := 0; w < 3; w++ {
wg.Add(1)
go worker(jobs, results, &wg)
}
go func() {
for job := 1; job <= 4; job++ {
jobs <- job
}
close(jobs)
wg.Wait()
close(results)
}()
squares := make([]int, 5)
for r := range results {
squares[r.job] = r.value
}
total := 0
for job := 1; job <= 4; job++ {
fmt.Println("Square:", squares[job])
total += squares[job]
}
fmt.Println("Total:", total)
}Key points
- Fan-out sends jobs from one channel to many workers.
- Fan-in gathers results from many channels into one channel.
- Indexed result slots keep results in job order.
- A merged channel closes after every producer channel closes.
- Sorting gathered values gives a fixed order for printing.
In this example, three workers pull four jobs from one channel and send back each square with its job number. Main stores each square in the slot for its job, so the print order matches the job order. The last line adds the four squares into a total.
Note
Main prints the results after the joins, so the printed order never changes.
Try it Yourself
Edit the code, then press Run.
Running is simulated · Go Playground
Exercise
Why do the workers in the first snippet write into indexed slots?