go - 同步 worker 以进行递归抓取

Question

我想用 n 个 worker 实现一个“爬虫”，每个 worker 都可以添加额外的工作。当没有剩下的工作并且所有 worker 都完成了工作时，程序应该停止。

我有以下代码(您可以在 https://play.golang.org/p/_j22p_OfYv 中使用它):

package main

import (
    "fmt"
    "sync"
)

func main() {
    pathChan := make(chan string)
    fileChan := make(chan string)
    workers := 3
    var wg sync.WaitGroup

    paths := map[string][]string{
        "/":     {"/test", "/foo", "a", "b"},
        "/test": {"aa", "bb", "cc"},
        "/foo":  {"/bar", "bbb", "ccc"},
        "/bar":  {"aaaa", "bbbb", "cccc"},
    }

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            for {
                path, ok := <-pathChan
                if !ok {
                    break
                }

                for _, f := range paths[path] {
                    if f[0] == '/' {
                        pathChan <- f
                    } else {
                        fileChan <- f
                    }
                }
            }

            wg.Done()
        }()
    }

    pathChan <- "/"

    for {
        filePath, ok := <-fileChan
        if !ok {
            break
        }

        fmt.Println(filePath)
    }

    wg.Wait()
    close(pathChan)
}

不幸的是，这以死锁告终。问题到底出在哪里？另外，编写此类功能的最佳做法是什么？ channel 是正确使用的功能吗？

编辑:

我更新了我的代码以使用两个 WaitGroup ，一个用于作业，一个用于工作人员(参见 https://play.golang.org/p/bueUJzMhqj):

package main

import (
    "fmt"
    "sync"
)

func main() {
    pathChan := make(chan string)
    fileChan := make(chan string)
    jobs := new(sync.WaitGroup)
    workers := new(sync.WaitGroup)
    nworkers := 2

    paths := map[string][]string{
        "/":     {"/test", "/foo", "a", "b"},
        "/test": {"aa", "bb", "cc"},
        "/foo":  {"/bar", "bbb", "ccc"},
        "/bar":  {"aaaa", "bbbb", "cccc"},
    }

    for i := 0; i < nworkers; i++ {
        workers.Add(1)
        go func() {
            defer workers.Done()
            for {
                path, ok := <-pathChan
                if !ok {
                    break
                }

                for _, f := range paths[path] {
                    if f[0] == '/' {
                        jobs.Add(1)
                        pathChan <- f
                    } else {
                        fileChan <- f
                    }
                }

                jobs.Done()
            }

        }()
    }

    jobs.Add(1)
    pathChan <- "/"

    go func() {
        jobs.Wait()
        close(pathChan)
        workers.Wait()
        close(fileChan)
    }()

    for {
        filePath, ok := <-fileChan
        if !ok {
            break
        }

        fmt.Println(filePath)
    }

}

这似乎确实可行，但显然如果将nworkers设置为1，仍然会发生死锁，因为单个worker在向 channel 添加内容时将永远等待路径陈。要解决此问题，可以增加 channel 缓冲区(例如 pathChan := make(chan string, 2))，但这仅在两个缓冲区未完全满时才有效。当然，缓冲区大小可以设置为一个很大的数字，比如 10000，但是代码仍然会遇到死锁。此外，这对我来说似乎不是一个干净的解决方案。

这是我意识到使用某种队列而不是 channel 会更容易的地方，在 channel 中可以添加和删除元素而不会阻塞，并且队列的大小不固定。 Go 标准库中是否存在这样的队列？

Answer 1

如果你想等待任意数量的 worker 完成，标准库包括sync.WaitGroup正是为了这个目的。

还有其他并发问题:

• 您正在使用 channel 关闭信号，但您有多个 goroutines 在同一个 channel 上发送。这通常是一种不好的做法:因为每个例程永远无法知道其他例程何时完成 channel ，您永远无法正确关闭 channel 。
• 关闭一个 channel 会等待另一个 channel 先关闭，但它永远不会关闭，因此会出现死锁。
• 它不会立即死锁的唯一原因是您的示例碰巧拥有比“/”下的目录更多的 worker 。在“/”下再添加两个目录，立即死锁。

有一些解决方案:

• 转储工作池并为每个子目录旋转一个 goroutine，让调度程序处理其余部分:https://play.golang.org/p/ck2DkNFnyF
• 每个根级目录使用一个 worker，并让每个 worker 递归地处理其目录，而不是将它找到的子目录排队到一个 channel 。