Go⚡Concurrencyintermediate
Pipelines
Go · Concurrency · Run with go run
Explanation
A pipeline is a series of stages connected by channels, where each stage receives values from upstream, transforms them, and sends to downstream. Pipelines make concurrent data processing composable and readable. `done` channels (or `context.Context`) propagate cancellation through the pipeline.
Real-World Scenario
An ETL pipeline: `readCSV` → `parseRows` → `validateRows` → `writeToDatabase`. Each stage runs in its own goroutine, stages communicate via channels, and the pipeline processes records as a stream without loading everything into memory.
Quick Snippet
snippet.go
// Stage 1: generate integers
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
// Stage 2: square each integer
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
// Wire stages together
func main() {
for n := range square(generate(2, 3, 4)) {
fmt.Println(n) // 4, 9, 16
}
}Full Example
pipelines.go
// Compile: go run pipelines.go
// Output:
// 4
// 9
// 16
package main
import "fmt"
// Stage 1: emit numbers
func generate(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
// Stage 2: square each number
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
func main() {
// Wire stages together: generate -> square
for n := range square(generate(2, 3, 4)) {
fmt.Println(n)
}
}
Run command
go run pipelines.go