几分钟后，Go 爬虫在从输出 channel 中选择时停止

Question

我写了一个简单的爬虫，看起来像这样:

type SiteData struct {
    // ...
}

func downloadURL(url string) (body []byte, status int) {
    resp, err := http.Get(url)

    if err != nil {
        return
    }

    status = resp.StatusCode
    defer resp.Body.Close()

    body, err = ioutil.ReadAll(resp.Body)
    body = bytes.Trim(body, "\x00")

    return
}


func processSiteData(resp []byte) SiteData {
    // ...
}    

func worker(input chan string, output chan SiteData) {

    // wait on the channel for links to process
    for url := range input {

        // fetch the http response and status code
        resp, status := downloadURL(url)

        if resp != nil && status == 200 {
            // if no errors in fetching link
            // process the data and send 
            // it back
            output <- processSiteData(resp)
        } else {
            // otherwise send the url for processing
            // once more
            input <- url
        }
    }
}

func crawl(urlList []string) {
    numWorkers := 4
    input := make(chan string)
    output := make(chan SiteData)

    // spawn workers
    for i := 0; i < numWorkers; i++ {
        go worker(input, output)
    }

    // enqueue urls
    go func() {
        for url := range urlList {
            input <- url
        }
    }()

    // wait for the results
    for {
        select {
        case data := <-output:
            saveToDB(data)
        }
    }

}

func main() {
    urlList := loadLinksFromDB()
    crawl(urlList)
}

它抓取单个网站并且运行良好 - 下载数据、处理数据并将其保存到数据库中。然而，几分钟后 (5-10) 左右，它会“卡住”并需要重新启动。该网站没有将我列入黑名单，我已经与他们核实过，并且可以在程序阻止后随时访问任何 url。此外，它会在所有 url 处理完成之前 阻止。显然，当列表用完时它会阻塞，但它远非如此。

我是不是做错了什么？我使用 for { select { ... } } 而不是 for _, _ = range urlList {//read output } 的原因是任何 url 都可以是如果未能处理，则重新排队。此外，数据库似乎也不是这里的问题。任何输入都会有所帮助 - 谢谢。

Answer 1

我相信当所有 N 个工作人员都在等待 input <- url 时，这会挂起，因此不再有 worker 从 input 中取出东西.换句话说，如果 4 个 URL 大致同时失败，它就会挂起。

解决方案是将失败的 URL 发送到某个不是工作人员输入 channel 的地方(以避免死锁)。

一种可能性是有一个单独的failed channel，匿名 goroutine 总是接受来自它的输入。像这样(未测试):

package main

func worker(intput chan string, output chan SiteData, failed chan string) {
    for url := range input {
        // ...
        if resp != nil && status == 200 {
            output <- processSideData(resp)
        } else {
            failed <- url
        }
    }
}

func crawl(urlList []string) {
    numWorkers := 4
    input := make(chan string)
    failed := make(chan string)
    output := make(chan SiteData)

    // spawn workers
    for i := 0; i < numWorkers; i++ {
        go worker(input, output, failed)
    }

    // Dispatch URLs to the workers, also receive failures from them.
    go func() {
        for {
            select {
            case input <- urlList[0]:
                urlList = urlList[1:]
            case url := <-failed:
                urlList = append(urlList, url)
            }
        }
    }()

    // wait for the results
    for {
        data := <-output
        saveToDB(data)
    }
}

func main() {
    urlList := loadLinksFromDB()
    crawl(urlList)
}

(请注意，正如您在评论中所说，不要在 for _, _ = range urlList { // read output } 函数中使用 crawl() 是正确的，因为 URL 可以重新排队；但据我所知，您不需要选择任何一个可以说。)