go - 搜寻器的并发问题

Question

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

有时我会收到16个网址，有时会得到17个网址(main中的调试打印)。我知道这是因为，当我什至将WriteToSlice更改为Read时，有时在Read中也从未达到“读取:结束，计数器=”，并且总是在获得16个网址时使用。

我在使用err channel 时遇到了麻烦，即使我使用类似于www.golang.org的地址运行我的主要Crawl方法时，在此 channel 中也没有收到任何消息，因此如果没有有效的架构错误，则应该通过err channel 发送

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

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()函数上，释放 WaitGroup ，在WriteToSlice函数以data和err channel 结束之前返回输出并打印输出。所以发生了什么事:

crawl()完成，将数据放入results.data和results.err中。

WaitGroup wait()解锁，导致main()打印结果的长度[]string

WriteToSlice将最后一个数据(或err)项添加到 channel

您不仅需要从 Crawl()返回数据，不仅要在将数据写到 channel 中完成时，还需要在从 channel 整体上读取完(包括缓冲区)时返回。做到这一点的一个好方法是，当您确定已完成 channel 时，请关闭 channel 。通过以这种方式组织代码，您可以阻止正在耗尽 channel 的执行例程，而无需使用 WaitGroup 将其释放给main，而是等到 channel 100％完成。
您可以看到此gobyexample https://gobyexample.com/closing-channels。请记住，关闭 channel 后，直到拍摄完最后一个项目，该 channel 仍可以使用。因此，您可以关闭一个缓冲的 channel ，阅读器仍将获得该 channel 中排队的所有项目。
有一些代码结构可以更改，以使其变得更整洁，但这是修复程序的快速方法。更改 Crawl以阻止 WriteToSlice。抓取功能完成后，关闭数据 channel ，然后等待 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
}

然后在写入 slice 时，您必须检查关闭的 channel 以退出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(在操场上不起作用)