Go Channels và Communication: Hướng Dẫn Hoàn Chỉnh

Trong bài trước về goroutines, chúng ta đã học cách tạo lightweight concurrent tasks. Giờ chúng ta sẽ làm chủ channels - cách thanh lịch của Go để communicate giữa các goroutines.

"Don't communicate by sharing memory; share memory by communicating." - Go proverb

Channels Là Gì?

Channel là một typed conduit mà qua đó bạn có thể send và receive values giữa các goroutines. Channels cho phép safe communication mà không cần explicit locks hoặc condition variables.

Đặc điểm chính:

Type-safe: Channels carry values của một type cụ thể
Thread-safe: Multiple goroutines có thể safely send/receive
Blocking: Send/receive operations block until ready
First-class: Channels là values có thể passed around

Channel Đầu Tiên

package main
 
import "fmt"
 
func main() {
    // Tạo channel of integers
    ch := make(chan int)
 
    // Send value trong goroutine
    go func() {
        ch <- 42 // Send 42 to channel
    }()
 
    // Receive value
    value := <-ch // Receive from channel
    fmt.Println("Received:", value)
}

Channel Basics

Tạo Channels

// Make a channel
ch := make(chan int)        // Unbuffered channel
ch := make(chan string, 10) // Buffered channel (capacity 10)
 
// Channel types
var readOnly <-chan int  // Chỉ có thể receive
var writeOnly chan<- int // Chỉ có thể send
var readWrite chan int   // Có thể send và receive

Sending và Receiving

ch <- value   // Send value to channel (blocks until received)
value := <-ch // Receive value from channel (blocks until sent)
value, ok := <-ch // Receive với status (ok là false nếu channel closed)

Closing Channels

close(ch) // Close channel (không còn sends allowed)
 
// Check nếu closed
value, ok := <-ch
if !ok {
    fmt.Println("Channel closed")
}

Unbuffered vs Buffered Channels

Unbuffered Channels (Synchronous)

ch := make(chan int) // Không có buffer
 
go func() {
    ch <- 42 // Blocks until someone receives
}()
 
value := <-ch // Blocks until someone sends
fmt.Println(value)

Đặc điểm:

Send blocks until receive happens
Receive blocks until send happens
Synchronization point giữa goroutines

Dùng khi: Bạn cần guaranteed synchronization

Buffered Channels (Asynchronous)

ch := make(chan int, 3) // Buffer size 3
 
// Send không block until buffer full
ch <- 1
ch <- 2
ch <- 3
// ch <- 4 // Sẽ block here
 
// Receive không block until buffer empty
fmt.Println(<-ch) // 1
fmt.Println(<-ch) // 2
fmt.Println(<-ch) // 3

Đặc điểm:

Send blocks chỉ khi buffer full
Receive blocks chỉ khi buffer empty
Decouples sender và receiver

Dùng khi: Bạn muốn allow some asynchrony

Select Statement

select cho phép bạn wait trên multiple channel operations:

package main
 
import (
    "fmt"
    "time"
)
 
func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)
 
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "from ch1"
    }()
 
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "from ch2"
    }()
 
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println("Received:", msg1)
        case msg2 := <-ch2:
            fmt.Println("Received:", msg2)
        }
    }
}

Select với Timeout

select {
case res := <-ch:
    fmt.Println("Got:", res)
case <-time.After(1 * time.Second):
    fmt.Println("Timeout!")
}

Channel Patterns

Pattern 1: Fan-Out (Distribute Work)

Multiple workers process từ single source:

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        time.Sleep(500 * time.Millisecond)
    }
}
 
func main() {
    jobs := make(chan int, 10)
    var wg sync.WaitGroup
 
    // Start 3 workers
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go worker(i, jobs, &wg)
    }
 
    // Send jobs
    for j := 1; j <= 9; j++ {
        jobs <- j
    }
    close(jobs)
 
    wg.Wait()
}

Pattern 2: Fan-In (Collect Results)

Merge multiple channels thành một:

func fanIn(channels ...<-chan string) <-chan string {
    out := make(chan string)
    var wg sync.WaitGroup
 
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan string) {
            defer wg.Done()
            for msg := range c {
                out <- msg
            }
        }(ch)
    }
 
    go func() {
        wg.Wait()
        close(out)
    }()
 
    return out
}

Pattern 3: Pipeline

Chain processing stages:

// Stage 1: Generate numbers
func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}
 
// Stage 2: Square numbers
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 main() {
    numbers := generator(1, 2, 3, 4, 5)
    squared := square(numbers)
 
    for n := range squared {
        fmt.Println(n)
    }
}

