降低了链接采集数量的并行性：Go Colly技术

大家好，我们又见面了啊~本文《降低了链接采集数量的并行性：Go Colly技术》的内容中将会涉及到等等。如果你正在学习Golang相关知识，欢迎关注我，以后会给大家带来更多Golang相关文章，希望我们能一起进步！下面就开始本文的正式内容~

我正在尝试构建一个网络抓取工具来从 internshala.com 抓取工作。我正在使用 go colly 构建网络抓取器。我访问每个页面，然后访问每个作业的后续链接以从中获取数据。以顺序方式执行此操作会抓取几乎所有链接，但如果我尝试使用 colly 的并行抓取来执行此操作，则抓取的链接数量会减少。我将所有数据写入 csv 文件中。

编辑 我的问题是为什么在并行报废时会发生这种情况以及如何解决它（即使在并行报废时如何才能刮取所有数据）。 或者是我做错了什么导致了这个问题。代码审查将会非常有帮助。谢谢:)

package main

import (
    "encoding/csv"
    "log"
    "os"
    "strconv"
    "sync"
    "time"

    "github.com/gocolly/colly"
)


func main(){

        parallel(10)
        seq(10)
}

由于显而易见的原因，我在运行之前注释掉了这两个函数之一。

并行函数:=

func parallel(n int){
    start := time.now()
    c := colly.newcollector(
        colly.alloweddomains("internshala.com", "https://internshala.com/internship/detail", 
        "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
        colly.async(true),
    )

    d := colly.newcollector(
        colly.alloweddomains("internshala.com", "https://internshala.com/internship/detail", 
        "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
        colly.async(true),
    ) 

    
    c.limit(&colly.limitrule{domainglob: "*", parallelism: 4})
    d.limit(&colly.limitrule{domainglob: "*", parallelism: 4})

    filename := "data.csv"
    file, err := os.create(filename)

    cnt := 0

    if err != nil{
        log.fatalf("could not create file, err: %q", err)
        return
    }

    defer file.close() // close the file after the main routine exits

    writer := csv.newwriter(file)
    defer writer.flush()
    var wg sync.waitgroup
    c.onhtml("a[href]", func(e *colly.htmlelement){

        if e.attr("class") != "view_detail_button"{
            return
        }

        detailslink := e.attr("href")

        d.visit(e.request.absoluteurl(detailslink))
        
    })
    
    d.onhtml(".detail_view", func(e *colly.htmlelement) {
        wg.add(1)       

        go func(wg *sync.waitgroup)  {
            writer.write([]string{
                e.childtext("span.profile_on_detail_page"),
                e.childtext(".company_name a"),
                e.childtext("#location_names a"),
                e.childtext(".internship_other_details_container > div:first-of-type > div:last-of-type .item_body"),
                e.childtext("span.stipend"),
                e.childtext(".applications_message"),
                e.childtext(".internship_details > div:nth-last-of-type(3)"),
                e.request.url.string(), 
            })
            wg.done()
        }(&wg)
        

    })

    c.onrequest(func(r *colly.request) {
        
        log.println("visiting", r.url.string())
    })

    d.onrequest(func(r *colly.request) {
        
        log.println("visiting", r.url.string())
        cnt++
    })

    for i := 1; i < n; i++ {
        c.visit("https://internshala.com/internships/page-"+strconv.itoa(i))
    }

    c.wait()
    d.wait()
    wg.wait()

    t := time.since(start)

    log.printf("time %v \n", t)
    log.printf("amount %v \n", cnt)
    log.printf("scrapping complete")
    log.println(c)

}

seq 函数:=

func seq(n int){
    start := time.Now()
    c := colly.NewCollector(
        colly.AllowedDomains("internshala.com", "https://internshala.com/internship/detail", 
        "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
    )

    d := colly.NewCollector(
        colly.AllowedDomains("internshala.com", "https://internshala.com/internship/detail", 
        "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
    ) 



    fileName := "data.csv"
    file, err := os.Create(fileName)

    cnt := 0

    if err != nil{
        log.Fatalf("Could not create file, err: %q", err)
        return
    }

    defer file.Close() // close the file after the main routine exits

    writer := csv.NewWriter(file)
    defer writer.Flush()

    c.OnHTML("a[href]", func(e *colly.HTMLElement){

        if e.Attr("class") != "view_detail_button"{
            return
        }

        detailsLink := e.Attr("href")

        d.Visit(e.Request.AbsoluteURL(detailsLink))
        
    })
    
    d.OnHTML(".detail_view", func(e *colly.HTMLElement) {
        
        
        writer.Write([]string{
            e.ChildText("span.profile_on_detail_page"),
            e.ChildText(".company_name a"),
            e.ChildText("#location_names a"),
            e.ChildText(".internship_other_details_container > div:first-of-type > div:last-of-type .item_body"),
            e.ChildText("span.stipend"),
            e.ChildText(".applications_message"),
            e.ChildText(".internship_details > div:nth-last-of-type(3)"),
            e.Request.URL.String(), 
        })
        

    })

    c.OnRequest(func(r *colly.Request) {
        
        log.Println("visiting", r.URL.String())
    })

    d.OnRequest(func(r *colly.Request) {
        
        log.Println("visiting", r.URL.String())
        cnt++
    })

    for i := 1; i < n; i++ {
        // Add URLs to the queue
        c.Visit("https://internshala.com/internships/page-"+strconv.Itoa(i))
    }

    t := time.Since(start)

    log.Printf("time %v \n", t)
    log.Printf("amount %v \n", cnt)
    log.Printf("Scrapping complete")
    log.Println(c)

}

任何帮助将不胜感激。 :)

正确答案

很抱歉在聚会上迟到，但我想出了一个有效的解决方案来解决您的问题。让我展示一下：

package main

import (
    "encoding/csv"
    "fmt"
    "log"
    "os"
    "strconv"
    "strings"
    "time"

    "github.com/gocolly/colly/v2"
    "github.com/gocolly/colly/v2/queue"
)

func parallel(n int) {
    start := time.Now()
    cnt := 0
    queue, _ := queue.New(8, &queue.InMemoryQueueStorage{MaxSize: 1000}) // tried up to 8 threads

    fileName := "data_par.csv"
    file, err := os.Create(fileName)
    if err != nil {
        log.Fatalf("Could not create file, err: %q", err)
        return
    }

    defer file.Close() // close the file after the main routine exits

    writer := csv.NewWriter(file)
    defer func() {
        writer.Flush()
        if err := writer.Error(); err != nil {
            panic(err)
        }
    }()

    c := colly.NewCollector(
        colly.AllowedDomains("internshala.com", "https://internshala.com/internship/detail",
            "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
    )

    c.OnHTML("a[href]", func(e *colly.HTMLElement) {
        if e.Attr("class") != "view_detail_button" {
            return
        }

        detailsLink := e.Attr("href")
        e.Request.Visit(detailsLink)
    })

    c.OnRequest(func(r *colly.Request) {
        writer.Write([]string{r.URL.String()})
    })

    for i := 1; i < n; i++ {
        queue.AddURL("https://internshala.com/internships/page-" + strconv.Itoa(i))
    }

    queue.Run(c)

    t := time.Since(start)
    log.Printf("time: %v\tamount: %d\n", t, cnt)
}

func seq(n int) {
    start := time.Now()
    c := colly.NewCollector(
        colly.AllowedDomains("internshala.com", "https://internshala.com/internship/detail",
            "https://internshala.com/internship/", "internshala.com/", "www.intershala.com"),
    )

    fileName := "data_seq.csv"
    file, err := os.Create(fileName)

    cnt := 0

    if err != nil {
        log.Fatalf("Could not create file, err: %q", err)
        return
    }

    defer file.Close() // close the file after the main routine exits

    writer := csv.NewWriter(file)
    defer func() {
        writer.Flush()
        if err := writer.Error(); err != nil {
            panic(err)
        }
    }()

    c.OnHTML("a[href]", func(e *colly.HTMLElement) {
        if e.Attr("class") != "view_detail_button" {
            return
        }

        detailsLink := e.Attr("href")
        e.Request.Visit(detailsLink)
    })

    c.OnRequest(func(r *colly.Request) {
        writer.Write([]string{r.URL.String()})
    })

    for i := 1; i < n; i++ {
        c.Visit("https://internshala.com/internships/page-" + strconv.Itoa(i))
    }

    t := time.Since(start)
    log.Printf("time: %v\tamount: %d\n", t, cnt)
}

func main() {
    fmt.Println("sequential")
    seq(6)
    fmt.Println(strings.Repeat("#", 50))
    fmt.Println("parallel")
    parallel(6)
}

问题

查看您的代码后，我认为一切都已正确实现。当然，事情可以以更好的方式完成，但至少在并发性方面，一切都设置正确。以下列表中列出了您可以改进的一些方面：

1. 刷新到底层 csv 文件时检查 error
2. 仅使用一个收集器而不是两个

同样，正如我已经说过的，这些只是小的改进。

实际问题

实际问题是，当您发出并发（并且可能是并行）请求时，colly 框架无法跟上它并开始丢失一些响应。当您增加执行次数时，这种趋势会呈指数级增长。

最简单的解决方案（imo）

gocolly 提供了非常适合这些挑战的 queue 类型。感谢他们，您将确保每个请求都会得到处理，就好像它们是同时完成的一样。步骤可概括如下：

1. 使用 queue 子包提供的 new 函数实例化一个新队列。您必须设置线程数量以及队列类型（在我们的例子中，可以使用内存中实现）。
2. 使用所有需要的回调实例化默认收集器。
3. 使用要查询的适当 url 对上述定义的 queue 变量调用 addurl 方法。
4. 调用 run 方法，将实际请求发送到目标网址并等待响应。

请告诉我这是否可以解决您的问题或分享您如何解决此问题，谢谢！

终于介绍完啦！小伙伴们，这篇关于《降低了链接采集数量的并行性：Go Colly技术》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！