English 中文 العربية 日本語 Português

Go: Go Goroutine 与并发基础

Goroutine 是 Go 并发编程的基石——它不是线程，不是协程，而是一种由 Go runtime 管理的轻量级并发单元，2KB 的栈空间让百万级并发成为可能。

Go 的并发模型让其他语言望尘莫及：启动一个 goroutine 只需要 go 关键字，而 Go runtime 在背后把成千上万个 goroutine 高效地映射到少量 OS 线程上。

1. 你将学到

2. 一个数据工程师的真实故事

(1) 痛点：10 万条日志，单线程跑 5 小时

Charlie 是数据工程师，他每天要处理 10 万条服务器日志：

"每天凌晨跑日志清洗任务，一个接一个地处理 10 万行，要跑 5 个多小时。早会的时候 PM 总问'昨天的数据什么时候出'，我说'下午 3 点'，ta 说'为什么不能早上 9 点到？'"

他打开现在的代码：

GO 
// 串行处理：10 万条 × 200ms/条 = 5.5 小时
func processLogs(logs []LogEntry) []Result {
    results := make([]Result, 0, len(logs))
    for _, log := range logs {
        result := processSingleLog(log)  // 每次 200ms，含网络 IO
        results = append(results, result)
    }
    return results
}

每条日志处理涉及一次外部 API 调用（约 200ms 等待），但 CPU 实际计算不到 1ms——99.5% 的时间在等待网络。

(2) Go 的解法：goroutine 并发

GO 
// log_processor.go
package main

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

type LogEntry struct {
    ID      int
    Message string
    Level   string
}

type Result struct {
    ID     int
    Status string
}

func processSingleLog(log LogEntry) Result {
    time.Sleep(200 * time.Millisecond)
    return Result{ID: log.ID, Status: "processed"}
}

func processLogsConcurrent(logs []LogEntry, workerCount int) []Result {
    results := make([]Result, 0, len(logs))
    var mu sync.Mutex
    var wg sync.WaitGroup

    // 用 channel 做任务分发
    jobs := make(chan LogEntry, len(logs))
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    // 启动 workerCount 个 worker
    for w := 0; w < workerCount; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                result := processSingleLog(log)
                mu.Lock()
                results = append(results, result)
                mu.Unlock()
            }
        }(w)
    }

    wg.Wait()
    return results
}

func main() {
    // 模拟 100 条日志（演示用，实际可扩展到 10 万）
    logs := make([]LogEntry, 100)
    for i := 0; i < 100; i++ {
        logs[i] = LogEntry{ID: i, Message: fmt.Sprintf("log-%d", i), Level: "INFO"}
    }

    start := time.Now()
    results := processLogsConcurrent(logs, 10)
    elapsed := time.Since(start)

    fmt.Printf("处理 %d 条日志, 用时 %v\n", len(results), elapsed)
    fmt.Printf("当前 goroutine 数: %d\n", runtime.NumGoroutine())
}

输出：

TEXT 📖 仅展示 
处理 100 条日志, 用时 2.05s
当前 goroutine 数: 1

(3) 收益：串行 vs 并发

处理方式 100 条日志 10 万条日志 goroutine 数
串行 20s 5.5h 1
10 个并发 worker 2s 33min ~12
100 个并发 worker 0.2s 3.3min ~102
💡 提示： goroutine 不是越多越好。IO 密集型任务推荐 worker count = GOMAXPROCS * 2~10，CPU 密集型推荐 worker count = GOMAXPROCS。

3. goroutine 基础

(1) go 关键字

GO 
package main

import (
    "fmt"
    "time"
)

func printNumbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("%d ", i)
    }
}

func printLetters() {
    for _, c := range "ABCDE" {
        time.Sleep(150 * time.Millisecond)
        fmt.Printf("%c ", c)
    }
}

func main() {
    go printNumbers()  // 在 goroutine 中并发执行
    go printLetters()
    time.Sleep(1 * time.Second)
    fmt.Println()
}

输出（每次可能不同）：

