Go Advanced Concurrency Patterns

calendar_today August 3, 2025 schedule 20 min read

Module: 4.3 - Advanced Concurrency Patterns
Duration: 3-4 days (8-10 hours)
Prerequisites: Go Basics, Goroutines Fundamentals, Channels and Communication
Start Date: August 3, 2025

🎯 Learning Objectives

By the end of this module, you will be able to:

Implement worker pool patterns for efficient resource utilization
Design fan-in and fan-out patterns for parallel processing
Use the context package for cancellation and timeouts
Build scalable concurrent applications with proper error handling
Apply advanced synchronization techniques
Optimize concurrent code for performance

1. Worker Pool Pattern

Overview

The worker pool pattern is essential for controlling the number of goroutines and efficiently processing large numbers of tasks.

Basic Worker Pool Implementation

package main

import (
    "fmt"
    "sync"
    "time"
)

// Job represents a unit of work
type Job struct {
    ID     int
    Data   string
}

// Result represents the outcome of processing a job
type Result struct {
    Job    Job
    Output string
    Error  error
}

// Worker processes jobs from the jobs channel
func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        
        // Simulate work
        time.Sleep(time.Millisecond * 100)
        
        result := Result{
            Job:    job,
            Output: fmt.Sprintf("Processed by worker %d: %s", id, job.Data),
            Error:  nil,
        }
        
        results <- result
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 10
    
    // Create channels
    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
    
    // Start workers
    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }
    
    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("Task %d", j)}
    }
    close(jobs)
    
    // Collect results
    for r := 1; r <= numJobs; r++ {
        result := <-results
        fmt.Printf("Result: %s\n", result.Output)
    }
}

Advanced Worker Pool with Context

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type WorkerPool struct {
    workers    int
    jobs       chan Job
    results    chan Result
    wg         sync.WaitGroup
    ctx        context.Context
    cancel     context.CancelFunc
}

func NewWorkerPool(workers int) *WorkerPool {
    ctx, cancel := context.WithCancel(context.Background())
    return &WorkerPool{
        workers: workers,
        jobs:    make(chan Job, workers*2),
        results: make(chan Result, workers*2),
        ctx:     ctx,
        cancel:  cancel,
    }
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.workers; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for {
        select {
        case job, ok := <-wp.jobs:
            if !ok {
                return
            }
            
            // Process job with timeout
            result := wp.processJob(job)
            
            select {
            case wp.results <- result:
            case <-wp.ctx.Done():
                return
            }
            
        case <-wp.ctx.Done():
            return
        }
    }
}

func (wp *WorkerPool) processJob(job Job) Result {
    // Simulate work with potential timeout
    done := make(chan Result, 1)
    
    go func() {
        time.Sleep(time.Millisecond * 50) // Simulate work
        done <- Result{
            Job:    job,
            Output: fmt.Sprintf("Processed: %s", job.Data),
            Error:  nil,
        }
    }()
    
    select {
    case result := <-done:
        return result
    case <-time.After(time.Millisecond * 100):
        return Result{
            Job:   job,
            Error: fmt.Errorf("job %d timed out", job.ID),
        }
    case <-wp.ctx.Done():
        return Result{
            Job:   job,
            Error: wp.ctx.Err(),
        }
    }
}

func (wp *WorkerPool) Submit(job Job) error {
    select {
    case wp.jobs <- job:
        return nil
    case <-wp.ctx.Done():
        return wp.ctx.Err()
    default:
        return fmt.Errorf("job queue full")
    }
}

func (wp *WorkerPool) Results() <-chan Result {
    return wp.results
}

func (wp *WorkerPool) Stop() {
    close(wp.jobs)
    wp.wg.Wait()
    close(wp.results)
    wp.cancel()
}

2. Fan-In and Fan-Out Patterns

Fan-Out Pattern

Distributing work across multiple goroutines.

package main

import (
    "fmt"
    "sync"
)

// fanOut distributes input across multiple channels
func fanOut(input <-chan int, workers int) []<-chan int {
    outputs := make([]<-chan int, workers)
    
    for i := 0; i < workers; i++ {
        output := make(chan int)
        outputs[i] = output
        
        go func(out chan<- int, workerID int) {
            defer close(out)
            for n := range input {
                // Distribute based on worker ID (round-robin style)
                if n%workers == workerID {
                    out <- n
                }
            }
        }(output, i)
    }
    
    return outputs
}

