当我在 goroutine 中运行 wg.Wait() 时，为什么我的代码可以正常工作？

Question

我有一个正在抓取的网址列表。我想要做的是将所有成功抓取的页面数据存储到一个通道中，当我完成后，将其转储到一个切片中。我不知道我会获得多少成功，所以我不能指定一个固定的长度。我希望代码到达wg.Wait()，然后等到所有wg.Done() 方法被调用，但我从未到达close(queue) 语句。寻找类似的答案，我遇到了这个 SO 答案

https://stackoverflow.com/a/31573574/5721702

作者做了类似的事情：

ports := make(chan string)
toScan := make(chan int)
var wg sync.WaitGroup

// make 100 workers for dialing
for i := 0; i < 100; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for p := range toScan {
            ports <- worker(*host, p)
        }
    }()
}

// close our receiving ports channel once all workers are done
go func() {
    wg.Wait()
    close(ports)
}()

只要我将wg.Wait() 包裹在 goroutine 中，就会到达 close(queue)：

urls := getListOfURLS()
activities := make([]Activity, 0, limit)
queue := make(chan Activity)
for i, activityURL := range urls {
    wg.Add(1)
    go func(i int, url string) {
        defer wg.Done()
        activity, err := extractDetail(url)
        if err != nil {
            log.Println(err)
            return
        }
        queue <- activity
    }(i, activityURL)
}
    // calling it like this without the goroutine causes the execution to hang
// wg.Wait() 
// close(queue)

    // calling it like this successfully waits
go func() {
    wg.Wait()
    close(queue)
}()
for a := range queue {
    // block channel until valid url is added to queue
    // once all are added, close it
    activities = append(activities, a)
}

如果我不为wg.Wait() 使用goroutine，为什么代码无法到达close？我认为所有defer wg.Done() 语句都被调用，所以最终它会清除，因为它到达wg.Wait()。是否与在我的频道中接收值有关？

Answer 1

您需要等待 goroutine 在单独的线程中完成，因为需要从中读取 queue。当您执行以下操作时：

queue := make(chan Activity)
for i, activityURL := range urls {
    wg.Add(1)
    go func(i int, url string) {
        defer wg.Done()
        activity, err := extractDetail(url)
        if err != nil {
            log.Println(err)
            return
        }
        queue <- activity // nothing is reading data from queue.
    }(i, activityURL)
}

wg.Wait() 
close(queue)

for a := range queue {
    activities = append(activities, a)
}

每个 goroutine 在 queue <- activity 处阻塞，因为 queue 是无缓冲的，没有任何东西从它读取数据。这是因为queue 上的范围循环在wg.Wait 之后的主线程中。

wg.Wait 只会在所有 goroutine 返回后解除阻塞。但如前所述，所有的 goroutine 在通道发送时都被阻塞了。

当你使用单独的 goroutine 等待时，代码执行实际上会到达queue 上的范围循环。

// wg.Wait does not block the main thread.
go func() {
    wg.Wait()
    close(queue)
}()

这导致 goroutine 在 queue <- activity 语句处解除阻塞（主线程开始读取 queue）并运行直到完成。依次调用每个人wg.Done。

一旦等待的 goroutine 通过了wg.Wait，queue 就会关闭并且主线程退出它上面的 range 循环。

Answer 2

queue 通道是无缓冲的，因此每个尝试写入它的 goroutine 都会被阻塞，因为读取器进程尚未启动。所以没有 goroutinte 可以写，他们都挂了 - 结果 wg.Wait 永远等待。 尝试在单独的 goroutine 中启动阅读器：

go func() {
    for a := range queue {
        // block channel until valid url is added to queue
        // once all are added, close it
       activities = append(activities, a)
    }
}()

然后启动服务员：

wg.Wait() 
close(queue)

这样你就可以不用把channel中的所有数据都累积起来并重载它，而是在数据到来时获取数据并放到目标切片中。