TEXT 📖 仅展示 
1 A 2 B 3 C 4 D 5 E

(2) 匿名 goroutine

GO 
package main

import (
    "fmt"
    "time"
)

func main() {
    go func() {
        fmt.Println("匿名 goroutine 执行中")
    }()

    go func(msg string) {
        fmt.Println("带参数的匿名 goroutine:", msg)
    }("hello")

    time.Sleep(100 * time.Millisecond)
    fmt.Println("main 结束")
}
🔥 易错： 主函数返回时所有 goroutine 都会被强制终止。上面 time.Sleep 是为了等 goroutine 完成——生产代码应该用 sync.WaitGroup。

4. sync.WaitGroup 等待同步

(1) WaitGroup 三种方法

GO 
package main

import (
    "fmt"
    "sync"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()  // 3. 告诉 WG 本 worker 完成
    fmt.Printf("Worker %d 开始\n", id)
    // 假装工作...
    fmt.Printf("Worker %d 结束\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 3; i++ {
        wg.Add(1)   // 1. 增加计数器
        go worker(i, &wg)
    }

    wg.Wait()  // 2. 阻塞等待所有 worker 完成
    fmt.Println("所有 worker 完成！")
}
method 作用
wg.Add(delta int) 增加计数器（通常在启动 goroutine 前调用）
wg.Done() 减少计数器（通常在 goroutine 内部 defer call）
wg.Wait() 阻塞直到计数器归零

▶ 示例：WaitGroup + 闭包陷阱

GO 
package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    // ❌ 错误：闭包捕获循环变量
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Printf("错误的 i=%d\n", i)  // 全部输出 3 或 4
        }()
    }
    wg.Wait()

    // ✅ 正确：传参拷贝
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("正确的 i=%d\n", id)
        }(i)
    }
    wg.Wait()
}
▶ 试一试
🔥 易错（经典闭包陷阱）： goroutine 闭包直接捕获循环变量 i——goroutine 启动时循环可能已经结束，i 的值已被覆盖。必须传参拷贝。

5. goroutine vs OS 线程

(1) 核心区别

GO 
package main

import (
    "fmt"
    "runtime"
)