// processor simulates work on a number
func processor(input <-chan int, output chan<- int, workerID int) {
    defer close(output)
    
    for n := range input {
        // Simulate processing
        result := n * n
        fmt.Printf("Worker %d: %d -> %d\n", workerID, n, result)
        output <- result
    }
}

Fan-In Pattern

Combining results from multiple goroutines.

// fanIn combines multiple input channels into one output channel
func fanIn(inputs []<-chan int) <-chan int {
    output := make(chan int)
    var wg sync.WaitGroup
    
    // Start a goroutine for each input channel
    for i, input := range inputs {
        wg.Add(1)
        go func(in <-chan int, workerID int) {
            defer wg.Done()
            for n := range in {
                select {
                case output <- n:
                    fmt.Printf("Fan-in received %d from worker %d\n", n, workerID)
                }
            }
        }(input, i)
    }
    
    // Close output channel when all inputs are done
    go func() {
        wg.Wait()
        close(output)
    }()
    
    return output
}

// Complete fan-out/fan-in example
func fanOutFanInExample() {
    // Input data
    input := make(chan int, 10)
    go func() {
        defer close(input)
        for i := 1; i <= 20; i++ {
            input <- i
        }
    }()
    
    // Fan-out to multiple processors
    const numWorkers = 3
    outputs := make([]<-chan int, numWorkers)
    
    for i := 0; i < numWorkers; i++ {
        output := make(chan int)
        outputs[i] = output
        go processor(input, output, i)
    }
    
    // Fan-in results
    results := fanIn(outputs)
    
    // Collect all results
    var allResults []int
    for result := range results {
        allResults = append(allResults, result)
    }
    
    fmt.Printf("Total results collected: %d\n", len(allResults))
}

3. Context Package for Cancellation

Basic Context Usage

package main

import (
    "context"
    "fmt"
    "time"
)

// longRunningTask simulates a task that can be cancelled
func longRunningTask(ctx context.Context, id int) error {
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():
            fmt.Printf("Task %d cancelled: %v\n", id, ctx.Err())
            return ctx.Err()
        case <-time.After(time.Second):
            fmt.Printf("Task %d: step %d completed\n", id, i+1)
        }
    }
    fmt.Printf("Task %d completed successfully\n", id)
    return nil
}

func contextExample() {
    // Context with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    
    go longRunningTask(ctx, 1)
    go longRunningTask(ctx, 2)
    
    // Wait for context to be done
    <-ctx.Done()
    fmt.Println("Main: context done")
    
    // Give goroutines time to clean up
    time.Sleep(time.Millisecond * 100)
}

Context with Values

type contextKey string

const (
    userIDKey    contextKey = "userID"
    requestIDKey contextKey = "requestID"
)

func processWithContext(ctx context.Context) {
    userID := ctx.Value(userIDKey)
    requestID := ctx.Value(requestIDKey)
    
    fmt.Printf("Processing request %v for user %v\n", requestID, userID)
    
    // Simulate work with cancellation check
    select {
    case <-time.After(time.Second):
        fmt.Println("Work completed")
    case <-ctx.Done():
        fmt.Printf("Work cancelled: %v\n", ctx.Err())
    }
}

func contextWithValuesExample() {
    ctx := context.Background()
    ctx = context.WithValue(ctx, userIDKey, "user123")
    ctx = context.WithValue(ctx, requestIDKey, "req456")
    
    // Add timeout
    ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
    defer cancel()
    
    processWithContext(ctx)
}

Context Propagation in HTTP-like Services

func handleRequest(ctx context.Context, request string) error {
    // Create child context with timeout
    childCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()
    
    // Chain of operations
    if err := authenticate(childCtx); err != nil {
        return err
    }
    
    if err := authorize(childCtx); err != nil {
        return err
    }
    
    return processRequest(childCtx, request)
}

