爬虫的并发问题

Question

我尝试基于 Tour 和其他一些 SO 答案构建并发爬虫。我目前的情况如下，但我认为这里有两个微妙的问题。

1. 有时我得到 16 个 url 作为响应，有时是 17 个（main 中的调试打印）。我知道这一点，因为当我什至将 WriteToSlice 更改为 Read 时，在 Read 中有时“读取：结束，计数器 =”永远不会到达，而且总是在我得到 16 个网址时。

2. 我在使用err 频道时遇到问题，我在此频道中没有收到任何消息，即使我使用www.golang.org 之类的地址运行我的主要Crawl 方法，因此应该通过err 频道发送没有有效架构错误的情况

并发是一个非常困难的话题，我们将不胜感激帮助和建议

package main

import (
    "fmt"
    "net/http"
    "sync"

    "golang.org/x/net/html"
)

type urlCache struct {
    urls map[string]struct{}
    sync.Mutex
}

func (v *urlCache) Set(url string) bool {
    v.Lock()
    defer v.Unlock()

    _, exist := v.urls[url]
    v.urls[url] = struct{}{}

    return !exist
}

func newURLCache() *urlCache {
    return &urlCache{
        urls: make(map[string]struct{}),
    }
}

type results struct {
    data chan string
    err  chan error
}

func newResults() *results {
    return &results{
        data: make(chan string, 1),
        err:  make(chan error, 1),
    }
}

func (r *results) close() {
    close(r.data)
    close(r.err)
}

func (r *results) WriteToSlice(s *[]string) {
    for {
        select {
        case data := <-r.data:
            *s = append(*s, data)
        case err := <-r.err:
            fmt.Println("e ", err)
        }
    }
}

func (r *results) Read() {
    fmt.Println("Read: start")
    counter := 0
    for c := range r.data {
        fmt.Println(c)
        counter++
    }
    fmt.Println("Read: end, counter = ", counter)
}

func crawl(url string, depth int, wg *sync.WaitGroup, cache *urlCache, res *results) {
    defer wg.Done()

    if depth == 0 || !cache.Set(url) {
        return
    }

    response, err := http.Get(url)
    if err != nil {
        res.err <- err
        return
    }
    defer response.Body.Close()

    node, err := html.Parse(response.Body)
    if err != nil {
        res.err <- err
        return
    }

    urls := grablUrls(response, node)

    res.data <- url

    for _, url := range urls {
        wg.Add(1)
        go crawl(url, depth-1, wg, cache, res)
    }
}

func grablUrls(resp *http.Response, node *html.Node) []string {
    var f func(*html.Node) []string
    var results []string

    f = func(n *html.Node) []string {
        if n.Type == html.ElementNode && n.Data == "a" {
            for _, a := range n.Attr {
                if a.Key != "href" {
                    continue
                }
                link, err := resp.Request.URL.Parse(a.Val)
                if err != nil {
                    continue
                }
                results = append(results, link.String())
            }
        }

        for c := n.FirstChild; c != nil; c = c.NextSibling {
            f(c)
        }

        return results
    }

    res := f(node)
    return res
}

// Crawl ...
func Crawl(url string, depth int) []string {
    wg := &sync.WaitGroup{}
    output := &[]string{}
    visited := newURLCache()
    results := newResults()
    defer results.close()

    wg.Add(1)
    go crawl(url, depth, wg, visited, results)
    go results.WriteToSlice(output)
    // go results.Read()
    wg.Wait()

    return *output
}

func main() {
    r := Crawl("https://www.golang.org", 2)
    // r := Crawl("www.golang.org", 2) // no schema, error should be generated and send via err

    fmt.Println(len(r))
}

Answer 1

您的问题 1 和 2 都是同一个错误的结果。

在 Crawl() 中，您不会等待这个 go 例程完成：go results.WriteToSlice(output)。在最后一个crawl() 函数上，等待组被释放，在WriteToSlice 函数完成data 和err 通道之前返回并打印输出。所以发生的事情是这样的：

1. crawl() 完成，将数据放入 results.data 和 results.err。
2. Waitgroup wait() 解除阻塞，导致main() 打印结果[]string 的长度
3. WriteToSlice 将最后一个数据（或错误）项添加到通道中

您需要从Crawl() 返回，不仅要在数据写入通道完成时，还要在通道全部读取完成时（包括缓冲区）。一个很好的方法是当你确定你已经完成了它们时关闭它们。通过以这种方式组织代码，您可以阻止正在耗尽通道的 go 例程，而不是使用等待组释放到主通道，而是等待通道 100% 完成。

你可以看到这个 gobyexample https://gobyexample.com/closing-channels。请记住，当您关闭一个频道时，该频道仍然可以使用，直到最后一个项目被取走。所以你可以关闭一个缓冲的通道，阅读器仍然会得到所有在通道中排队的项目。

有一些代码结构可以更改以使其更清晰，但这里有一个快速修复程序的方法。将Crawl 更改为阻止WriteToSlice。爬取完成后关闭数据通道，等待WriteToSlice完成。

// Crawl ...
func Crawl(url string, depth int) []string {
    wg := &sync.WaitGroup{}
    output := &[]string{}
    visited := newURLCache()
    results := newResults()

    go func() {
        wg.Add(1)
        go crawl(url, depth, wg, visited, results)
        wg.Wait()
        // All data is written, this makes `WriteToSlice()` unblock
        close(results.data)
    }()

    // This will block until results.data is closed
    results.WriteToSlice(output)
    close(results.err)

    return *output
}

然后在写入切片时，您必须检查关闭的通道以退出 for 循环：

func (r *results) WriteToSlice(s *[]string) {
    for {
        select {
        case data, open := <-r.data:
            if !open {
                return // All data done
            }
            *s = append(*s, data)
        case err := <-r.err:
            fmt.Println("e ", err)
        }
    }
}

这里是完整代码：https://play.golang.org/p/GBpGk-lzrhd（在操场上不起作用）