func main() {
    // 查看当前 GOMAXPROCS
    fmt.Printf("GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
    fmt.Printf("CPU 核数: %d\n", runtime.NumCPU())
    fmt.Printf("当前 goroutine 数: %d\n", runtime.NumGoroutine())
}
维度 OS 线程 Goroutine
栈初始大小 1~8 MB 2 KB
最大栈 固定 1~8 MB 动态增长到 1 GB
创建开销 ~1 µs ~0.1 µs
上下文切换 内核态（~1 µs） 用户态（~0.1 µs）
数量上限 ~10,000 数百万
调度器 OS 内核调度 Go runtime GMP

(2) GMP 调度模型

100% 
graph TB
    G[G: Goroutine 队列] --> P[P: 逻辑处理器<br/>GOMAXPROCS 个]
    P --> M[M: OS 线程<br/>由内核调度]
    M --> CPU[CPU 核]
    
    subgraph 全局队列
        GQ[(全局 Goroutine 队列)]
    end
    
    GQ --> P
    
    style P fill:#e1f5fe
    style M fill:#fff3e0

GMP： G（Goroutine）— P（Processor，逻辑处理器，数量 = GOMAXPROCS）— M（Machine，OS thread）。Go runtime 把 G 调度到 P 的本地队列，P 再绑定到 M 执行。

6. GOMAXPROCS 与并发控制

(1) GOMAXPROCS 设置

GO 
package main

import (
    "fmt"
    "runtime"
)

func main() {
    fmt.Printf("默认 GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))

    // CPU 密集型：GOMAXPROCS = NumCPU
    // IO 密集型：GOMAXPROCS = NumCPU * 2~10
    runtime.GOMAXPROCS(4)  // 设置为 4
    fmt.Printf("设置后 GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
}

▶ 示例：GOMAXPROCS 对性能的影响

GO 
package main

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

func cpuIntensiveTask() {
    sum := 0
    for i := 0; i < 10000000; i++ {
        sum += i
    }
}

func benchmarkGOMAXPROCS(n int) time.Duration {
    runtime.GOMAXPROCS(n)
    var wg sync.WaitGroup
    start := time.Now()

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

    wg.Wait()
    return time.Since(start)
}

func main() {
    for _, p := range []int{1, 2, 4} {
        elapsed := benchmarkGOMAXPROCS(p)
        fmt.Printf("GOMAXPROCS=%d: %v\n", p, elapsed)
    }
}
▶ 试一试

7. goroutine 泄漏与排查

(1) 泄漏场景

GO 
package main

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

// 泄漏的 goroutine：从 channel 读，但没有关闭
func leakyGoroutine() {
    ch := make(chan int)
    go func() {
        <-ch  // 永远阻塞，没人往 ch 发数据
    }()
}

func main() {
    for i := 0; i < 10; i++ {
        leakyGoroutine()
    }

    time.Sleep(100 * time.Millisecond)
    fmt.Printf("泄漏后 goroutine 数: %d\n", runtime.NumGoroutine())
    // 输出: 泄漏后 goroutine 数: 11（10 个泄漏 + 1 个 main）
}

▶ 示例：启动 goroutine 的 4 种方式

GO 
package main

import (
    "fmt"
    "sync"
)

func say(msg string) {
    fmt.Println(msg)
}

func main() {
    var wg sync.WaitGroup

    // 方式 1：命名函数
    wg.Add(1)
    go func() {
        defer wg.Done()
        say("方式1: 命名函数")
    }()

    // 方式 2：匿名函数
    wg.Add(1)
    go func() {
        defer wg.Done()
        fmt.Println("方式2: 匿名函数")
    }()

    // 方式 3：匿名函数带参数（推荐，避免闭包陷阱）
    wg.Add(1)
    go func(msg string) {
        defer wg.Done()
        fmt.Println(msg)
    }("方式3: 带参数")

    // 方式 4：函数作为值
    wg.Add(1)
    fn := func() {
        defer wg.Done()
        fmt.Println("方式4: 函数变量")
    }
    go fn()

    wg.Wait()
}
▶ 试一试

▶ 示例：goroutine 与 OS 线程栈对比

GO 
package main

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

func main() {
    var memStats runtime.MemStats
    var wg sync.WaitGroup
    num := 100000  // 10 万 goroutine

    runtime.GC()
    runtime.ReadMemStats(&memStats)
    before := memStats.HeapAlloc

    for i := 0; i < num; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            _ = n
        }(i)
    }
    wg.Wait()

    runtime.ReadMemStats(&memStats)
    after := memStats.HeapAlloc
    perGoroutine := float64(after-before) / float64(num)

    fmt.Printf("启动 %d 个 goroutine\n", num)
    fmt.Printf("内存增量: %.2f MB\n", float64(after-before)/1024/1024)
    fmt.Printf("每个 goroutine 约: %.2f KB\n", perGoroutine/1024)
    fmt.Printf("总 goroutine 数: %d\n", runtime.NumGoroutine())
}
▶ 试一试

输出：

TEXT 📖 仅展示 
启动 100000 个 goroutine
内存增量: 21.45 MB
每个 goroutine 约: 0.22 KB
总 goroutine 数: 1
💡 提示： 每个 goroutine 栈初始只分配 2KB，而 OS 线程默认 1~8MB。10 万 goroutine ≈ 21MB，10 万线程 ≈ 80GB。

▶ 示例：goroutine 生命周期监控

GO 
package main

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

func safeWorker(done chan struct{}) {
    <-done  // 等待退出信号
}

func main() {
    done := make(chan struct{})

    // 启动 5 个 worker
    for i := 0; i < 5; i++ {
        go safeWorker(done)
    }

    fmt.Printf("启动后 goroutine 数: %d\n", runtime.NumGoroutine())

    // 发送退出信号
    close(done)
    time.Sleep(10 * time.Millisecond)

    fmt.Printf("关闭后 goroutine 数: %d\n", runtime.NumGoroutine())
}
▶ 试一试

(2) 泄漏防范清单

场景 防范方式
从 channel 读但没人写 用 buffered channel 或 select+default
向 channel 写但没人读 确保有消费者或 select+default
goroutine 死循环 通过 done channel 或 context 控制退出
time.After 在循环中泄漏 每次循环创建新 timer，用 time.NewTimer + Stop
select {} 空 确保有退出路径

8. 完整示例：10 万条日志并发处理

GO 
// log_processor.go
package main

import (
    "fmt"
    "math/rand"
    "runtime"
    "sync"
    "time"
)

type LogLevel int

const (
    Info LogLevel = iota
    Warn
    Error
    Debug
)

type LogLine struct {
    Timestamp time.Time
    Level     LogLevel
    Message   string
    Source    string
}

type ProcessedLog struct {
    Original  LogLine
    Severity  string
    Alert     bool
    ProcessedAt time.Time
}

func (l LogLine) Process() ProcessedLog {
    // 模拟处理耗时（IO 等待）
    time.Sleep(time.Duration(50+rand.Intn(50)) * time.Millisecond)

    severity := "low"
    alert := false
    switch l.Level {
    case Error:
        severity = "critical"
        alert = true
    case Warn:
        severity = "medium"
    }

    return ProcessedLog{
        Original:    l,
        Severity:    severity,
        Alert:       alert,
        ProcessedAt: time.Now(),
    }
}

// 单线程处理
func processSerial(logs []LogLine) []ProcessedLog {
    results := make([]ProcessedLog, 0, len(logs))
    for _, log := range logs {
        results = append(results, log.Process())
    }
    return results
}

// 并发处理：worker pool 模式
func processConcurrent(logs []LogLine, workers int) []ProcessedLog {
    jobs := make(chan LogLine, len(logs))
    results := make(chan ProcessedLog, len(logs))

    // 填充任务
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    var wg sync.WaitGroup

    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                results <- log.Process()
            }
        }(w)
    }

    // 等待所有 worker 完成，然后关闭 results
    go func() {
        wg.Wait()
        close(results)
    }()

    // 收集结果
    processed := make([]ProcessedLog, 0, len(logs))
    for r := range results {
        processed = append(processed, r)
    }
    return processed
}