func authenticate(ctx context.Context) error {
    select {
    case <-time.After(time.Millisecond * 100):
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func authorize(ctx context.Context) error {
    select {
    case <-time.After(time.Millisecond * 150):
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func processRequest(ctx context.Context, request string) error {
    select {
    case <-time.After(time.Second):
        fmt.Printf("Request processed: %s\n", request)
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

4. Pipeline Pattern

Sequential Pipeline

package main

import (
    "fmt"
    "sync"
)

// Stage represents a pipeline stage
type Stage func(<-chan interface{}) <-chan interface{}

// pipeline combines multiple stages
func pipeline(stages ...Stage) Stage {
    return func(input <-chan interface{}) <-chan interface{} {
        output := input
        for _, stage := range stages {
            output = stage(output)
        }
        return output
    }
}

// Example stages
func generator(nums []int) <-chan interface{} {
    output := make(chan interface{})
    go func() {
        defer close(output)
        for _, n := range nums {
            output <- n
        }
    }()
    return output
}

func square() Stage {
    return func(input <-chan interface{}) <-chan interface{} {
        output := make(chan interface{})
        go func() {
            defer close(output)
            for n := range input {
                if num, ok := n.(int); ok {
                    output <- num * num
                }
            }
        }()
        return output
    }
}

func filter(predicate func(interface{}) bool) Stage {
    return func(input <-chan interface{}) <-chan interface{} {
        output := make(chan interface{})
        go func() {
            defer close(output)
            for item := range input {
                if predicate(item) {
                    output <- item
                }
            }
        }()
        return output
    }
}

func pipelineExample() {
    // Create pipeline
    numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    
    // Pipeline: generate -> square -> filter even numbers
    pipe := pipeline(
        square(),
        filter(func(x interface{}) bool {
            if num, ok := x.(int); ok {
                return num%2 == 0
            }
            return false
        }),
    )
    
    // Run pipeline
    input := generator(numbers)
    output := pipe(input)
    
    // Collect results
    for result := range output {
        fmt.Printf("Result: %v\n", result)
    }
}

Parallel Pipeline with Worker Pools

func parallelStage(workers int, work func(interface{}) interface{}) Stage {
    return func(input <-chan interface{}) <-chan interface{} {
        output := make(chan interface{})
        var wg sync.WaitGroup
        
        // Start workers
        for i := 0; i < workers; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                for item := range input {
                    result := work(item)
                    output <- result
                }
            }()
        }
        
        // Close output when all workers are done
        go func() {
            wg.Wait()
            close(output)
        }()
        
        return output
    }
}

func parallelPipelineExample() {
    numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    
    // Heavy computation stage with multiple workers
    heavyComputation := parallelStage(3, func(x interface{}) interface{} {
        if num, ok := x.(int); ok {
            // Simulate heavy work
            time.Sleep(time.Millisecond * 100)
            return num * num * num
        }
        return x
    })
    
    input := generator(numbers)
    output := heavyComputation(input)
    
    for result := range output {
        fmt.Printf("Computed: %v\n", result)
    }
}

5. Rate Limiting and Throttling

Token Bucket Rate Limiter

package main

import (
    "fmt"
    "sync"
    "time"
)

type TokenBucket struct {
    tokens   int
    capacity int
    refill   int
    interval time.Duration
    mu       sync.Mutex
    lastRefill time.Time
}

func NewTokenBucket(capacity, refill int, interval time.Duration) *TokenBucket {
    return &TokenBucket{
        tokens:     capacity,
        capacity:   capacity,
        refill:     refill,
        interval:   interval,
        lastRefill: time.Now(),
    }
}

func (tb *TokenBucket) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()
    
    // Refill tokens based on elapsed time
    now := time.Now()
    elapsed := now.Sub(tb.lastRefill)
    if elapsed >= tb.interval {
        periods := int(elapsed / tb.interval)
        tokensToAdd := periods * tb.refill
        tb.tokens = min(tb.capacity, tb.tokens+tokensToAdd)
        tb.lastRefill = now
    }
    
    // Check if we can consume a token
    if tb.tokens > 0 {
        tb.tokens--
        return true
    }
    
    return false
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

// Rate limited worker
func rateLimitedWorker(id int, rateLimiter *TokenBucket, jobs <-chan Job) {
    for job := range jobs {
        // Wait for rate limiter to allow request
        for !rateLimiter.Allow() {
            time.Sleep(time.Millisecond * 10)
        }
        
        fmt.Printf("Worker %d processing job %d at %v\n", 
            id, job.ID, time.Now().Format("15:04:05.000"))
        
        // Simulate work
        time.Sleep(time.Millisecond * 50)
    }
}

Channel-Based Rate Limiter

type ChannelRateLimiter struct {
    limiter chan struct{}
    ticker  *time.Ticker
}

func NewChannelRateLimiter(rate int, per time.Duration) *ChannelRateLimiter {
    limiter := make(chan struct{}, rate)
    
    // Fill the initial bucket
    for i := 0; i < rate; i++ {
        limiter <- struct{}{}
    }
    
    ticker := time.NewTicker(per / time.Duration(rate))
    
    rl := &ChannelRateLimiter{
        limiter: limiter,
        ticker:  ticker,
    }
    
    // Refill tokens
    go func() {
        for range ticker.C {
            select {
            case limiter <- struct{}{}:
            default:
                // Bucket is full
            }
        }
    }()
    
    return rl
}

func (rl *ChannelRateLimiter) Wait() {
    <-rl.limiter
}

func (rl *ChannelRateLimiter) Stop() {
    rl.ticker.Stop()
}

func channelRateLimiterExample() {
    rateLimiter := NewChannelRateLimiter(5, time.Second) // 5 requests per second
    defer rateLimiter.Stop()
    
    for i := 0; i < 20; i++ {
        go func(id int) {
            rateLimiter.Wait()
            fmt.Printf("Request %d processed at %v\n", 
                id, time.Now().Format("15:04:05.000"))
        }(i)
    }
    
    time.Sleep(5 * time.Second)
}

6. Advanced Synchronization

Once Pattern for Initialization

package main

import (
    "fmt"
    "sync"
    "time"
)

type Config struct {
    data map[string]string
}

var (
    config *Config
    once   sync.Once
)

func getConfig() *Config {
    once.Do(func() {
        fmt.Println("Initializing config...")
        time.Sleep(time.Millisecond * 100) // Simulate expensive initialization
        config = &Config{
            data: map[string]string{
                "database_url": "localhost:5432",
                "cache_url":    "localhost:6379",
            },
        }
        fmt.Println("Config initialized")
    })
    return config
}

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    
    cfg := getConfig()
    fmt.Printf("Worker %d got config: %v\n", id, cfg.data["database_url"])
}

func onceExample() {
    var wg sync.WaitGroup
    
    // Start multiple workers that need config
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
    
    wg.Wait()
}

Condition Variables for Complex Synchronization

package main

import (
    "fmt"
    "sync"
    "time"
)

type Buffer struct {
    data     []int
    capacity int
    mu       sync.Mutex
    notEmpty *sync.Cond
    notFull  *sync.Cond
}

func NewBuffer(capacity int) *Buffer {
    b := &Buffer{
        data:     make([]int, 0, capacity),
        capacity: capacity,
    }
    b.notEmpty = sync.NewCond(&b.mu)
    b.notFull = sync.NewCond(&b.mu)
    return b
}

func (b *Buffer) Put(item int) {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    // Wait while buffer is full
    for len(b.data) == b.capacity {
        b.notFull.Wait()
    }
    
    b.data = append(b.data, item)
    fmt.Printf("Put %d, buffer size: %d\n", item, len(b.data))
    
    // Signal that buffer is not empty
    b.notEmpty.Signal()
}

func (b *Buffer) Get() int {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    // Wait while buffer is empty
    for len(b.data) == 0 {
        b.notEmpty.Wait()
    }
    
    item := b.data[0]
    b.data = b.data[1:]
    fmt.Printf("Got %d, buffer size: %d\n", item, len(b.data))
    
    // Signal that buffer is not full
    b.notFull.Signal()
    
    return item
}

func conditionVariableExample() {
    buffer := NewBuffer(3)
    
    // Start producers
    for i := 1; i <= 2; i++ {
        go func(id int) {
            for j := 0; j < 5; j++ {
                buffer.Put(id*10 + j)
                time.Sleep(time.Millisecond * 200)
            }
        }(i)
    }
    
    // Start consumers
    for i := 1; i <= 2; i++ {
        go func(id int) {
            for j := 0; j < 5; j++ {
                item := buffer.Get()
                fmt.Printf("Consumer %d got: %d\n", id, item)
                time.Sleep(time.Millisecond * 300)
            }
        }(i)
    }
    
    time.Sleep(5 * time.Second)
}

7. Error Handling in Concurrent Code

Error Aggregation Pattern

package main

import (
    "errors"
    "fmt"
    "strings"
    "sync"
)

type ErrorCollector struct {
    errors []error
    mu     sync.Mutex
}

func (ec *ErrorCollector) Add(err error) {
    if err != nil {
        ec.mu.Lock()
        ec.errors = append(ec.errors, err)
        ec.mu.Unlock()
    }
}

func (ec *ErrorCollector) HasErrors() bool {
    ec.mu.Lock()
    defer ec.mu.Unlock()
    return len(ec.errors) > 0
}

func (ec *ErrorCollector) Error() error {
    ec.mu.Lock()
    defer ec.mu.Unlock()
    
    if len(ec.errors) == 0 {
        return nil
    }
    
    var messages []string
    for _, err := range ec.errors {
        messages = append(messages, err.Error())
    }
    
    return errors.New(strings.Join(messages, "; "))
}

func riskyTask(id int) error {
    if id%3 == 0 {
        return fmt.Errorf("task %d failed", id)
    }
    return nil
}

func errorAggregationExample() {
    var wg sync.WaitGroup
    errorCollector := &ErrorCollector{}
    
    // Run tasks that might fail
    for i := 1; i <= 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            if err := riskyTask(id); err != nil {
                errorCollector.Add(err)
            }
        }(i)
    }
    
    wg.Wait()
    
    if errorCollector.HasErrors() {
        fmt.Printf("Errors occurred: %v\n", errorCollector.Error())
    } else {
        fmt.Println("All tasks completed successfully")
    }
}

Circuit Breaker Pattern

package main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

type State int

const (
    Closed State = iota
    Open
    HalfOpen
)

type CircuitBreaker struct {
    name          string
    maxFailures   int
    resetTimeout  time.Duration
    state         State
    failures      int
    lastFailTime  time.Time
    mu            sync.RWMutex
}

func NewCircuitBreaker(name string, maxFailures int, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        name:         name,
        maxFailures:  maxFailures,
        resetTimeout: resetTimeout,
        state:        Closed,
    }
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    // Check if we should try resetting
    if cb.state == Open && time.Since(cb.lastFailTime) > cb.resetTimeout {
        cb.state = HalfOpen
        cb.failures = 0
    }
    
    // Reject calls if circuit is open
    if cb.state == Open {
        return errors.New("circuit breaker is open")
    }
    
    // Execute the function
    err := fn()
    
    if err != nil {
        cb.onFailure()
        return err
    }
    
    cb.onSuccess()
    return nil
}

