深入 Go 语言并发模式：从 Goroutine 调度到 Channel 高级原语

// 获取当前逻辑处理器数量（通常等于 CPU 核心数）
n := runtime.GOMAXPROCS(0) // 返回当前 P 的数量

// 某个 Goroutine 阻塞时，调度器会：
// 1. 解绑 M 和 P
// 2. P 带着本地 G 队列绑定到空闲或新的 M
// 3. 原 M 继续等待阻塞操作完成

type hchan struct {
    qcount   uint           // 缓冲区中当前元素个数
    dataqsiz uint           // 缓冲区容量（无缓冲为 0）
    buf      unsafe.Pointer // 环形缓冲区指针
    elemsize uint16         // 单个元素大小
    closed   uint32         // 关闭标记
    elemtype *_type         // 元素类型信息
    sendx    uint           // 发送索引
    recvx    uint           // 接收索引
    recvq    waitq          // 接收等待队列
    sendq    waitq          // 发送等待队列
    lock     mutex          // 全局互斥锁
}

// Generator：生成整数流
func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// SQ：计算平方
func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func main() {
    // gen → sq → 消费
    for n := range sq(sq(generator(2, 3, 4, 5))) {
        fmt.Println(n) // 输出 16, 25, 64, 100
    }
}

// Fan-out: 从源 Channel 读取，分发到多个 Worker
func fanOut(source <-chan int, workers int) []<-chan int {
    outputs := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan int)
        outputs[i] = ch
        go func() {
            defer close(ch)
            for v := range source {
                ch <- v * 2  // 模拟处理
            }
        }()
    }
    return outputs
}

// Fan-in: 将多个 Channel 合并为一个
func fanIn(channels ...<-chan int) <-chan int {
    merged := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(channels))
    
    for _, ch := range channels {
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                merged <- v
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(merged)
    }()
    return merged
}

type Job struct {
    ID      int
    Payload interface{}
    Result  chan interface{}
}

func WorkerPool(numWorkers int, jobs <-chan Job) {
    for i := 0; i < numWorkers; i++ {
        go func(workerID int) {
            for job := range jobs {
                result := process(job.Payload)
                job.Result <- result
            }
        }(i)
    }
}

// 最多允许 10 个并发
sem := make(chan struct{}, 10)

for _, task := range tasks {
    go func(t Task) {
        sem <- struct{}{}        // P 操作：获取许可
        defer func() { <-sem }()  // V 操作：释放许可
        process(t)
    }(task)
}

import "sync/errgroup"

var g errgroup.Group
for _, url := range urls {
    u := url
    g.Go(func() error {
        return fetch(u)
    })
}
if err := g.Wait(); err != nil {
    // 首个错误触发取消，其余 Goroutine 通过 ctx.Done 收到信号
}