Pattern 4: Worker Pool

Fixed number of workers processing jobs:

type Job struct {
    ID   int
    Data string
}
 
type Result struct {
    Job    Job
    Output string
}
 
func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        results <- Result{
            Job:    job,
            Output: fmt.Sprintf("Processed %s", job.Data),
        }
    }
}
 
func main() {
    const numWorkers = 3
    jobs := make(chan Job, 10)
    results := make(chan Result, 10)
 
    // Start workers
    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }
 
    // Send jobs
    for j := 1; j <= 10; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("task-%d", j)}
    }
    close(jobs)
 
    // Collect results
    for r := 1; r <= 10; r++ {
        result := <-results
        fmt.Printf("Job %d result: %s\n", result.Job.ID, result.Output)
    }
}

Context cho Cancellation

Dùng context.Context để cancel goroutines gracefully:

func worker(ctx context.Context, id int, results chan<- int) {
    for i := 0; ; i++ {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d: Cancelled\n", id)
            return
        case results <- i:
            fmt.Printf("Worker %d: Sent %d\n", id, i)
            time.Sleep(500 * time.Millisecond)
        }
    }
}
 
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
 
    results := make(chan int)
    go worker(ctx, 1, results)
 
    for {
        select {
        case result := <-results:
            fmt.Println("Received:", result)
        case <-ctx.Done():
            return
        }
    }
}

Channels vs Mutexes: Khi Nào Dùng Mỗi Loại

Dùng Channels Khi:

✅ Passing ownership của data
✅ Distributing work to multiple workers
✅ Communicating results giữa goroutines
✅ Coordinating goroutine lifecycles

Dùng Mutexes Khi:

✅ Protecting shared state (caches, counters)
✅ Short critical sections
✅ Simple data access without coordination

Common Channel Pitfalls

Pitfall 1: Sending trên Closed Channel

// ❌ BAD: Panics!
ch := make(chan int)
close(ch)
ch <- 42 // panic: send on closed channel

Pitfall 2: Closing Channel Multiple Times

// ❌ BAD: Panics!
ch := make(chan int)
close(ch)
close(ch) // panic: close of closed channel

Pitfall 3: Deadlock (No Receiver)

// ❌ BAD: Deadlock!
ch := make(chan int)
ch <- 42 // Blocks forever (no receiver)

Pitfall 4: Goroutine Leak

// ❌ BAD: Goroutine leaks nếu channel không bao giờ receives
func leak() <-chan int {
    ch := make(chan int)
    go func() {
        ch <- computeExpensiveValue() // Blocks forever nếu no receiver
    }()
    return ch
}

Best Practices

1. Close Channels từ Sender

// ✅ GOOD
func producer(ch chan<- int) {
    defer close(ch) // Sender closes
    for i := 0; i < 10; i++ {
        ch <- i
    }
}

2. Dùng Directional Channels

// ✅ GOOD: Type safety
func send(ch chan<- int) { // Chỉ có thể send
    ch <- 42
}
 
func receive(ch <-chan int) { // Chỉ có thể receive
    value := <-ch
}

3. Range Over Channels

// ✅ GOOD: Automatically stops khi closed
for value := range ch {
    fmt.Println(value)
}

4. Check nếu Channel Closed

// ✅ GOOD
value, ok := <-ch
if !ok {
    fmt.Println("Channel closed")
    return
}

Tóm Tắt và Key Takeaways

✅ Channels enable safe communication giữa goroutines
✅ Unbuffered channels provide synchronization (blocking send/receive)
✅ Buffered channels allow asynchronous communication
✅ Select statement multiplexes multiple channel operations
✅ Close channels từ sender, không phải receiver
✅ Dùng range để iterate until channel closed
✅ Context enables graceful cancellation
✅ Channels cho coordination, mutexes cho shared state
✅ Common patterns: Fan-out, fan-in, pipeline, worker pool
✅ Avoid: Sending on closed channels, closing multiple times, deadlocks

Tiếp Theo Là Gì?

Bạn đã làm chủ Go's concurrency primitives! Bước tiếp theo:

Continue Go Series:

GO-10: Building Web Services in Go
GO-11: Testing in Go

Deep Dives:

Context Package (cancellation, deadlines, values)
Advanced Concurrency Patterns
Go Performance Optimization

Tài Nguyên Bổ Sung

Tài liệu chính thức:

Advanced Topics:

Sách:

"Concurrency in Go" by Katherine Cox-Buday

Có câu hỏi về channels hoặc concurrency? Để lại comment bên dưới!

Chúc concurrent programming vui vẻ! 🚀