func (cb *CircuitBreaker) onFailure() {
    cb.failures++
    cb.lastFailTime = time.Now()
    
    if cb.failures >= cb.maxFailures {
        cb.state = Open
        fmt.Printf("Circuit breaker %s opened\n", cb.name)
    }
}

func (cb *CircuitBreaker) onSuccess() {
    if cb.state == HalfOpen {
        cb.state = Closed
        fmt.Printf("Circuit breaker %s closed\n", cb.name)
    }
    cb.failures = 0
}

func (cb *CircuitBreaker) GetState() State {
    cb.mu.RLock()
    defer cb.mu.RUnlock()
    return cb.state
}

// Unreliable service simulation
func unreliableService(failureRate float64) func() error {
    calls := 0
    return func() error {
        calls++
        if float64(calls%10) < failureRate*10 {
            return errors.New("service failed")
        }
        return nil
    }
}

func circuitBreakerExample() {
    cb := NewCircuitBreaker("test-service", 3, 2*time.Second)
    service := unreliableService(0.7) // 70% failure rate
    
    for i := 0; i < 20; i++ {
        err := cb.Call(service)
        if err != nil {
            fmt.Printf("Call %d failed: %v\n", i+1, err)
        } else {
            fmt.Printf("Call %d succeeded\n", i+1)
        }
        time.Sleep(time.Millisecond * 300)
    }
}

