去达到进程的最大线程数？

Question

我正在尝试使用 Go 来做一些文件系统使用分析，我通过将几乎所有东西都作为 goroutine 产生并依靠 Go VM(和 GOMAXPROCS)来管理它来尽可能快地编写代码。我一直在观察这段代码运行(非常快)，直到它停止运行。我检查了顶部，它列出了我的进程有 1500 个线程。

我想也许我已经达到了某个极限，因此进程在等待操作系统时陷入僵局。我检查了我的操作系统 (FreeBSD) 限制，果然它被列为每个进程最多 1500 个线程。

很惊讶，我查看了 Go 文档，它说 GOMAXPROCS 只是对运行线程的限制，但阻塞的线程不算数。

所以我的问题:

• 可以说我不能依赖 Go VM 作为全局池来防止达到此类操作系统限制吗？

• 有没有一种惯用的方法来处理这个问题(很好，这只是我使用 Go 的第二天)？

  • 特别是，除了在使用完 channel 后同步关闭 channel 外，我还没有找到其他好方法。有没有更好的办法？

  • 我想抽象出样板文件(使用 go routines 的并行映射和完成后关闭 channel )，是否有一种类型安全的方法可以在没有泛型的情况下执行此操作？

这是我当前的代码:

func AnalyzePaths(paths chan string) chan AnalyzedPath {
    analyzed := make(chan AnalyzedPath)
    go func() {
        group := sync.WaitGroup{}
        for path := range paths {
            group.Add(1)
            go func(path string) {
                defer group.Done()
                analyzed <- Analyze(path)
            }(path)
        }
        group.Wait()
        close(analyzed)
    }()
    return analyzed
}

func GetPaths(roots []string) chan string {
    globbed := make(chan string)
    go func() {
        group := sync.WaitGroup{}
        for _, root := range roots {
            group.Add(1)
            go func(root string) {
                defer group.Done()
                for _, path := range glob(root) {
                    globbed <- path
                }
            }(root)
        }
        group.Wait()
        close(globbed)
    }()
    return globbed
}

func main() {
    paths := GetPaths(patterns)
    for analyzed := range AnalyzePaths(paths) {
        fmt.Println(analyzed)
    }
}

Answer 1

大约 2 个月前(或更多)语言开发人员谈到了线程计数控制(以及其他一些限制)的入侵。所以我们可以期待很快看到它。一个月或更多年前，我开发了这个问题，发现在我的 Linux 机器上 GOMAXPROCS 没有超过 256 的值。如果我向它发送 300 或更多，结果总是 256。但我发现 goroutines 不是线程。 Goroutines 可以存在于一个线程中。

至于惯用的同步——我认为没有必要同步太多。在我的代码中，我通常使用 goroutines 仅通过 channel 进行通信的想法。 channel 应该作为 goroutines 的参数传递。

func main() {
    ch1 := make(chan SomeType1)
    ch2 := make(chan SomeType2)
    go generator(ch1, ch2)
    go processor(ch1, ch2)
    // here main func becomes waiting until it capture 2 of ch2-finished-signals 
    <- ch2
    <- ch2
    // usually we don't need the exact values of ch2-signals,
    // so we assign it to nothing 
}

func generator(ch1 chan SomeType1, ch2 chan SomeType2) {
    for (YOUR_CONDITION){
        // generate something
        //....
        // send to channel
        ch1 <- someValueOfType1
    }
    ch1 <- magicStopValue
    ch2 <- weAreFinishedSignal1
}

func processor(ch1 chan SomeType1, ch2 chan SomeType2) {
    // "read" value from ch1 
    value := <-ch1
    for value != magicStopValue {
        // make some processing
        // ....
        //get next value from ch1 and replay processing
        value = <- ch1
    }
    // here we can send signal that goroutine2 is finished
    ch2 <- weAreFinishedSignal2
}

如果 goroutines 在一个线程中，它们的通信速度会更快。对我而言， channel 性能远谈不上好，但足以满足许多目的。