Overview

Master Go's concurrency model using goroutines, channels, and synchronization primitives for building concurrent applications.

Goroutines

Creating goroutines:

package main

import (
    "fmt"
    "time"
)

func sayHello() {
    fmt.Println("Hello from goroutine")
}

func main() {
    // Launch goroutine
    go sayHello()

    // Anonymous function goroutine
    go func() {
        fmt.Println("Hello from anonymous goroutine")
    }()

    // Give goroutines time to execute
    time.Sleep(time.Second)
}

Goroutines with parameters:

func printNumber(n int) {
    fmt.Println(n)
}

func main() {
    for i := 0; i < 10; i++ {
        go printNumber(i)
    }
    time.Sleep(time.Second)
}

Channels

Basic channel operations:

func main() {
    // Create unbuffered channel
    ch := make(chan int)

    // Send in goroutine (non-blocking)
    go func() {
        ch <- 42
    }()

    // Receive (blocks until value available)
    value := <-ch
    fmt.Println(value) // 42
}

Buffered channels:

func main() {
    // Buffered channel with capacity 2
    ch := make(chan string, 2)

    // Can send up to 2 values without blocking
    ch <- "first"
    ch <- "second"

    fmt.Println(<-ch) // first
    fmt.Println(<-ch) // second
}

Channel direction:

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

// Receive-only channel
func receive(ch <-chan int) int {
    return <-ch
}

func main() {
    ch := make(chan int)

    go send(ch)
    value := receive(ch)

    fmt.Println(value)
}

Closing channels:

func main() {
    ch := make(chan int, 3)

    ch <- 1
    ch <- 2
    ch <- 3
    close(ch) // Close channel

    // Receive until channel is closed
    for value := range ch {
        fmt.Println(value)
    }

    // Check if channel is closed
    value, ok := <-ch
    fmt.Printf("Value: %d, Open: %v\n", value, ok) // Value: 0, Open: false
}

Select Statement

Multiplexing channels:

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(time.Second)
        ch1 <- "from ch1"
    }()

    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "from ch2"
    }()

    // Wait for both
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println(msg1)
        case msg2 := <-ch2:
            fmt.Println(msg2)
        }
    }
}

Select with default:

func main() {
    ch := make(chan int, 1)

    select {
    case val := <-ch:
        fmt.Println(val)
    default:
        fmt.Println("No value ready") // Executed
    }
}

Select with timeout:

func main() {
    ch := make(chan string)

    go func() {
        time.Sleep(2 * time.Second)
        ch <- "result"
    }()

    select {
    case msg := <-ch:
        fmt.Println(msg)
    case <-time.After(time.Second):
        fmt.Println("Timeout") // Executed after 1 second
    }
}

Worker Pools

Implementing worker pool pattern:

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        time.Sleep(time.Second)
        results <- job * 2
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    // Start 3 workers
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // Send 5 jobs
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // Collect results
    for a := 1; a <= 5; a++ {
        <-results
    }
}

sync.WaitGroup

Waiting for goroutines to complete:

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

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // Decrement counter when done

    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1) // Increment counter
        go worker(i, &wg)
    }

    wg.Wait() // Wait for all to complete
    fmt.Println("All workers done")
}

sync.Mutex

Protecting shared state:

import (
    "fmt"
    "sync"
)

type Counter struct {
    mu    sync.Mutex
    value int
}

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

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

func main() {
    var wg sync.WaitGroup
    counter := Counter{}

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }

    wg.Wait()
    fmt.Println(counter.Value()) // 1000
}

sync.RWMutex

Read-write locks:

type Cache struct {
    mu    sync.RWMutex
    items map[string]string
}

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock() // Read lock
    defer c.mu.RUnlock()
    val, ok := c.items[key]
    return val, ok
}

func (c *Cache) Set(key, value string) {
    c.mu.Lock() // Write lock
    defer c.mu.Unlock()
    c.items[key] = value
}

