我正在尝试并发运行多个任务，如果有任何错误则立即返回，而无需等待所有例程返回。代码如下所示。我已经去除了噪音以使其更容易消化，但如果泄漏不明显，我可以发布完整的代码。 值得注意的是，我正在将其部署在 google app engine 上。我无法在我的机器上重现泄漏，但是当我在 //Consume the results 评论之后替换并发时，应用程序工作正常，但我不明白为什么因为代码对我来说看起来是正确的.

package main

import "fmt"
import "sync"
import "errors"

func main() {
    indexes := []int{1, 2, 3, 4, 5, 6, 7}
    devCh := make(chan int, 7)
    stopCh := make(chan struct{})
    errCh := make(chan error, 7)
    var wg sync.WaitGroup
    go func() {
        for _, sub := range indexes {
            wg.Add(1)
            go func(sub int) {
                defer wg.Done()
                // some code which creates other
                // wait groups and spans other go routines
                // handle errors
                if sub == 99 { // unreachable 
                    errCh <- errors.New("new error")

                }
            }(sub)
            select {
            // If there is any error we better stop the
            // loop
            case <-stopCh:
                return
            default:
            }
            devCh <- sub
        }
        wg.Wait()
        close(devCh)
    }()
    // Consume the results
    var results []int
    var wt sync.WaitGroup
    wt.Add(1)
    go func() {
        defer wt.Done()
        for s := range devCh {
            results = append(results, s)
        }
        return
    }()
    done := make(chan struct{})
    go func() {
        wt.Wait()
        close(done)
    }()

L:
    for {
        select {
        case err := <-errCh:
            fmt.Printf("error was %v", err)
            close(stopCh)
            return
        case <-done:
            break L
        default:
        }
    }
    fmt.Printf("all done, %v", results)
}

编辑:添加了一些工作代码。

编辑:添加了更接近真实代码的代码，这可能解释了 for 循环的需要。

package main

import "fmt"
import "sync"
import "errors"

func main() {
    indexes := []int{1, 2, 3, 4, 5, 6, 7}
    indexesString := []string{"a", "b", "c", "d"}
    devChS := make(chan string, 1000)

    devCh := make(chan int, 7)
    stopCh := make(chan struct{})
    errCh := make(chan error, 7)
    var wg sync.WaitGroup
    go func() {
        for _, sub := range indexes {
            wg.Add(1)
            go func(sub int) {
                defer wg.Done()
                // some code which creates other
                // wait groups and spans other go routines
                // handle errors
                if sub == 99 { // unreachable
                    errCh <- errors.New("new error")

                }
                wg.Add(1)
                go func(sub int) {
                    defer wg.Done()
                    for _, s := range indexesString {
                        devChS <- fmt.Sprintf("%s %s", s, sub)

                    }

                    return
                }(sub)
            }(sub)
            select {
            // If there is any error we better stop the
            // loop
            case <-stopCh:
                return
            default:
            }
            devCh <- sub
        }
        wg.Wait()
        close(devCh)
        close(devChS)
    }()
    // Consume the results
    var results = struct {
        integers []int
        strings  []string
    }{}
    var wt sync.WaitGroup
    wt.Add(1)
    go func() {
        defer wt.Done()
        for s := range devCh {
            results.integers = append(results.integers, s)
        }
        return
    }()
    wt.Add(1)
    go func() {
        defer wt.Done()
        for s := range devChS {
            results.strings = append(results.strings, s)
        }
        return
    }()
    done := make(chan struct{})
    go func() {
        wt.Wait()
        close(done)
    }()

L:
    for {
        select {
        case err := <-errCh:
            fmt.Printf("error was %v", err)
            close(stopCh)
            return
        case <-done:
            break L
        default:
        }
    }
    fmt.Printf("all done, can return the results: %v", results)
}

最佳答案

tl;dr:一个只重复非阻塞检查直到成功的循环可能会导致难以诊断的问题(至少，它会过度使用 CPU)；使用阻塞检查可以修复它。

我不太确定你的案例的细节；我写了一个像你一样的循环，在 Playground 上一直挂起“进程耗时太长”，但是当我在本地运行它时它确实完成了。

正如我所说，我的目标也是更简单的设计。

Go only has limited pre-emption of running goroutines : 运行线程只有在发生阻塞操作(如 I/O 或 channel 操作或等待获取锁)时才会将控制权交给 goroutine 调度程序。

因此对于 GOMAXPROCS=1，如果(一个)正在运行的线程开始循环，则其他任何线程都不一定有机会运行。

for { select { ...default: } } 因此可以开始循环检查 channel 中的项目，但绝不会放弃对主线程的控制，以便另一个 goroutine 可以写入项目.当 GOMAXPROCS 超过 1 时，其他代码仍会运行，但不会像在 App Engine(或 Playground)上那样为 1。该行为不仅取决于 GOMAXPROCS，还取决于哪个 goroutine 最先运行，这不一定定义。

为避免这种情况，请删除 default:，这样 select 是一个阻塞操作，当调度程序无法接收项目时，它会屈服于调度程序，从而允许其他代码运行。您可以将其推广到其他可能循环执行非阻塞检查的情况；当阻塞调用不会时，它们中的任何一个都可以使资源忙于不断地重新检查。当 GOMAXPROCS>1 或运行时的有限抢占为您节省时间时，轮询(调用重复检查)仍然会比阻塞消耗更多的 CPU。

例如，这会失败并显示“处理时间太长”on the Playground ，虽然很烦人，但它在我的机器上可靠地完成了:

package main

import "fmt"

func main() {
    c := make(chan struct{})
    go func() { c <- struct{}{} }()
    for {
        select {
        case <-c:
            fmt.Println("success")
            return
        default:
        }
    }
} 

我不知道是否还有其他问题，但值得注意的是类似于示例的模式的挂起。

关于google-app-engine - Go例程泄漏在哪里？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/29269416/