8. Performance Optimization

Memory Pool Pattern

package main

import (
    "sync"
)

type ObjectPool struct {
    pool sync.Pool
}

func NewObjectPool(newFunc func() interface{}) *ObjectPool {
    return &ObjectPool{
        pool: sync.Pool{
            New: newFunc,
        },
    }
}

func (p *ObjectPool) Get() interface{} {
    return p.pool.Get()
}

func (p *ObjectPool) Put(obj interface{}) {
    p.pool.Put(obj)
}

// Example: Buffer pool
type Buffer struct {
    data []byte
}

func (b *Buffer) Reset() {
    b.data = b.data[:0]
}

var bufferPool = NewObjectPool(func() interface{} {
    return &Buffer{
        data: make([]byte, 0, 1024), // Pre-allocate 1KB
    }
})

func processData(data []byte) []byte {
    // Get buffer from pool
    buf := bufferPool.Get().(*Buffer)
    defer func() {
        buf.Reset()
        bufferPool.Put(buf)
    }()
    
    // Use buffer for processing
    buf.data = append(buf.data, data...)
    buf.data = append(buf.data, []byte(" processed")...)
    
    // Return copy since we're returning the buffer to pool
    result := make([]byte, len(buf.data))
    copy(result, buf.data)
    return result
}

