Go Concurrency Patterns: Goroutines, Channels, and Best Practices

// Simple goroutine
go func() {
    fmt.Println("Running in goroutine")
}()

// With function
go processData(data)

import "sync"

var wg sync.WaitGroup

func main() {
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            processItem(id)
        }(i)
    }
    wg.Wait() // Wait for all goroutines
}

// Unbuffered channel
ch := make(chan int)

// Send
ch <- 42

// Receive
value := <-ch

// Close
close(ch)

// Buffered channel (capacity 10)
ch := make(chan int, 10)

// Non-blocking send if buffer not full
ch <- 1
ch <- 2

// Send-only channel
func sendOnly(ch chan<- int) {
    ch <- 42
}

// Receive-only channel
func receiveOnly(ch <-chan int) {
    value := <-ch
}

// Bidirectional (default)
func bidirectional(ch chan int) {
    ch <- 42
    value := <-ch
}

func workerPool(jobs <-chan int, results chan<- int, numWorkers int) {
    var wg sync.WaitGroup
    
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                result := processJob(job)
                results <- result
            }
        }()
    }
    
    wg.Wait()
    close(results)
}

// Fan-out: Distribute work
func fanOut(input <-chan int, outputs []chan int) {
    for value := range input {
        for _, output := range outputs {
            output <- value
        }
    }
    for _, output := range outputs {
        close(output)
    }
}

// Fan-in: Combine results
func fanIn(inputs []<-chan int) <-chan int {
    output := make(chan int)
    var wg sync.WaitGroup
    
    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for value := range ch {
                output <- value
            }
        }(input)
    }
    
    go func() {
        wg.Wait()
        close(output)
    }()
    
    return output
}

func pipeline(input <-chan int) <-chan int {
    stage1 := make(chan int)
    stage2 := make(chan int)
    
    // Stage 1
    go func() {
        defer close(stage1)
        for value := range input {
            stage1 <- value * 2
        }
    }()
    
    // Stage 2
    go func() {
        defer close(stage2)
        for value := range stage1 {
            stage2 <- value + 1
        }
    }()
    
    return stage2
}

import "context"

func processWithContext(ctx context.Context, data []int) error {
    for _, item := range data {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            processItem(item)
        }
    }
    return nil
}

// Usage
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

err := processWithContext(ctx, data)

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

result, err := fetchData(ctx)

select {
case value := <-ch1:
    // Handle value from ch1
case value := <-ch2:
    // Handle value from ch2
case ch3 <- 42:
    // Successfully sent to ch3
default:
    // No channel ready
}

select {
case result := <-ch:
    return result
case <-time.After(5 * time.Second):
    return errors.New("timeout")
}

type SafeCounter struct {
    mu    sync.Mutex
    value int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *SafeCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

type SafeMap struct {
    mu   sync.RWMutex
    data map[string]int
}

func (m *SafeMap) Get(key string) int {
    m.mu.RLock()
    defer m.mu.RUnlock()
    return m.data[key]
}

func (m *SafeMap) Set(key string, value int) {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.data[key] = value
}

// Good
func process() {
    mu.Lock()
    defer mu.Unlock()
    // Process
}

// Bad
func process() {
    mu.Lock()
    // Process
    mu.Unlock() // Might not execute if panic
}

// Good: Use context for cancellation
func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
            // Work
        }
    }
}

// Bad: Goroutine runs forever
func worker() {
    for {
        // Work forever
    }
}

// Good: Close channel when done
func producer(ch chan int) {
    defer close(ch)
    for i := 0; i < 10; i++ {
        ch <- i
    }
}

// Bad: Race condition
var counter int

func increment() {
    counter++ // Not thread-safe
}

// Good: Use mutex
var (
    counter int
    mu      sync.Mutex
)

func increment() {
    mu.Lock()
    defer mu.Unlock()
    counter++
}

// Bad: Panic if channel already closed
close(ch)
close(ch) // Panic!

// Good: Check or use sync.Once
var once sync.Once
once.Do(func() { close(ch) })

// For known capacity
ch := make(chan int, 100)

// Use worker pool instead of unlimited goroutines
const maxWorkers = 10
semaphore := make(chan struct{}, maxWorkers)

import _ "net/http/pprof"

// Add to main
go func() {
    log.Println(http.ListenAndServe("localhost:6060", nil))
}()