并发

Goroutine 最佳实践

1. 合理控制 Goroutine 数量：

package main

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

// 使用 worker pool 模式限制 goroutine 数量
type WorkerPool struct {
    tasks   chan int
    workers int
    wg      sync.WaitGroup
}

func NewWorkerPool(workers int) *WorkerPool {
    return &WorkerPool{
        tasks:   make(chan int, 100),
        workers: workers,
    }
}

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 task := range wp.tasks {
        fmt.Printf("Worker %d 处理任务 %d\n", id, task)
        time.Sleep(100 * time.Millisecond)
    }
}

func (wp *WorkerPool) Submit(task int) {
    wp.tasks <- task
}

func (wp *WorkerPool) Stop() {
    close(wp.tasks)
    wp.wg.Wait()
}

func main() {
    pool := NewWorkerPool(5) // 限制为 5 个 worker
    pool.Start()
    
    // 提交 20 个任务
    for i := 1; i <= 20; i++ {
        pool.Submit(i)
    }
    
    pool.Stop()
    fmt.Println("所有任务完成")
}

2. 避免 Goroutine 泄露：

package main

import (
    "fmt"
    "time"
)

// 错误示例：goroutine 泄露
func leak() {
    ch := make(chan int)
    go func() {
        val := <-ch // 永远阻塞，goroutine 泄露
        fmt.Println(val)
    }()
}

// 正确示例：使用 context 控制
import "context"

func noLeak(ctx context.Context) {
    ch := make(chan int)
    go func() {
        select {
        case val := <-ch:
            fmt.Println(val)
        case <-ctx.Done():
            fmt.Println("goroutine 被取消")
            return
        }
    }()
}

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

3. 使用 WaitGroup 等待 Goroutine：

package main

import (
    "fmt"
    "sync"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Worker %d 完成\n", id)
}

func main() {
    var wg sync.WaitGroup
    
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
    
    wg.Wait()
    fmt.Println("所有 worker 完成")
}

4. 使用 ErrGroup 处理错误：

package main

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

func task(name string, shouldFail bool) error {
    fmt.Printf("Task %s 开始\n", name)
    if shouldFail {
        return errors.New("task failed")
    }
    fmt.Printf("Task %s 完成\n", name)
    return nil
}

func main() {
    var g errgroup.Group
    
    g.Go(func() error {
        return task("A", false)
    })
    
    g.Go(func() error {
        return task("B", true) // 这个会失败
    })
    
    g.Go(func() error {
        return task("C", false)
    })
    
    if err := g.Wait(); err != nil {
        fmt.Println("错误:", err)
    }
}

5. 使用 Channel 进行通信：

package main

import (
    "fmt"
    "time"
)

func producer(ch chan<- int) {
    for i := 1; i <= 5; i++ {
        ch <- i
        fmt.Printf("生产: %d\n", i)
        time.Sleep(100 * time.Millisecond)
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for val := range ch {
        fmt.Printf("消费: %d\n", val)
        time.Sleep(150 * time.Millisecond)
    }
}

func main() {
    ch := make(chan int, 3)
    
    go producer(ch)
    go consumer(ch)
    
    time.Sleep(2 * time.Second)
}

6. 使用 Context 取消 Goroutine：

package main

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

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d 被取消\n", id)
            return
        default:
            fmt.Printf("Worker %d 工作中...\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() // 取消所有 goroutine
    
    time.Sleep(1 * time.Second)
}

7. 使用 Mutex 保护共享资源：

package main

import (
    "fmt"
    "sync"
)

type Counter struct {
    mu    sync.Mutex
    value int
}

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

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

func main() {
    var counter Counter
    var wg sync.WaitGroup
    
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }
    
    wg.Wait()
    fmt.Printf("最终值: %d\n", counter.Value())
}

8. 性能优化技巧：

• 避免过度创建 goroutine：使用 worker pool 模式
• 合理设置 GOMAXPROCS：根据 CPU 核心数调整
• 使用缓冲 channel：减少阻塞
• 避免锁竞争：尽量使用 channel 通信
• 使用 sync.Pool：重用对象，减少 GC 压力
• 监控 goroutine 数量：使用 runtime.NumGoroutine()

package main

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

func monitorGoroutines() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        fmt.Printf("当前 goroutine 数量: %d\n", runtime.NumGoroutine())
    }
}

func main() {
    go monitorGoroutines()
    
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            time.Sleep(2 * time.Second)
        }()
    }
    
    wg.Wait()
    fmt.Println("所有 goroutine 完成")
}