Lock-Free Data Structures

package main

import (
    "sync/atomic"
    "unsafe"
)

// Lock-free queue using atomic operations
type LockFreeQueue struct {
    head unsafe.Pointer
    tail unsafe.Pointer
}

type queueNode struct {
    data interface{}
    next unsafe.Pointer
}

func NewLockFreeQueue() *LockFreeQueue {
    node := &queueNode{}
    q := &LockFreeQueue{
        head: unsafe.Pointer(node),
        tail: unsafe.Pointer(node),
    }
    return q
}

func (q *LockFreeQueue) Enqueue(data interface{}) {
    node := &queueNode{data: data}
    
    for {
        tail := (*queueNode)(atomic.LoadPointer(&q.tail))
        next := (*queueNode)(atomic.LoadPointer(&tail.next))
        
        if tail == (*queueNode)(atomic.LoadPointer(&q.tail)) {
            if next == nil {
                if atomic.CompareAndSwapPointer(&tail.next, unsafe.Pointer(next), unsafe.Pointer(node)) {
                    atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(node))
                    break
                }
            } else {
                atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
            }
        }
    }
}

func (q *LockFreeQueue) Dequeue() interface{} {
    for {
        head := (*queueNode)(atomic.LoadPointer(&q.head))
        tail := (*queueNode)(atomic.LoadPointer(&q.tail))
        next := (*queueNode)(atomic.LoadPointer(&head.next))
        
        if head == (*queueNode)(atomic.LoadPointer(&q.head)) {
            if head == tail {
                if next == nil {
                    return nil // Queue is empty
                }
                atomic.CompareAndSwapPointer(&q.tail, unsafe.Pointer(tail), unsafe.Pointer(next))
            } else {
                data := next.data
                if atomic.CompareAndSwapPointer(&q.head, unsafe.Pointer(head), unsafe.Pointer(next)) {
                    return data
                }
            }
        }
    }
}

9. Practical Exercises

Exercise 1: Web Scraper with Rate Limiting

Build a concurrent web scraper that:

Uses worker pools to process URLs
Implements rate limiting to be respectful
Handles timeouts and cancellation
Aggregates results and errors

package main

import (
    "context"
    "fmt"
    "io"
    "net/http"
    "sync"
    "time"
)

type ScrapingResult struct {
    URL     string
    Content string
    Error   error
}

type WebScraper struct {
    client      *http.Client
    rateLimiter *ChannelRateLimiter
    workers     int
}

