爬虫的并发问题

你在学习Golang相关的知识吗？本文《爬虫的并发问题》，主要介绍的内容就涉及到，如果你想提升自己的开发能力，就不要错过这篇文章，大家要知道编程理论基础和实战操作都是不可或缺的哦！

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

1. 有时我会收到 16 个 url 响应，有时会收到 17 个（在 main 中调试打印）。我知道这一点，因为当我什至将 writetoslice 更改为 read 时，然后在 read 中，有时“read: end, counter = ”永远不会达到，而且总是在我获得 16 个网址时出现。

2. 我在使用 err 通道时遇到问题，即使我使用 www.golang.org 这样的地址运行我的主 crawl 方法，也不会在此通道中收到任何消息，因此应通过 err 通道发送没有有效架构的错误

p>

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

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))
}

解决方案

您的问题 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（它在演示中不起作用）

到这里，我们也就讲完了《爬虫的并发问题》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！