深入理解 Golang 调度器之 GMP 模型

 1// G 的核心字段（简化版）
 2type g struct {
 3    stack       stack    // 栈内存范围 [lo, hi)
 4    sched       gobuf    // 保存调度上下文（PC、SP 等）
 5    gopc        uintptr  // 创建该 G 的 PC
 6    status      uint32   // 状态：_Gidle, _Grunnable, _Grunning 等
 7    goid        int64    // Goroutine ID
 8    m           *m       // 当前绑定的 M
 9    waitreason  string   // 等待原因
10}

 1// M 的核心字段（简化版）
 2type m struct {
 3    g0       *g      // 特殊的 G，用于执行调度代码
 4    curg     *g      // 当前运行的 G
 5    p        *p      // 绑定的 P
 6    nextp    *p      // 即将绑定的 P
 7    spinning bool    // 是否处于自旋状态（正在寻找 G）
 8    blocked  bool    // 是否被阻塞
 9    park     note    // 休眠唤醒机制
10}

 1// P 的核心字段（简化版）
 2type p struct {
 3    id          int32       // P 的 ID
 4    status      uint32      // 状态：_Pidle, _Prunning 等
 5    m           *m          // 绑定的 M
 6    runqhead    uint32      // 本地运行队列头
 7    runqtail    uint32      // 本地运行队列尾
 8    runq        [256]*g     // 本地运行队列（环形队列）
 9    runnext     *g          // 优先运行的 G
10    gFree       *g          // 空闲 G 列表（复用）
11    gcBgMarkWorker *g       // GC 后台标记 Worker
12}

1// gobuf 保存 G 的调度信息
2type gobuf struct {
3    sp   uintptr  // 栈指针
4    pc   uintptr  // 程序计数器
5    g    guintptr // 指向 G
6    ret  uintptr  // 返回值
7}

 1// 简化的调度代码
 2func schedule() {
 3    // 每 61 次调度，优先从全局队列取 G
 4    if schedCount%61 == 0 {
 5        if gp := globrunqget(_p_, 1); gp != nil {
 6            return gp
 7        }
 8    }
 9    // ...
10}

 1// go func() 的实现（简化版）
 2func newproc(fn *funcval) {
 3    // 1. 从 P 的空闲列表获取 G，或新建 G
 4    gp := gfget(_p_)
 5    if gp == nil {
 6        gp = malg(_StackMin)  // 分配 2KB 栈
 7    }
 8
 9    // 2. 初始化 G
10    gp.sched.pc = fn.pc
11    gp.sched.sp = gp.stack.hi
12
13    // 3. 放入运行队列
14    runqput(_p_, gp, true)  // true 表示放入 runnext
15}

1# 显示调度器详细信息
2GODEBUG=schedtrace=1000 ./myprogram
3
4# 输出示例
5SCHED 1000ms: gomaxprocs=8 idleprocs=0 threads=10 spinningthreads=1 idlethreads=0 runqueue=0 [0 0 0 0 0 0 0 0]

 1import (
 2    _ "net/http/pprof"
 3    "net/http"
 4)
 5
 6func main() {
 7    go func() {
 8        http.ListenAndServe("localhost:6060", nil)
 9    }()
10
11    // 你的程序...
12}

1# 查看 Goroutine 数量和状态
2curl http://localhost:6060/debug/pprof/goroutine?debug=1
3
4# 使用 go tool pprof 分析
5go tool pprof http://localhost:6060/debug/pprof/goroutine

 1import (
 2    "runtime/trace"
 3    "os"
 4)
 5
 6func main() {
 7    f, _ := os.Create("trace.out")
 8    defer f.Close()
 9
10    trace.Start(f)
11    defer trace.Stop()
12
13    // 你的程序...
14}

1# 生成追踪文件后，使用 go tool trace 分析
2go tool trace trace.out

1// 默认等于 CPU 核心数
2// 在容器环境中可能需要手动设置
3import "runtime"
4
5func init() {
6    runtime.GOMAXPROCS(4)  // 设置为 4 个 P
7}

1// 使用 automaxprocs 自动设置
2import _ "go.uber.org/automaxprocs"

 1// ❌ 错误示例：Goroutine 泄漏
 2func leak() {
 3    ch := make(chan int)
 4    go func() {
 5        <-ch  // 永远阻塞，因为没有人发送数据
 6    }()
 7    // 函数返回，但 Goroutine 仍在运行
 8}
 9
10// ✅ 正确示例：使用 context 取消
11func noLeak(ctx context.Context) {
12    ch := make(chan int)
13    go func() {
14        select {
15        case <-ch:
16            // 处理数据
17        case <-ctx.Done():
18            return  // 提前退出
19        }
20    }()
21}

 1// ❌ 错误示例：无限制创建 Goroutine
 2func processAll(items []Item) {
 3    for _, item := range items {
 4        go process(item)  // 可能创建数百万个 Goroutine
 5    }
 6}
 7
 8// ✅ 正确示例：使用工作池
 9func processAll(items []Item, workers int) {
10    sem := make(chan struct{}, workers)
11    var wg sync.WaitGroup
12
13    for _, item := range items {
14        wg.Add(1)
15        go func(item Item) {
16            defer wg.Done()
17            sem <- struct{}{}  // 获取信号量
18            defer func() { <-sem }()  // 释放信号量
19            process(item)
20        }(item)
21    }
22    wg.Wait()
23}

 1// ❌ 错误示例：每次请求创建新 Goroutine
 2func handleRequest(req Request) {
 3    go func() {
 4        process(req)
 5    }()
 6}
 7
 8// ✅ 正确示例：使用 Goroutine 池
 9var pool = &sync.Pool{
10    New: func() interface{} {
11        return new(Worker)
12    },
13}
14
15func handleRequest(req Request) {
16    w := pool.Get().(*Worker)
17    defer pool.Put(w)
18    w.process(req)
19}

 1// ❌ 错误示例：所有 Goroutine 捕获同一个变量
 2for i := 0; i < 10; i++ {
 3    go func() {
 4        fmt.Println(i)  // 可能全部打印 10
 5    }()
 6}
 7
 8// ✅ 正确示例：传递参数
 9for i := 0; i < 10; i++ {
10    go func(n int) {
11        fmt.Println(n)
12    }(i)
13}

 1func processConcurrently(items []Item) error {
 2    var wg sync.WaitGroup
 3    errCh := make(chan error, len(items))
 4
 5    for _, item := range items {
 6        wg.Add(1)
 7        go func(item Item) {
 8            defer wg.Done()
 9            if err := process(item); err != nil {
10                errCh <- err
11            }
12        }(item)
13    }
14
15    wg.Wait()
16    close(errCh)
17
18    // 返回第一个错误
19    for err := range errCh {
20        return err
21    }
22    return nil
23}

1import "runtime"
2
3func main() {
4    fmt.Printf("Goroutine 数量: %d\n", runtime.NumGoroutine())
5}

1# 使用 GODEBUG 查看调度信息
2GODEBUG=schedtrace=1000,scheddetail=1 ./myprogram

1// 使用 runtime.LockOSThread 将 G 绑定到当前 M
2func main() {
3    runtime.LockOSThread()
4    defer runtime.UnlockOSThread()
5
6    // 这段代码始终在同一个 OS 线程上执行
7    // 适用于需要线程局部存储或特定 CPU 亲和性的场景
8}