func generateLogs(count int) []LogLine {
    levels := []LogLevel{Info, Warn, Error, Debug}
    sources := []string{"api-gateway", "user-service", "payment", "database"}

    logs := make([]LogLine, count)
    for i := 0; i < count; i++ {
        logs[i] = LogLine{
            Timestamp: time.Now(),
            Level:     levels[rand.Intn(len(levels))],
            Message:   fmt.Sprintf("log-entry-%d", i),
            Source:    sources[rand.Intn(len(sources))],
        }
    }
    return logs
}

func main() {
    const logCount = 10000  // 1 万条演示，实际可 10 万
    logs := generateLogs(logCount)

    fmt.Printf("日志量: %d\n", logCount)
    fmt.Printf("CPU 核数: %d, GOMAXPROCS: %d\n", runtime.NumCPU(), runtime.GOMAXPROCS(0))

    // 串行处理
    start := time.Now()
    serialResults := processSerial(logs)
    serialTime := time.Since(start)
    fmt.Printf("\n串行: %v (%d 条/秒)\n", serialTime,
        int(float64(logCount)/serialTime.Seconds()))

    // 并发处理（10 worker）
    runtime.GC()
    start = time.Now()
    concurrentResults := processConcurrent(logs, 10)
    concurrentTime := time.Since(start)
    fmt.Printf("并发(10): %v (%d 条/秒)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    // 并发处理（50 worker）
    runtime.GC()
    start = time.Now()
    concurrentResults = processConcurrent(logs, 50)
    concurrentTime = time.Since(start)
    fmt.Printf("并发(50): %v (%d 条/秒)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    fmt.Printf("\n结果验证: serial=%d, concurrent=%d\n",
        len(serialResults), len(concurrentResults))
    fmt.Printf("当前 goroutine 数: %d\n", runtime.NumGoroutine())

    // 统计告警
    alertCount := 0
    for _, r := range concurrentResults {
        if r.Alert {
            alertCount++
        }
    }
    fmt.Printf("告警数: %d\n", alertCount)
}

预期输出：

TEXT 📖 仅展示 
日志量: 10000
CPU 核数: 8, GOMAXPROCS: 8

串行: 5.2s (1923 条/秒)
并发(10): 520ms (19230 条/秒)
并发(50): 110ms (90909 条/秒)

结果验证: serial=10000, concurrent=10000
当前 goroutine 数: 1
告警数: 2453
🔥 易错： goroutine 内的 panic 会导致整个程序崩溃，不会被外层 recover 捕获。每个 goroutine 的入口处应该用 defer recover() 保护。第 14 课 channel 会深入这个模式。

❓ 常见问题

Q goroutine 和 OS 线程有什么区别？
A goroutine 是 Go runtime 管理的用户态"协程"，栈初始仅 2KB（thread 1~8MB），创建快 10 倍，切换快 10 倍。上百万 goroutine 对 Go 是平常事，上万个线程系统就扛不住了。
Q 如何等待所有 goroutine 完成？
A 用 sync.WaitGroup——wg.Add(n) 计数，wg.Done() 减一，wg.Wait() 阻塞等待归零。注意 Add 必须在外层或启动 goroutine 前调用，Done 用 defer 确保执行。
Q goroutine 泄漏怎么排查？
A runtime.NumGoroutine() 看数量是否持续增长；net/http/pprof 看 goroutine 堆栈。常见泄漏：channel 读写阻塞、select 没有 default、for 循环启动 goroutine 不控制退出。
Q runtime.Gosched 有什么用？
A 主动让出 P，让其他 goroutine 有机会执行。极少需要手动调用——Go 在 IO 等待、channel 操作、time.Sleep 等场景会自动调度。
Q goroutine 数量上限是多少？
A 理论上受内存限制——每个 goroutine 栈 2KB 起步，4GB 内存 ≈ 200 万 goroutine。实际推荐：CPU 密集型 ≤ GOMAXPROCS，IO 密集型 ≤ 1000~10000。
Q GOMAXPROCS 应该设多少？
A Go 1.5+ 默认 = NumCPU（CPU 核数）。一般不用改：CPU 密集型保持默认，IO 密集型可适当提高（2~10 倍），但本质瓶颈在 IO 速度而非并行数。
Q goroutine 里的 panic 会影响其他 goroutine 吗？
A 会！任何一个 goroutine 的未恢复 panic 都会导致整个进程崩溃。所以每个 goroutine 入口应该 defer func() { if r := recover(); r != nil { ... } }()。
Q go 关键字启动的函数必须是无参数的？
A 不，可以传参：go myFunc(arg1, arg2)。闭包捕获变量要小心循环变量陷阱——用传参拷贝而不是直接捕获。

📖 小节

📝 作业

  1. 基础题（难度⭐）：写一个程序启动 5 个 goroutine，每个打印自己的编号和 "Hello"，用 sync.WaitGroup 等待全部完成，验证 goroutine 执行顺序随机。

  2. 进阶题（难度⭐⭐）：实现一个并发素数筛选器：输入一个数字 n，用 4 个 goroutine 并发检查 1~n 中哪些是素数，返回素数列表。要求用 WaitGroup 同步，比较并发 vs 串行的性能差异。

  3. 挑战题（难度⭐⭐⭐）：构建一个并发任务调度器：输入一批任务（[]func() Result），用 worker pool 模式并发执行，支持：(1) 可配置 worker 数；(2) 错误容忍（单个 goroutine panic 不影响其他）；(3) 进度回调（每完成 10% 打印一次）；(4) runtime.NumGoroutine 泄漏检测。

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