Go Pipelines
A pipeline is a chain of stages joined by channels. Each stage receives values from one channel and sends changed values to the next channel.
A generator feeds a squarer
A stage is one step in a pipeline that receives values and sends changed values on. The generate stage makes the values, and the square stage changes each one.
package main
import "fmt"
func generate(out chan<- int) {
for _, number := range []int{2, 3, 4} {
out <- number
}
close(out)
}
func square(in <-chan int, out chan<- int) {
for number := range in {
out <- number * number
}
close(out)
}
func main() {
first := make(chan int)
second := make(chan int)
go generate(first)
go square(first, second)
for value := range second {
fmt.Println(value)
}
}A filter stage keeps the even squares
A filter stage sends on only the values that pass its test. The square stage here makes even and odd squares, and the filter keeps the even ones.
package main
import "fmt"
func generate(out chan<- int) {
for _, number := range []int{1, 2, 3, 4, 5, 6} {
out <- number
}
close(out)
}
func square(in <-chan int, out chan<- int) {
for number := range in {
out <- number * number
}
close(out)
}
func keepEven(in <-chan int, out chan<- int) {
for number := range in {
if number%2 == 0 {
out <- number
}
}
close(out)
}
func main() {
first := make(chan int)
second := make(chan int)
third := make(chan int)
go generate(first)
go square(first, second)
go keepEven(second, third)
for value := range third {
fmt.Println(value)
}
}Closing ends the chain and a total prints
Each stage closes its output when its input ends. The main range then ends, and the total prints after the loop.
package main
import "fmt"
func generate(out chan<- int) {
for _, number := range []int{5, 10, 15} {
out <- number
}
close(out)
}
func double(in <-chan int, out chan<- int) {
for number := range in {
out <- number * 2
}
close(out)
}
func main() {
first := make(chan int)
second := make(chan int)
go generate(first)
go double(first, second)
sum := 0
for value := range second {
sum += value
}
fmt.Println("Total:", sum)
}Example
package main
import "fmt"
func generate(out chan<- int) {
for _, number := range []int{1, 2, 3, 4} {
out <- number
}
close(out)
}
func cube(in <-chan int, out chan<- int) {
for number := range in {
out <- number * number * number
}
close(out)
}
func main() {
first := make(chan int)
second := make(chan int)
go generate(first)
go cube(first, second)
sum := 0
for value := range second {
sum += value
}
fmt.Println("Sum:", sum)
}Key points
- A pipeline joins stages in a line with channels.
- Each stage runs in its own goroutine.
- A stage closes its output after its input closes and empties.
- Closing the first channel brings the whole chain to an end.
- Values keep their order from one stage to the next.
- The last channel hands the final values to main.
In this example, the generate stage sends 1, 2, 3, and 4 into the first channel and closes it. The cube stage reads each value and sends its cube into the second channel. It closes that channel when its input ends. Main adds the cubes as they arrive, and the sum prints as 100 after the range ends.
Note
These pipelines need no WaitGroup, because closing each channel carries the end signal.
Try it Yourself
Edit the code, then press Run.
Running is simulated · Go Playground
Exercise
What brings a pipeline to a clean end?