func main() {
    cache := Cache{items: make(map[string]string)}

    // Multiple readers can access simultaneously
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            cache.Get("key")
        }()
    }

    wg.Wait()
}

sync.Once

Execute once initialization:

var (
    instance *Database
    once     sync.Once
)

type Database struct {
    conn string
}

func GetDatabase() *Database {
    once.Do(func() {
        fmt.Println("Initializing database")
        instance = &Database{conn: "connected"}
    })
    return instance
}

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            db := GetDatabase() // Only initializes once
            fmt.Println(db.conn)
        }()
    }

    wg.Wait()
}

Context Package

Using context for cancellation:

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

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d cancelled\n", id)
            return
        default:
            fmt.Printf("Worker %d working\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }

    time.Sleep(2 * time.Second)
    cancel() // Cancel all workers
    time.Sleep(time.Second)
}

Context with timeout:

func slowOperation(ctx context.Context) error {
    select {
    case <-time.After(3 * time.Second):
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func main() {
    ctx, cancel := context.WithTimeout(
        context.Background(),
        2*time.Second,
    )
    defer cancel()

    err := slowOperation(ctx)
    if err != nil {
        fmt.Println("Operation timed out:", err)
    }
}

Context with values:

func processRequest(ctx context.Context) {
    userID := ctx.Value("userID")
    fmt.Println("Processing for user:", userID)
}

func main() {
    ctx := context.WithValue(
        context.Background(),
        "userID",
        "user123",
    )
    processRequest(ctx)
}

Error Handling in Concurrent Code

Using errgroup:

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
    "time"
)

func fetchUser(ctx context.Context, id int) error {
    time.Sleep(time.Second)
    if id == 3 {
        return fmt.Errorf("user %d not found", id)
    }
    fmt.Printf("Fetched user %d\n", id)
    return nil
}

func main() {
    g, ctx := errgroup.WithContext(context.Background())

    userIDs := []int{1, 2, 3, 4, 5}

    for _, id := range userIDs {
        id := id // Capture loop variable
        g.Go(func() error {
            return fetchUser(ctx, id)
        })
    }

    // Wait for all goroutines
    if err := g.Wait(); err != nil {
        fmt.Println("Error:", err)
    }
}

Fan-Out Fan-In Pattern

Distributing work and collecting results:

func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    wg.Add(len(cs))
    for _, c := range cs {
        go func(ch <-chan int) {
            defer wg.Done()
            for n := range ch {
                out <- n
            }
        }(c)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    in := generator(1, 2, 3, 4, 5)

    // Fan out
    c1 := square(in)
    c2 := square(in)

    // Fan in
    for n := range merge(c1, c2) {
        fmt.Println(n)
    }
}

When to Use This Skill

Use go-concurrency when you need to:

Execute multiple operations concurrently
Build concurrent servers or workers
Implement producer-consumer patterns
Process data streams concurrently
Handle multiple I/O operations simultaneously
Implement timeout and cancellation
Coordinate multiple goroutines
Build fan-out/fan-in pipelines
Share state safely between goroutines
Implement rate limiting or throttling

Best Practices

Use channels for communication, mutexes for state
Close channels from sender side only
Always use WaitGroup to wait for goroutines
Pass contexts for cancellation and deadlines
Use buffered channels judiciously
Protect shared state with mutexes
Avoid goroutine leaks with proper cleanup
Use select with default for non-blocking ops
Prefer sync.Once for initialization
Document goroutine ownership and lifecycle

Common Pitfalls

Goroutine leaks (forgetting to exit)
Race conditions from unprotected shared state
Deadlocks from improper channel usage
Sending on closed channels (panics)
Not checking channel close status
Overusing mutexes instead of channels
Creating too many goroutines
Forgetting to call WaitGroup.Done()
Passing loop variables to goroutines
Not handling context cancellation

📖 go-concurrency