Go: Go Concurrency Patterns
Last updated: 2026-08-26
Concurrency in Go relies not on libraries, but on patterns—Fan-in, Fan-out, and Pipeline. These combined patterns can solve 90% of concurrency problems.
When you need to process large amounts of data—reading from a generator, having multiple workers process it in parallel, and finally merging the results—how can you design a solution that's elegant? In this lesson, you'll master the Go community's most classic concurrency pattern.
1. You will learn
- Pipeline Pattern
- Fan-out / Fan-in
- Worker Pool
- Or-Done Channel Simplified Mode
- Cancellation Propagation
golang.org/x/sync/errgroupخطأ group- Hands-On: Real-Time Data Processing Pipeline
2. A True Story of a Data Engineer
(1) Pain Point: Processing 10 million log entries in a single خيط took all night
Fatima is a data platform engineer. The company generates 10 million access logs every day that need to be processed in real time:
"The first version used a for حلقة: read a record → parse it → write it to the قاعدة بيانات. On the first day it went live, we found that the processing speed couldn't keep up with the production rate—the log queue was growing by 2 million entries per hour. My boss said, 'Tomorrow's reports depend on this data.' I added more servers, but the code was still single-threaded—the CPU was only using 5% of its capacity."
Her code at the time:
GO
// Bad code: single-threaded, not utilizing CPU
func processLogs(logs []string) {
for _, log := range logs {
parsed := parse(log) // CPU-intensive
enriched := enrich(parsed) // CPU-intensive
save(enriched) // I/O-intensive
}
// 10 million log entries: 3 hours!
}
(2) Go's Approach to Concurrency: Pipeline + Fan-out
GO
// pipeline.go
package main
import (
"fmt"
"sync"
"time"
)
type LogEntry struct {
Raw سلسلة
Level سلسلة
Message سلسلة
Time time.Time
}
func main() {
logs := make([]سلسلة, 100)
for i := range logs {
logs[i] = fmt.Sprintf("log-%d", i)
}
start := time.Now()
// Pipeline: generator → parse → enrich → save
rawCh := generator(logs)
parsedCh := parseStage(rawCh, 4) // 4 parse workers
enrichedCh := enrichStage(parsedCh, 4) // 4 enrich workers
saveStage(enrichedCh, 2) // 2 save workers
fmt.Printf("Processing complete, elapsed: %v\n", time.Since(start))
}
// Stage 1: Generator (data source)
func generator(logs []سلسلة) <-chan سلسلة {
out := make(chan سلسلة, 100)
go func() {
defer close(out)
for _, log := range logs {
out <- log
}
}()
return out
}
// Stage 2: Parse (Fan-out)
func parseStage(in <-chan سلسلة, workers int) <-chan LogEntry {
out := make(chan LogEntry, 100)
for i := 0; i < workers; i++ {
go func() {
for raw := range in {
time.Sleep(1 * time.Millisecond) // Simulate parsing
out <- LogEntry{Raw: raw, Level: "INFO", Message: raw, Time: time.Now()}
}
}()
}
return out
}
// Stage 3: Enrich (Fan-out)
func enrichStage(in <-chan LogEntry, workers int) <-chan LogEntry {
out := make(chan LogEntry, 100)
for i := 0; i < workers; i++ {
go func() {
for entry := range in {
time.Sleep(2 * time.Millisecond) // Simulate data enrichment
out <- entry
}
}()
}
return out
}
// Stage 4: Save (Fan-in to 2 workers)
func saveStage(in <-chan LogEntry, workers int) {
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for entry := range in {
_ = entry // Simulate writing to قاعدة بيانات
time.Sleep(500 * time.Microsecond)
}
}()
}
wg.Wait()
}
(3) Performance: Single-Threaded vs. Pipeline
| Metric | Single-threaded | Pipeline (4+4+2 workers) | Improvement |
|---|---|---|---|
| Processing time for 10 million records | ~3 hours | ~12 minutes | 15x |
| CPU Usage | 5% | 85% | 17x |
| Code Complexity | Simple | Moderate | — |
| Scalability | Add a server | Add a worker (change a number) | — |
3. Pipeline
▶ Example: Numeric Processing Pipeline
GO
package main
import (
"fmt"
)
// Stage 1: Generate
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
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
}
// Stage 3: Double
func double(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * 2
}
}()
return out
}
func main() {
// Pipeline: generate → square → double
for result := range double(square(generate(1, 2, 3, 4, 5))) {
fmt.Println(result)
}
// Output: 2, 8, 18, 32, 50 (n²×2)
}
graph LR
A[generate] -->|chan int| B[square]
B -->|chan int| C[double]
C -->|chan int| D[main]
💡 Tip: The key to a pipeline is that each stage returns a
<-chan T (read-only channel) and accepts a <-chan T as input. Each stage runs in its own goroutine and is connected via channels—this is Go's CSP model.
4. Fan-out / Fan-in
▶ Example: Fan-out + Fan-in
GO
📖 Display only
package main
import (
"fmt"
"sync"
)
// Stage 1: Generate
func generate(nums ...int) <-chan int {
out := make(chan int, 10)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
// Stage 2: Worker (Fan-out target)
func worker(id int, in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
result := n * n
fmt.Printf("Worker %d: %d^2 = %d\n", id, n, result)
out <- result
}
}()
return out
}
// Fan-in: merge multiple channels
func fanIn(channels ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, ch := range channels {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
// Fan-out: one generator distributes to 3 workers
in := generate(1, 2, 3, 4, 5, 6)
w1 := worker(1, in)
w2 := worker(2, in)
w3 := worker(3, in)
// Fan-in: merge results from 3 workers into one channel
results := fanIn(w1, w2, w3)
for result := range results {
fmt.Printf("Result: %d\n", result)
}
}
graph LR
G[Generate] -->|fan-out| W1[Worker 1]
G -->|fan-out| W2[Worker 2]
G -->|fan-out| W3[Worker 3]
W1 -->|fan-in| R[Results]
W2 -->|fan-in| R
W3 -->|fan-in| R
5. Worker Pool
▶ Example: Worker Pool
GO
📖 Display only
package main
import (
"fmt"
"sync"
"time"
)
type Job struct {
ID int
Payload string
}
type Result struct {
JobID int
Output string
Err error
}
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
fmt.Printf("Worker %d processing job %d\n", id, job.ID)
time.Sleep(100 * time.Millisecond) // Simulate work
results <- Result{
JobID: job.ID,
Output: fmt.Sprintf("processed by worker %d", id),
}
}
}
func main() {
const numJobs = 20
const numWorkers = 5
jobs := make(chan Job, numJobs)
results := make(chan Result, numJobs)
// Start Worker Pool
var wg sync.WaitGroup
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go worker(i, jobs, results, &wg)
}
// Send jobs
for i := 0; i < numJobs; i++ {
jobs <- Job{ID: i, Payload: fmt.Sprintf("data-%d", i)}
}
close(jobs)
// Wait for all workers to complete
wg.Wait()
close(results)
// Collect results
for result := range results {
fmt.Printf("Job %d → %s\n", result.JobID, result.Output)
}
}
(2) Pipeline vs Worker Pool vs Fan-out/Fan-in
| Pattern | Core Concept | Applicable Scenarios |
|---|---|---|
| Pipeline | Data flows through multiple stages, with each stage performing a single step | There are clearly defined processing steps (parse → transform → save) |
| Worker Pool | A fixed number of workers retrieve tasks from the job queue | Batch task processing (image compression, email sending) |
| Fan-out/Fan-in | Distribute to multiple workers for parallel processing, then دمج the results | Stateless parallel computation (numerical operations, data filtering) |
6. Or-Done Channel
▶ Example: Or-Done Pattern
GO
📖 Display only
package main
import (
"fmt"
"time"
)
// orDone wraps a channel, making it support cancellation via a done channel
func orDone(done <-chan struct{}, c <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for {
select {
case <-done:
return
case v, ok := <-c:
if !ok {
return
}
select {
case out <- v:
case <-done:
return
}
}
}
}()
return out
}
func generateWithCancel(done <-chan struct{}, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case <-done:
return
case out <- n:
}
}
}()
return out
}
func main() {
done := make(chan struct{})
// Start generator (will be canceled after reaching 5)
nums := generateWithCancel(done, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
// Consume safely via orDone
for n := range orDone(done, nums) {
fmt.Printf("Processing: %d\n", n)
if n >= 5 {
close(done) // Cancel all operations
}
}
fmt.Println("Canceled")
time.Sleep(100 * time.Millisecond) // Wait for goroutines to exit
}
🔥 Common Mistake: Failing to consume from the channel can lead to goroutine leaks—the generator goroutine will block on sending. The Or-Done pattern ensures that no goroutines are leaked upon cancellation. Wrap the channel with Or-Done when both the upstream sender and downstream consumer need to be aware of cancellation.
7. errgroup Error Propagation
⚙️ Prerequisite: Run
go get golang.org/x/sync/errgroupbefore using this package.
▶ Example: errgroup
GO
package main
import (
"fmt"
"time"
"golang.org/x/sync/errgroup"
)
func main() {
// Create an errgroup
// The first goroutine to return an خطأ will cancel the others
g := errgroup.Group{}
// Start 3 goroutines
for i := 0; i < 3; i++ {
id := i
g.Go(func() خطأ {
return doWork(id)
})
}
// Wait for all goroutines to complete, get the first خطأ
if err := g.Wait(); err != nil {
fmt.Printf("Task failed: %v\n", err)
} else {
fmt.Println("All succeeded")
}
}
func doWork(id int) خطأ {
fmt.Printf("Worker %d starting\n", id)
time.Sleep(time.Duration(id+1) * 500 * time.Millisecond)
if id == 1 {
return fmt.Errorf("worker %d failed", id)
}
fmt.Printf("Worker %d complete\n", id)
return nil
}
▶ Example: errgroup + Context timeout
GO
package main
import (
"context"
"fmt"
"time"
"golang.org/x/sync/errgroup"
)
func main() {
// errgroup with Context
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
g, ctx := errgroup.WithContext(ctx)
// Workers check if Context is canceled
for i := 0; i < 5; i++ {
id := i
g.Go(func() error {
select {
case <-time.After(time.Duration(id+1) * time.Second):
fmt.Printf("Worker %d complete\n", id)
return nil
case <-ctx.Done():
fmt.Printf("Worker %d canceled: %v\n", id, ctx.Err())
return ctx.Err()
}
})
}
if err := g.Wait(); err != nil {
fmt.Printf("errgroup stopped: %v\n", err)
}
}
(3) sync.WaitGroup vs errgroup
| Property | sync.WaitGroup | errgroup |
|---|---|---|
| Error Collection | Not Supported | ✅ Return the first خطأ |
| Cancel Propagation | Not Supported | ✅ Automatic Cancellation with WithContext |
| Goroutine Management | Manual Add/Done | ✅ Automatic via Go طريقة |
| Timeout Control | Manual Implementation | ✅ Easily Implemented Using Context |
8. Complete Example: Real-Time Data Processing Pipeline
GO
// data_pipeline.go
package main
import (
"context"
"fmt"
"math/rand"
"sync"
"time"
)
// ---------- Data types ----------
type DataPoint struct {
ID int
Value float64
Timestamp time.Time
}
type ProcessedData struct {
DataPoint
Normalized bool
Anomaly bool
Category string
}
// ---------- Pipeline Stages ----------
// Stage 1: Source (data source)
func source(ctx context.Context, count int) <-chan DataPoint {
out := make(chan DataPoint, 100)
go func() {
defer close(out)
for i := 0; i < count; i++ {
select {
case <-ctx.Done():
return
case out <- DataPoint{
ID: i,
Value: rand.Float64() * 100,
Timestamp: time.Now(),
}:
}
}
}()
return out
}
// Stage 2: Normalize (Fan-out)
func normalize(ctx context.Context, in <-chan DataPoint, workers int) <-chan ProcessedData {
out := make(chan ProcessedData, 100)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for dp := range orDoneDataPoint(ctx.Done(), in) {
select {
case <-ctx.Done():
return
case out <- ProcessedData{
DataPoint: dp,
Normalized: true,
Category: categorize(dp.Value),
}:
}
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
// Stage 3: Detect Anomaly (Fan-out)
func detectAnomaly(ctx context.Context, in <-chan ProcessedData, workers int) <-chan ProcessedData {
out := make(chan ProcessedData, 100)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for pd := range orDoneProcessedData(ctx.Done(), in) {
pd.Anomaly = pd.Value > 90 || pd.Value < 10
select {
case <-ctx.Done():
return
case out <- pd:
}
}
}()
}
go func() {
wg.Wait()
close(out)
}()
return out
}
// Stage 4: Sink (output, Fan-in)
func sink(ctx context.Context, in <-chan ProcessedData) {
var normalCount, anomalyCount int
for pd := range orDoneProcessedData(ctx.Done(), in) {
if pd.Anomaly {
anomalyCount++
fmt.Printf("[ANOMALY] ID=%d, Value=%.2f, Category=%s\n",
pd.ID, pd.Value, pd.Category)
} else {
normalCount++
}
}
fmt.Printf("\nSummary: normal=%d, anomaly=%d\n", normalCount, anomalyCount)
}
// ---------- Helper functions ----------
func orDoneDataPoint(done <-chan struct{}, in <-chan DataPoint) <-chan DataPoint {
out := make(chan DataPoint)
go func() {
defer close(out)
for {
select {
case <-done:
return
case v, ok := <-in:
if !ok {
return
}
select {
case out <- v:
case <-done:
return
}
}
}
}()
return out
}
func orDoneProcessedData(done <-chan struct{}, in <-chan ProcessedData) <-chan ProcessedData {
out := make(chan ProcessedData)
go func() {
defer close(out)
for {
select {
case <-done:
return
case v, ok := <-in:
if !ok {
return
}
select {
case out <- v:
case <-done:
return
}
}
}
}()
return out
}
func categorize(value float64) string {
switch {
case value < 30:
return "low"
case value < 70:
return "medium"
default:
return "high"
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
fmt.Println("Starting real-time data processing pipeline...")
// Build Pipeline
data := source(ctx, 1000)
normalized := normalize(ctx, data, 5) // 5 normalize workers
detected := detectAnomaly(ctx, normalized, 3) // 3 anomaly detection workers
sink(ctx, detected)
fmt.Println("Pipeline processing complete")
}
💡 Tip: In a production environment, each stage of the pipeline should have its own buffer size, retry logic, and monitoring metrics. You can use
expvar or prometheus to expose each stage's processing speed, queue length, and error rate.
❓ FAQ
Q What is the difference between a Pipeline and a Worker Pool?
A A Pipeline is a data flow that passes through multiple processing stages, with each stage performing a single step (e.g., parsing → transforming → saving), and the data shape may change. A Worker Pool consists of a fixed number of workers that retrieve tasks from the same job queue, and the data format remains unchanged (e.g., image compression).
Q What are the core concepts behind fan-out and fan-in?
A Fan-out distributes data from a single channel to multiple workers for parallel processing (to increase throughput). Fan-in merges the results from multiple channels into a single channel (for unified consumption). The two are typically used in tandem.
Q When must you use Or-Done?
A When reading from a channel whose goroutine lifecycle you cannot control (such as a channel returned by a third-party library). Or-Done ensures that the reading goroutine will not block on channel operations if the caller cancels. For simple scenarios, you can use
select with a done channel.Q How do I choose between
errgroup and sync.WaitGroup?A Use
errgroup when you need to collect errors or propagate cancellations. Use sync.WaitGroup (which is lighter-weight) when you just need to wait for a goroutine to finish. errgroup is a superset of WaitGroup in terms of functionality, but it introduces an additional goroutine to manage cancellations.Q How do you determine the buffer size for a channel in a pipeline?
A The basic principle is to ensure that producers do not get blocked by consumers. A buffer that is too large wastes memory, while one that is too small increases goroutine context switching. Rule of thumb: Buffer size = production rate × desired latency. Determine the actual size through load testing in production.
Q How can goroutine leaks be prevented?
A Three rules: (1) Wherever a goroutine is created, there must be an exit mechanism (closing the channel or a
done signal); (2) Use errgroup or WaitGroup to manage the lifecycle; (3) Use Or-Done to wrap uncontrolled channels. Never create a goroutine if you don't know when it will exit.Q Can these patterns be used in combination?
A Yes, and they should be. For example, within each stage of a Pipeline, you can use a Worker Pool combined with Fan-out/Fan-in. An errgroup can serve as a Pipeline orchestrator to centrally manage cancellations and errors.
📖 Summary
- Pipeline: Data flows through multiple processing stages, with one goroutine per stage
- Fan-out: A single channel is distributed to multiple workers (for parallel processing)
- Fan-in: Merging multiple channels into a single channel (for unified consumption)
- Worker Pool: A fixed number of workers retrieve tasks from the job queue
- Or-Done: Safely consume uncontrollable channels with cancellation support
- errgroup: Manages goroutine groups, supporting error collection and cancellation propagation
- Combination mode: Pipeline uses a Worker Pool + Fan-out/Fan-in internally
📝 Exercises
-
Basic Problem (Difficulty ⭐): Implement a 3-stage numeric processing pipeline: generate(1–100) → filter (keep only even numbers) → sum (accumulate). Each stage runs in a separate goroutine and is connected via a channel.
-
Advanced Problem (Difficulty ⭐⭐): Implement a concurrent log processing system: 10 parse workers (fan-out) + 3 write workers (fan-in) + errgroup error handling. Simulate 1,000 log entries and output statistics (total elapsed time, number of entries processed, number of errors).
-
Challenge (Difficulty: ⭐⭐⭐): Design and implement a worker pool with load balancing and dynamic scaling. Requirements: (1) Start with 5 workers; (2) Automatically increase the number of workers (up to 20) when the job queue backlog exceeds a threshold; (3) Reduce the number of workers (down to 2) when the queue is empty; (4) Use
Contextto control worker termination; (5) Use-raceto verify that there is no race condition.
