Go協程洩漏排查：2026年結構化並發5種模式徹底告別goroutine洩漏

编程语言

你是不是也遇過這種詭異場景？

線上服務運行幾小時後，記憶體佔用持續攀升，CPU飆到90%，pprof一看——幾千個goroutine堆積在channel接收處，永遠等不到資料。重啟後恢復正常，但過幾小時又復現。這就是典型的goroutine洩漏，Go並發程式設計中最隱蔽也最致命的bug。

更可怕的是，goroutine洩漏不會直接報錯，它像慢性毒藥一樣慢慢吞噬系統資源，直到OOM Killer出手。2026年了，我們不該再用「裸啟goroutine」的方式寫並發程式碼——結構化並發（Structured Concurrency） 才是正解。

什麼是結構化並發？

結構化並發源自2018年Martin Odersky的提議，核心思想是：並發子任務的生命週期必須被父任務嚴格管控。就像結構化程式設計消滅了goto帶來的混亂控制流，結構化並發消滅了「fire-and-forget」帶來的goroutine洩漏。

特性 非結構化並發 結構化並發
生命週期 不可控，可能洩漏 父任務退出時子任務必退出
錯誤傳播 需手動處理 自動向上傳播
資源回收 無保證 保證回收
除錯難度 極高 可追蹤
代表模式 go func(){} errgroup、Context取消

Go語言雖然沒有語言級結構化並發支援，但透過 context、errgroup、sync 等標準庫，我們可以建構完整的結構化並發體系。

goroutine洩漏的5大根因

1. 向未緩衝channel發送資料

func leak1() {
    ch := make(chan int)
    go func() {
        ch <- 42 // 永遠阻塞，無人接收
    }()
    // 函式返回，goroutine洩漏
}

2. 從未關閉channel接收資料

func leak2() {
    ch := make(chan int, 10)
    go func() {
        for v := range ch { // ch永不關閉，永遠阻塞
            fmt.Println(v)
        }
    }()
}

3. Context未取消

func leak3(ctx context.Context) {
    ctx2 := context.Background() // 沒繼承原ctx的取消信號
    go func() {
        select {
        case <-ctx2.Done(): // 永遠不會觸發
        case <-time.After(10 * time.Hour):
        }
    }()
}

4. WaitGroup計數不匹配

func leak4() {
    var wg sync.WaitGroup
    wg.Add(5)
    for i := 0; i < 3; i++ { // 只啟動3個，但Add了5
        go func() { wg.Done() }()
    }
    wg.Wait() // 永遠阻塞
}

5. HTTP連線未關閉回應體

func leak5() {
    resp, _ := http.Get("https://example.com")
    // defer resp.Body.Close() // 忘記關閉！
    // 導致底層transport的goroutine洩漏
}

模式一：errgroup——帶錯誤傳播的並發

errgroup 是結構化並發最基礎的建構區塊，它保證：所有子goroutine完成（或任一出錯時取消其餘）後，父任務才繼續。

分步實操

Step 1: 安裝errgroup

go get golang.org/x/sync/errgroup

Step 2: 基礎用法——並發請求多個API

package main

import (
    "context"
    "fmt"
    "net/http"
    "time"

    "golang.org/x/sync/errgroup"
)

func fetchAPI(ctx context.Context, url string) ([]byte, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return nil, err
    }
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    return nil, nil
}

func fetchAll(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)
    results := make([][]byte, len(urls))

    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            data, err := fetchAPI(ctx, url)
            if err != nil {
                return err
            }
            results[i] = data
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return fmt.Errorf("fetch failed: %w", err)
    }
    return nil
}

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

    urls := []string{
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    }

    if err := fetchAll(ctx, urls); err != nil {
        fmt.Println("Error:", err)
    }
}

模式二：信號量——限制並發度

當需要控制並發度（如限制資料庫連線數），用 semaphore 套件。

完整程式碼

package main

import (
    "context"
    "fmt"
    "sync/atomic"
    "time"

    "golang.org/x/sync/errgroup"
    "golang.org/x/sync/semaphore"
)

func processWithLimit(ctx context.Context, tasks []string, maxConcurrency int64) error {
    sem := semaphore.NewWeighted(maxConcurrency)
    g, ctx := errgroup.WithContext(ctx)
    var activeCount int64

    for _, task := range tasks {
        task := task
        if err := sem.Acquire(ctx, 1); err != nil {
            return fmt.Errorf("acquire semaphore: %w", err)
        }

        g.Go(func() error {
            defer sem.Release(1)
            current := atomic.AddInt64(&activeCount, 1)
            defer atomic.AddInt64(&activeCount, -1)
            fmt.Printf("Processing %s (active: %d/%d)\n", task, current, maxConcurrency)

            select {
            case <-time.After(500 * time.Millisecond):
                return nil
            case <-ctx.Done():
                return ctx.Err()
            }
        })
    }

    return g.Wait()
}