func NewWebScraper(workers int, requestsPerSecond int) *WebScraper {
    return &WebScraper{
        client: &http.Client{
            Timeout: 10 * time.Second,
        },
        rateLimiter: NewChannelRateLimiter(requestsPerSecond, time.Second),
        workers:     workers,
    }
}

func (ws *WebScraper) Scrape(ctx context.Context, urls []string) []ScrapingResult {
    jobs := make(chan string, len(urls))
    results := make(chan ScrapingResult, len(urls))
    
    // Start workers
    var wg sync.WaitGroup
    for i := 0; i < ws.workers; i++ {
        wg.Add(1)
        go ws.worker(ctx, jobs, results, &wg)
    }
    
    // Send jobs
    go func() {
        defer close(jobs)
        for _, url := range urls {
            select {
            case jobs <- url:
            case <-ctx.Done():
                return
            }
        }
    }()
    
    // Close results when workers are done
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // Collect results
    var allResults []ScrapingResult
    for result := range results {
        allResults = append(allResults, result)
    }
    
    return allResults
}

func (ws *WebScraper) worker(ctx context.Context, jobs <-chan string, results chan<- ScrapingResult, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for {
        select {
        case url, ok := <-jobs:
            if !ok {
                return
            }
            
            // Rate limit
            ws.rateLimiter.Wait()
            
            // Scrape with timeout
            result := ws.scrapeURL(ctx, url)
            
            select {
            case results <- result:
            case <-ctx.Done():
                return
            }
            
        case <-ctx.Done():
            return
        }
    }
}

func (ws *WebScraper) scrapeURL(ctx context.Context, url string) ScrapingResult {
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return ScrapingResult{URL: url, Error: err}
    }
    
    resp, err := ws.client.Do(req)
    if err != nil {
        return ScrapingResult{URL: url, Error: err}
    }
    defer resp.Body.Close()
    
    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return ScrapingResult{URL: url, Error: err}
    }
    
    return ScrapingResult{
        URL:     url,
        Content: string(body[:min(len(body), 200)]), // First 200 chars
        Error:   nil,
    }
}

func scraperExample() {
    urls := []string{
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/404",
    }
    
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    
    scraper := NewWebScraper(2, 1) // 2 workers, 1 request per second
    results := scraper.Scrape(ctx, urls)
    
    for _, result := range results {
        if result.Error != nil {
            fmt.Printf("Error scraping %s: %v\n", result.URL, result.Error)
        } else {
            fmt.Printf("Scraped %s: %d bytes\n", result.URL, len(result.Content))
        }
    }
}

Exercise 2: Real-time Data Processor

Build a system that:

Processes streaming data using pipelines
Implements backpressure handling
Provides metrics and monitoring
Handles different data types

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "math/rand"
    "sync"
    "sync/atomic"
    "time"
)

type DataPoint struct {
    ID        string      `json:"id"`
    Timestamp time.Time   `json:"timestamp"`
    Value     interface{} `json:"value"`
    Type      string      `json:"type"`
}

type ProcessingMetrics struct {
    Processed int64
    Errors    int64
    Dropped   int64
}

type DataProcessor struct {
    input       chan DataPoint
    output      chan DataPoint
    bufferSize  int
    metrics     ProcessingMetrics
    ctx         context.Context
    cancel      context.CancelFunc
    wg          sync.WaitGroup
}

func NewDataProcessor(bufferSize int) *DataProcessor {
    ctx, cancel := context.WithCancel(context.Background())
    return &DataProcessor{
        input:      make(chan DataPoint, bufferSize),
        output:     make(chan DataPoint, bufferSize),
        bufferSize: bufferSize,
        ctx:        ctx,
        cancel:     cancel,
    }
}

func (dp *DataProcessor) Start(workers int) {
    // Start processing workers
    for i := 0; i < workers; i++ {
        dp.wg.Add(1)
        go dp.worker(i)
    }
    
    // Start metrics reporter
    dp.wg.Add(1)
    go dp.metricsReporter()
}

