4. Goroutine数量控制

在Go高并发编程中，goroutine的轻量级特性（初始栈仅2KB）容易让开发者产生“可以无限创建”的错觉。然而，无限制地创建goroutine会迅速耗尽系统资源，导致性能急剧下降甚至服务崩溃。本节将系统讲解goroutine数量控制的四种核心策略，帮助你在高并发场景下实现稳定、可控的并发模型。

4.1 无限制创建goroutine的危害

每个goroutine虽然轻量，但并非零成本。当goroutine数量失控时，会引发以下连锁问题：

4.2 使用WaitGroup：等待goroutine完成

sync.WaitGroup 是Go标准库提供的goroutine同步原语，用于等待一组goroutine全部执行完毕。它不直接限制goroutine数量，但能确保主goroutine在子任务完成前不会退出，是并发控制的基础组件。

4.2.1 基本用法

package main

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

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // 任务完成时计数器减1
    fmt.Printf("Worker %d 开始\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d 结束\n", id)
}

func main() {
    var wg sync.WaitGroup
    for i := 1; i <= 5; i++ {
        wg.Add(1) // 计数器加1
        go worker(i, &wg)
    }
    wg.Wait() // 阻塞直到计数器归零
    fmt.Println("所有worker完成")
}

4.2.2 WaitGroup的三大方法

4.3 使用Channel信号量：控制并发上限

通过带缓冲的channel作为信号量（Semaphore），可以精确控制同时运行的goroutine数量。这是最灵活、最常用的限流手段。

4.3.1 基于channel的信号量模式

package main

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

func main() {
    const maxConcurrency = 3 // 最大并发数
    sem := make(chan struct{}, maxConcurrency)
    var wg sync.WaitGroup
    tasks := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

    for _, task := range tasks {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            sem <- struct{}{} // 获取信号量，满时阻塞
            defer func() { <-sem }() // 释放信号量

            fmt.Printf("处理任务 %d (并发数: %d)\n", id, len(sem))
            time.Sleep(time.Second)
        }(task)
    }
    wg.Wait()
    fmt.Println("所有任务完成")
}

4.3.2 信号量模式的核心机制

4.3.3 信号量模式的变体

• 加权信号量： 使用 chan int 传递权重值，不同任务占用不同数量的信号量资源。
• 超时获取： 结合 select 和 time.After，在信号量获取超时时执行降级逻辑。
• 动态调整： 通过 reflect.Select 或通道关闭机制，运行时调整并发上限。

4.4 使用第三方库：errgroup与tunny

标准库提供了基础工具，而第三方库在错误处理、goroutine池管理等方面提供了更高级的抽象。

4.4.1 errgroup：带错误传播的并发控制

golang.org/x/sync/errgroup 在WaitGroup基础上增加了错误收集和取消传播能力，特别适合需要“任一失败即取消所有”的场景。

package main

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

func main() {
    g := new(errgroup.Group)
    urls := []string{
        "https://example.com",
        "https://invalid-url-xxx.com", // 会失败
        "https://httpbin.org/get",
    }

    for _, url := range urls {
        url := url // 捕获循环变量
        g.Go(func() error {
            resp, err := http.Get(url)
            if err != nil {
                return fmt.Errorf("请求 %s 失败: %w", url, err)
            }
            defer resp.Body.Close()
            fmt.Printf("请求 %s 成功，状态码: %d\n", url, resp.StatusCode)
            return nil
        })
    }

    // 等待所有goroutine完成，返回第一个非nil错误
    if err := g.Wait(); err != nil {
        fmt.Printf("发生错误: %v\n", err)
    } else {
        fmt.Println("所有请求成功")
    }
}

4.4.2 tunny：固定大小的goroutine池

github.com/Jeffail/tunny 提供了固定大小的goroutine池，适合需要严格控制并发数的批量任务处理场景。

package main

import (
    "fmt"
    "github.com/Jeffail/tunny"
    "time"
)

func main() {
    // 创建大小为4的goroutine池
    pool, _ := tunny.NewFunc(4, func(payload interface{}) interface{} {
        num := payload.(int)
        time.Sleep(time.Second)
        return num * 2
    })
    defer pool.Close()

    // 提交任务并获取结果
    results := make([]int, 10)
    for i := 0; i < 10; i++ {
        result, err := pool.ProcessTimed(i, time.Second*2)
        if err != nil {
            fmt.Printf("任务 %d 超时: %v\n", i, err)
            continue
        }
        results[i] = result.(int)
        fmt.Printf("任务 %d 结果: %d\n", i, results[i])
    }
}

4.5 四种控制策略对比总结