func main() {
    ctx := context.Background()
    tasks := make([]string, 20)
    for i := range tasks {
        tasks[i] = fmt.Sprintf("task-%d", i)
    }

    if err := processWithLimit(ctx, tasks, 5); err != nil {
        fmt.Println("Error:", err)
    }
}

模式三：Pipeline——串流處理

Pipeline模式將資料流經多個階段，每個階段是一組goroutine，階段間用channel連線。

完整程式碼

package main

import (
    "context"
    "fmt"
    "sync"

    "golang.org/x/sync/errgroup"
)

func generate(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case out <- n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func filter(ctx context.Context, in <-chan int, predicate func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if predicate(n) {
                select {
                case out <- n:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

func runPipeline(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)
    var mu sync.Mutex
    results := []int{}

    ch := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    ch = square(ctx, ch)
    ch = filter(ctx, ch, func(n int) bool { return n > 20 })

    g.Go(func() error {
        for n := range ch {
            mu.Lock()
            results = append(results, n)
            mu.Unlock()
        }
        return nil
    })

    if err := g.Wait(); err != nil {
        return err
    }
    fmt.Println("Results:", results)
    return nil
}

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

模式四：Fan-out/Fan-in——分發聚合

將任務分發給多個worker（fan-out），再聚合結果（fan-in）。

完整程式碼

package main

import (
    "context"
    "fmt"
    "sync"

    "golang.org/x/sync/errgroup"
)

func fanOut(ctx context.Context, in <-chan int, workerCount int) []<-chan int {
    channels := make([]<-chan int, workerCount)
    for i := 0; i < workerCount; i++ {
        channels[i] = processWorker(ctx, in, i)
    }
    return channels
}

func processWorker(ctx context.Context, in <-chan int, id int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            result := n * n
            select {
            case out <- result:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    for _, ch := range channels {
        ch := ch
        go func() {
            defer wg.Done()
            for n := range ch {
                select {
                case out <- n:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }

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

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

    in := make(chan int, 10)
    for i := 1; i <= 10; i++ {
        in <- i
    }
    close(in)

    workerChs := fanOut(ctx, in, 3)
    merged := fanIn(ctx, workerChs...)

    for result := range merged {
        fmt.Println(result)
    }
}

模式五：Worker Pool——可控工作池

最通用的結構化並發模式，嚴格控制goroutine數量和生命週期。

完整程式碼

package main

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

type Task struct {
    ID    int
    Input string
}

type Result struct {
    TaskID int
    Output string
    Err    error
}

type WorkerPool struct {
    workerCount int
    tasks       chan Task
    results     chan Result
    wg          sync.WaitGroup
    cancel      context.CancelFunc
}

func NewWorkerPool(ctx context.Context, workerCount, taskBufSize int) *WorkerPool {
    ctx, cancel := context.WithCancel(ctx)
    pool := &WorkerPool{
        workerCount: workerCount,
        tasks:       make(chan Task, taskBufSize),
        results:     make(chan Result, taskBufSize),
        cancel:      cancel,
    }
    pool.start(ctx)
    return pool
}

func (p *WorkerPool) start(ctx context.Context) {
    for i := 0; i < p.workerCount; i++ {
        p.wg.Add(1)
        go func(workerID int) {
            defer p.wg.Done()
            for {
                select {
                case task, ok := <-p.tasks:
                    if !ok {
                        return
                    }
                    result := p.process(ctx, workerID, task)
                    select {
                    case p.results <- result:
                    case <-ctx.Done():
                        return
                    }
                case <-ctx.Done():
                    return
                }
            }
        }(i)
    }
}

func (p *WorkerPool) process(ctx context.Context, workerID int, task Task) Result {
    select {
    case <-time.After(100 * time.Millisecond):
        return Result{
            TaskID: task.ID,
            Output: fmt.Sprintf("worker-%d processed: %s", workerID, task.Input),
        }
    case <-ctx.Done():
        return Result{TaskID: task.ID, Err: ctx.Err()}
    }
}

func (p *WorkerPool) Submit(task Task) { p.tasks <- task }
func (p *WorkerPool) Results() <-chan Result { return p.results }

func (p *WorkerPool) Shutdown() {
    close(p.tasks)
    p.wg.Wait()
    close(p.results)
}

func (p *WorkerPool) Cancel() { p.cancel() }

func main() {
    ctx := context.Background()
    pool := NewWorkerPool(ctx, 4, 100)

    go func() {
        for i := 0; i < 20; i++ {
            pool.Submit(Task{ID: i, Input: fmt.Sprintf("data-%d", i)})
        }
        pool.Shutdown()
    }()

    for result := range pool.Results() {
        fmt.Printf("Task %d: %s (err: %v)\n", result.TaskID, result.Output, result.Err)
    }
}

避坑指南

坑1：errgroup中吞掉錯誤

// ❌ 錯誤：g.Go裡返回nil，錯誤被吞
g.Go(func() error {
    if err := doWork(); err != nil {
        log.Println(err)
        return nil
    }
    return nil
})

// ✅ 正確：必須返回錯誤
g.Go(func() error {
    return doWork()
})

坑2：Context不傳播

// ❌ 錯誤：用了context.Background()，取消信號丟失
g.Go(func() error {
    return fetch(context.Background(), url)
})

// ✅ 正確：使用errgroup的ctx
g.Go(func() error {
    return fetch(ctx, url)
})

坑3：迴圈變數捕獲錯誤

// ❌ 錯誤：Go 1.22前，i是迴圈變數引用
for i := 0; i < 10; i++ {
    g.Go(func() error {
        return process(i)
    })
}

// ✅ 正確：Go 1.22+自動修復，或手動綁定
for i := 0; i < 10; i++ {
    i := i
    g.Go(func() error {
        return process(i)
    })
}

坑4：channel未關閉導致range死鎖

// ❌ 錯誤：生產者不關閉channel
func produce() <-chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            ch <- i
        }
        // 忘記close(ch)
    }()
    return ch
}

// ✅ 正確：必須defer close
func produce() <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        for i := 0; i < 10; i++ {
            ch <- i
        }
    }()
    return ch
}

坑5：信號量在errgroup外使用導致洩漏

// ❌ 錯誤：Acquire在g.Go外呼叫
sem.Acquire(ctx, 1)
g.Go(func() error {
    defer sem.Release(1)
    return work()
})

// ✅ 正確：在g.Go內Acquire
g.Go(func() error {
    if err := sem.Acquire(ctx, 1); err != nil {
        return err
    }
    defer sem.Release(1)
    return work()
})

報錯排查

序號 報錯訊息 原因 解決方法
1 fatal error: all goroutines are asleep - deadlock! 所有goroutine阻塞在channel操作 檢查channel是否有生產者/消費者，確保channel會被關閉
2 context deadline exceeded 操作超時，context超時取消 增加超時時間或最佳化操作效能，檢查是否有慢查詢
3 errgroup: Go called after Wait Wait之後又呼叫Go 確保所有Go呼叫在Wait之前完成
4 panic: sync: WaitGroup is reused before previous Wait has returned WaitGroup重用不當 每次使用建立新WaitGroup，或確保上一輪Wait完成
5 panic: close of closed channel 重複關閉channel 用sync.Once保護close，或只由一個goroutine負責關閉
6 panic: send on closed channel channel關閉後仍發送 確保發送方知道channel何時關閉，用select檢測
7 goroutine profile: total XXX (數量持續增長) goroutine洩漏 用pprof排查，檢查未退出的goroutine棧
8 runtime: out of memory goroutine過多導致記憶體耗盡 限制並發度，使用worker pool或semaphore
9 signal: killed (OOM Killer) 系統記憶體不足殺行程 減少goroutine數量，增加記憶體限制或最佳化記憶體使用
10 import cycle not allowed 並發程式碼套件循環依賴 重新組織套件結構，將共享型別提取到獨立套件

進階最佳化

1. 使用runtime/pprof即時監控goroutine數量

go func() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for range ticker.C {
        fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
    }
}()

2. 為每個goroutine新增超時保護

func safeGoroutine(ctx context.Context, fn func() error) error {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()
    done := make(chan error, 1)
    go func() { done <- fn() }()
    select {
    case err := <-done:
        return err
    case <-ctx.Done():
        return fmt.Errorf("goroutine timed out: %w", ctx.Err())
    }
}

3. 使用go-leak測試框架檢測洩漏

import "go.uber.org/goleak"

func TestMain(m *testing.M) {
    goleak.VerifyTestMain(m)
}

4. 結構化日誌關聯goroutine

func tracedGoroutine(ctx context.Context, name string, fn func(context.Context) error) error {
    logger := slog.With("goroutine", name, "traceID", ctx.Value("traceID"))
    logger.Info("goroutine started")
    err := fn(ctx)
    logger.Info("goroutine finished", "error", err)
    return err
}

對比分析

維度 errgroup Semaphore Pipeline Fan-out/Fan-in Worker Pool
並發控制 無限制 限制並發度 階段串行 無限制 固定worker數
錯誤處理 自動傳播 需手動 需手動 需手動 需手動
適用場景 並發請求 限流/資源控制 串流處理 計算密集型 通用任務處理
實現複雜度 低 中 中 高 高
goroutine安全 ✅ ✅ ⚠️需注意 ⚠️需注意 ✅
背壓支援 ❌ ✅ ✅ ❌ ✅
取消傳播 ✅自動 ✅自動 ⚠️需手動 ⚠️需手動 ✅

總結：goroutine洩漏的本質是「失控的並發」——子任務生命週期脫離父任務管控。結構化並發透過errgroup（錯誤傳播）、semaphore（並發限制）、pipeline（串流處理）、fan-out/fan-in（分發聚合）、worker pool（可控池化）五種模式，從不同維度確保並發安全。2026年，請告別裸 go func(){}，擁抱結構化並發。

線上工具推薦

本站提供瀏覽器本地工具，免註冊即可試用 →

#Go#协程#结构化并发#goroutine#context#并发模式#errgroup#并发控制