func (dp *DataProcessor) worker(id int) {
    defer dp.wg.Done()
    
    for {
        select {
        case data := <-dp.input:
            processed, err := dp.processData(data)
            if err != nil {
                atomic.AddInt64(&dp.metrics.Errors, 1)
                continue
            }
            
            // Try to send processed data with backpressure handling
            select {
            case dp.output <- processed:
                atomic.AddInt64(&dp.metrics.Processed, 1)
            case <-time.After(time.Millisecond * 100):
                // Drop data if output buffer is full
                atomic.AddInt64(&dp.metrics.Dropped, 1)
            case <-dp.ctx.Done():
                return
            }
            
        case <-dp.ctx.Done():
            return
        }
    }
}

func (dp *DataProcessor) processData(data DataPoint) (DataPoint, error) {
    // Simulate processing time
    time.Sleep(time.Millisecond * time.Duration(rand.Intn(50)))
    
    // Transform data based on type
    switch data.Type {
    case "number":
        if val, ok := data.Value.(float64); ok {
            data.Value = val * 2 // Double numbers
        }
    case "string":
        if val, ok := data.Value.(string); ok {
            data.Value = fmt.Sprintf("processed_%s", val)
        }
    }
    
    data.Timestamp = time.Now()
    return data, nil
}

func (dp *DataProcessor) metricsReporter() {
    defer dp.wg.Done()
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            processed := atomic.LoadInt64(&dp.metrics.Processed)
            errors := atomic.LoadInt64(&dp.metrics.Errors)
            dropped := atomic.LoadInt64(&dp.metrics.Dropped)
            
            fmt.Printf("Metrics - Processed: %d, Errors: %d, Dropped: %d\n",
                processed, errors, dropped)
                
        case <-dp.ctx.Done():
            return
        }
    }
}

func (dp *DataProcessor) Submit(data DataPoint) bool {
    select {
    case dp.input <- data:
        return true
    case <-time.After(time.Millisecond * 10):
        return false // Input buffer full
    }
}

func (dp *DataProcessor) Results() <-chan DataPoint {
    return dp.output
}

func (dp *DataProcessor) Stop() {
    dp.cancel()
    close(dp.input)
    dp.wg.Wait()
    close(dp.output)
}

func dataProcessorExample() {
    processor := NewDataProcessor(100)
    processor.Start(3) // 3 workers
    
    // Start data generator
    go func() {
        for i := 0; i < 1000; i++ {
            data := DataPoint{
                ID:        fmt.Sprintf("data_%d", i),
                Timestamp: time.Now(),
                Type:      []string{"number", "string"}[rand.Intn(2)],
            }
            
            if data.Type == "number" {
                data.Value = rand.Float64() * 100
            } else {
                data.Value = fmt.Sprintf("item_%d", i)
            }
            
            if !processor.Submit(data) {
                fmt.Printf("Failed to submit data %d\n", i)
            }
            
            time.Sleep(time.Millisecond * 10)
        }
    }()
    
    // Consume results
    go func() {
        for result := range processor.Results() {
            // Process results (could write to database, send to another service, etc.)
            _ = result
        }
    }()
    
    // Run for 30 seconds
    time.Sleep(30 * time.Second)
    processor.Stop()
}

📚 Summary

In this module, you’ve learned advanced concurrency patterns that are essential for building scalable Go applications:

Key Patterns Covered

Worker Pool - Efficiently manage goroutines and process work
Fan-In/Fan-Out - Distribute and aggregate work across multiple goroutines
Context - Handle cancellation, timeouts, and request-scoped values
Pipeline - Chain processing stages for data transformation
Rate Limiting - Control resource usage and API call rates
Advanced Sync - Use sync.Once, condition variables for complex coordination
Error Handling - Aggregate errors and implement circuit breakers
Performance - Memory pools and lock-free data structures

Best Practices

Always use context for cancellation and timeouts
Implement proper backpressure handling
Monitor and collect metrics from concurrent systems
Use appropriate synchronization primitives
Design for graceful shutdown
Handle errors appropriately in concurrent contexts

Next Steps

Practice implementing these patterns in real applications
Study production Go codebases to see these patterns in action
Experiment with performance optimizations
Learn about distributed systems patterns
Explore Go’s runtime and scheduler in detail

🔗 Additional Resources

Date Created: August 3, 2025
Estimated Completion Time: 8-10 hours over 3-4 days
Difficulty Level: Advanced