Golang基于带缓冲Channel实现并行Map时的死锁问题排查

顺着代码的执行路径逐层拆解，卡死的核心是两个逻辑上的闭环问题，和通道带不带缓冲没有关系：

• 通道关闭的责任主体完全错位
  你写的汇聚goroutine逻辑是这样的：

  for i := 0; i < p; i++ {
      for val := range output[i] { // range会一直阻塞，等通道关闭才退出循环
          out <- val
      }
      close(output[i]) // 必须等上面的range退出，才能执行close
  }
  

  这刚好踩中了Go通道的核心规则：range 遍历通道的退出条件是「通道被主动关闭，且缓冲区所有值被读取完毕」。你把close(output[i])放在range循环的后面，等于让range等close触发退出，又让close等range执行完才能触发，直接形成逻辑死循环。
  实际运行的流程是：
  1. 第一个分片的worker往output[0]写满10个值，把10的缓冲槽全占满，因为没有关闭通道，写完直接执行wg.Done()退出
  2. 汇聚goroutine从output[0]里把10个缓冲值全读出来，依次写到out通道，写完之后依然卡在range循环上，等着output[0]发来第11个值——但worker早就写完了，没人会发新值了
  3. 因为卡在i=0的分支上，汇聚goroutine根本不会走到i=1的循环去读output[1]的内容，这就是你看到output[1]从来没被读取的直接原因。
• out通道的关闭时机倒置
  你原来的close(out)是放在主协程的wg.Wait()后面的，但消费out通道的goroutine是靠range out读数据的，这个goroutine必须等out被关闭才能退出、执行wg.Done()。可现在汇聚goroutine已经卡在output[0]的range上，永远不会执行wg.Done()，主协程就永远等不到wg.Wait()返回，自然也没法关闭out通道，整条链路全卡死。
• 代码里的time.Sleep(time.Second * 6)完全没用，死锁是逻辑闭环导致的，不是调度时序的问题，睡再久卡死的状态也不会变。

对着上面的问题逐点修正，核心是把通道关闭的逻辑还给写端，理顺整个链路的退出顺序：

• 每个写分片的worker goroutine，写完自己负责的所有数据后，立刻主动关闭对应的output通道。毕竟worker最清楚自己什么时候写完了所有值，不能把“没有新值了”的信号传递责任扔给读端。
• 删掉汇聚goroutine里冗余的close(output[i])：worker已经会关闭通道，range自然会在读完全部值后退出，不需要读端再做close操作。
• 调整out通道的关闭时机：等所有分片的结果都被读出来、搬到out通道后，立刻主动关闭out，保证消费端的range能正常退出。
• 顺手处理一下切分分片的边界问题：如果输入长度不能被p整除，原来的切分逻辑会漏最后几个元素，给最后一个分片补全剩余的索引范围。

修复后的可运行代码：

func pmmap(inp []int, f func(int) int, p int) {
    var wg sync.WaitGroup
    var output []chan int
    tot := len(inp)
    out := make(chan int, tot)
    slice := tot / p

    // 提前初始化所有分片的通道
    for i := 0; i < p; i++ {
        output = append(output, make(chan int, slice))
    }

    // 启动分片计算worker
    for i := 0; i < p; i++ {
        startInd := slice * i
        endInd := startInd + slice
        // 最后一个分片兜住剩余元素，避免漏数据
        if i == p-1 {
            endInd = tot
        }

        wg.Add(1)
        go func(si, ei int, ch chan int) {
            defer wg.Done()
            for idx := si; idx < ei; idx++ {
                ch <- f(inp[idx])
            }
            // 写完所有数据立刻关通道，通知读端没有新值了
            close(ch)
        }(startInd, endInd, output[i])
    }

    // 启动汇聚任务
    wg.Add(1)
    go func() {
        defer wg.Done()
        // 依次读完所有分片的通道内容
        for _, ch := range output {
            for val := range ch {
                out <- val
            }
        }
        // 所有值都搬完，关out通道
        close(out)
    }()

    // 消费out通道：range会在out关闭、所有值读完后自动退出
    for val := range out {
        fmt.Println(val)
    }

    wg.Wait()
}

func add4(i int) int {
    return i + 4
}

func main() {
    temp := []int{}
    for i := 1; i <= 20; i++ {
        temp = append(temp, i)
    }
    pmmap(temp, add4, 2)
}

这个版本跑起来会正常输出5到24的全部计算结果，不会再触发死锁。

内容的提问来源于stack exchange，提问作者lightyagami1