071-并发爬虫

Posted 2022-12-15 --Allen--

学习从来都不是一件困难的事情。那为啥我们学习会如此痛苦？其实难在坚持。这就好比跑步并不是一件困难的事，但是难在十年如一日的坚持。

如果你仔细一点，会发现我的 csdn 头像是《海贼王》这部动漫的主角头像。海贼王这部动漫从 1997 年开始连载，至今已经连载 21 年了。将一部作品连载至今，需要的不仅仅是智慧，更多的是毅力。

如果我们学习也能如此，日复一日，年复一年，我想也应该能有所成就吧。

废话不多说，我们接上一篇的话题，如何控制 goroutine 的并发度呢？总的来说，有两种办法，待会你会看到。

1. 使用 channel 控制并发度

上一篇我们写的代码本身没有错，但错在无穷无尽的并发会对系统造成影响，这会耗尽系统的文件描述符。一个解决办法就是控制最高并发度。

我们的程序从 url “池子”每拿到一个 url 就开启一个 goroutine，如果池子里的 url 数量非常大，一个不小心就能开启上万个 goroutine，我们希望能得到控制。

• 使用 channel 占位控制并发度

假设我们限制 20 并发，怎么做？一个简单的做法就是每开启一个 goroutine 前，就向 channel 里放入一个占位标记，当 goroutine 运行结束后，就把占位标记移除。由于 channel 的缓冲区是固定的，一旦 channel 被占满，就再也无法开启新的 goroutine，除非有旧的 goutine 运行结束，并将 channel 中的标记删除。伪代码如下：

// tokens 是一个大小为 20 的 channel
tokens := make(chan struct, 20)
for 
    tokens <- struct
    go f()
    <-tokens

上面的程序就能控制同时最多 20 个 goroutine 运行。

• 使用固定数量的 long-lived goroutine 控制并发度

另一种控制并发度的方法，是使用 long-lived goroutine，即长时间存活的 goroutine（简称长活协程），这有点像我们以前常说的线程池，在这里你可以说叫协程池。伪代码如下：

tasks := make(chan Type)
for i := 0; i < 20; i++ 
    go func() 
        for task := tasks 
            run(task)
        
    

有经验的同学一看就能知道，这是一个 producter-consumer 模型，即生产者消费者模型。生产者源源不断的将待执行的任务丢入缓冲区 tasks，而消费者（我们开启的 20 个 long-lived goroutine) 源源不断的消费缓冲区的任务。

上面这两种方法各有千秋，下面是具体的程序。

2. 程序

下面的两份代码都在 gopl/goroutine/concurrence 目录下面。

2.1 使用 channel 控制并发

package main

import (
    "fmt"
    "gopl/goroutine/link"
    "log"
    "os"
)

var tokens = make(chan struct, 20)

func crawl(url string) []string 
    fmt.Println(url)
    // 占位
    tokens <- struct
    urls, err := link.ExtractLinks(url)
    // 移除占位标记
    <-tokens
    if err != nil 
        log.Print(fmt.Sprintf("\\x1b[31m%v\\x1b[0m", err))
    
    return urls


func main() 
    if len(os.Args) < 2 
        fmt.Println("Usage:\\n\\tgo run crawl.go <url>")
        os.Exit(1)
    

    workList := make(chan []string)
    seen := make(map[string]bool)

    var n int
    n++
    go func() 
        workList <- os.Args[1:]
    ()

    for ; n > 0; n-- 
        list := <-workList
        for _, url := range list 
            if seen[url] 
                continue
            
            n++
            seen[url] = true
            go func(url string) 
                workList <- crawl(url)
            (url)
        
    

2.2 使用 long-lived goroutine

package main

import (
    "fmt"
    "gopl/goroutine/link"
    "log"
    "os"
)

func crawl(url string) []string 
    fmt.Println(url)
    urls, err := link.ExtractLinks(url)
    if err != nil 
        log.Print(fmt.Sprintf("\\x1b[31m%v\\x1b[0m", err))
    
    return urls


func main() 
    if len(os.Args) < 2 
        fmt.Println("Usage:\\n\\tgo run crawl.go <url>")
        os.Exit(1)
    

    workList := make(chan []string)
    unseenLinks := make(chan string)
    seen := make(map[string]bool)

    var n int
    n++
    go func() 
        workList <- os.Args[1:]
    ()

    // 开启 20 个固定的 long-lived goroutine
    for i := 0; i < 20; i++ 
        go func() 
            for url := range unseenLinks 
                urls := crawl(url)
                go func()  workList <- urls ()
            
        ()
    

    for list := range workList 
        for _, url := range list 
            if seen[url] 
                continue
            
            unseenLinks <- url
        
    

如此一来，你就可以再次运行上面的代码，就不会出现之前的 too many open files 的问题了。运行方法还是和上一篇一样，赶紧试试吧。

3. 总结

• 掌握控制并发度的方法