Golang多線程爬蟲高效抓取大量數(shù)據(jù)的利器

更新時間：2023年05月09日 10:28:37 作者：Luyoungs

Golang多線程爬蟲是一種高效抓取大量數(shù)據(jù)的利器。Golang語言天生支持并發(fā)和多線程，可以輕松實現(xiàn)多線程爬蟲的開發(fā)。通過使用Golang的協(xié)程和通道，可以實現(xiàn)爬蟲的高效并發(fā)抓取、數(shù)據(jù)處理和存儲

前言

Golang 是一種并發(fā)友好的語言，使用 goroutines 和 channels 可以輕松地實現(xiàn)多線程爬蟲。具體地說，實現(xiàn)的是多協(xié)程。協(xié)程是一種比線程更輕量化的最小邏輯可運行單位，它不受操作系統(tǒng)調(diào)度，由用戶調(diào)度。因此對于協(xié)程并發(fā)的控制，有較高的要求。

goroutine(Go 協(xié)程)

Go 協(xié)程（Goroutine）是與其他函數(shù)同時運行的函數(shù)?？梢哉J(rèn)為 Go 協(xié)程是輕量級的線程，由 Go 運行時來管理。在函數(shù)調(diào)用前加上 go 關(guān)鍵字，這次調(diào)用就會在一個新的 goroutine 中并發(fā)執(zhí)行。當(dāng)被調(diào)用的函數(shù)返回時，這個 goroutine 也自動結(jié)束。

比如：

func son() {
	for {
		fmt.Printf("son says:hello, world!\n")
		time.Sleep(time.Second)
	}
}
func father() {
	go son()
	for {
		fmt.Printf("father says:你好，世界!\n")
		time.Sleep(time.Second)
	}
}
func main() {
	father()
}

運行結(jié)果：

father says:你好，世界!

son says:hello, world!

son says:hello, world!

father says:你好，世界!

father says:你好，世界!

son says:hello, world!

…

在這個例子中，main() 函數(shù)里面執(zhí)行 father()，而 father()中又開啟了一個協(xié)程 son()，之后兩個死循環(huán)分別執(zhí)行。

當(dāng)然，如果主協(xié)程運行結(jié)束時子協(xié)程還沒結(jié)束，那么就會被 kill 掉。需要注意的是，如果這個函數(shù)有返回值，那么這個返回值會被丟棄。

func main() {
	go loop()
	fmt.Println("hello,world!")
}
func loop() {
	for i := 0; i < 10000; i++ {
		fmt.Println(i)
	}
}

運行結(jié)果：

hello,world!

0

1

可以看到，子協(xié)程剛打印了 0、1 之后 main()函數(shù)就結(jié)束了，這顯然不符合我們的預(yù)期。我們有多種方法來解決這個問題。比如在主協(xié)程里面 sleep()，或者使用 channel、waitGroup 等。

Go 協(xié)程（Goroutine）之間通過信道（channel）進行通信，簡單的說就是多個協(xié)程之間通信的管道。信道可以防止多個協(xié)程訪問共享內(nèi)存時發(fā)生資源爭搶的問題。Go 中的 channel 是 goroutine 之間的通信機制。這就是為什么我們之前說過 Go 實現(xiàn)并發(fā)的方式是：“不是通過共享內(nèi)存通信，而是通過通信共享內(nèi)存。”

比如：

var (
	myMap = make(map[int]int, 10)
)
// 計算n！并放入到map里
func operation(n int) {
	res := 1
	for i := 1; i <= n; i++ {
		res *= i
	}
	myMap[n] = res
}
func Test3() {
	//我們開啟多個協(xié)程去完成這個任務(wù)
	for i := 1; i <= 200; i++ {
		go operation(i)
	}
	time.Sleep(time.Second * 10)
	fmt.Println(myMap)
}
func main() {
	Test3()
}

運行結(jié)果：

fatal error: concurrent map writes

goroutine 42 [running]:

這里產(chǎn)生了一個 fatal error，因為我們創(chuàng)建的 myMap 不支持同時訪問，這有點像 Java 里面的非線程安全概念。因此我們需要一種“線程安全”的數(shù)據(jù)結(jié)構(gòu)，就是 Go 中的channel。

channel(通道)

channel 分為無緩沖和有緩沖的。無緩沖是同步的，例如make(chan int)，就是一個送信人去你家門口送信，你不在家他不走，你一定要接下信，他才會走，無緩沖保證信能到你手上。有緩沖是異步的，例如make(chan int, 1)，就是一個送信人去你家仍到你家的信箱，轉(zhuǎn)身就走，除非你的信箱滿了，他必須等信箱空下來，有緩沖的保證信能進你家的郵箱。

換句話說，有緩存的channel使用環(huán)形數(shù)組實現(xiàn)，當(dāng)緩存未滿時，向channel發(fā)送消息不會阻塞，當(dāng)緩存滿時，發(fā)送操作會阻塞，直到其他goroutine從channel中讀取消息；同理，當(dāng)channel中消息不為空時，讀取消息不會阻塞，當(dāng)channel為空時，讀取操作會阻塞，直至其他goroutine向channel發(fā)送消息。

// 非緩存channel
ch := make(chan int)
// 緩存channel
bch := make(chan int, 2)

channel和map類似，make創(chuàng)建了底層數(shù)據(jù)結(jié)構(gòu)的引用，當(dāng)賦值或參數(shù)傳遞時，只是拷貝了一個channel的引用，其指向同一channel對象，與其引用類型一樣，channel的空值也為nil。使用==可以對類型相同的channel進行比較，只有指向相同對象或同為nil時，結(jié)果為true。

channel 的初始化

channel在使用前，需要初始化，否則永遠阻塞。

ch := make(chan int)
ch <- x
y <- ch

channel的關(guān)閉

golang提供了內(nèi)置的close函數(shù)，對channel進行關(guān)閉操作。

// 初始化channel
ch := make(chan int)
// 關(guān)閉channel ch
close(ch)

關(guān)于channel的關(guān)閉，需要注意以下事項：

關(guān)閉未初始化的channle(nil)會panic
重復(fù)關(guān)閉同一channel會panic
向以關(guān)閉channel發(fā)送消息會panic
從已關(guān)閉channel讀取數(shù)據(jù)，不會panic，若存在數(shù)據(jù)，則可以讀出未被讀取的消息，若已被讀出，則獲取的數(shù)據(jù)為零值，可以通過ok-idiom的方式，判斷channel是否關(guān)閉
channel的關(guān)閉操作，會產(chǎn)生廣播消息，所有向channel讀取消息的goroutine都會接受到消息

waitGroup 的使用

正常情況下，新激活的goroutine的結(jié)束過程是不可控制的，唯一可以保證終止goroutine的行為是main goroutine的終止。

也就是說，我們并不知道哪個goroutine什么時候結(jié)束。

但很多情況下，我們正需要知道goroutine是否完成。這需要借助sync包的WaitGroup來實現(xiàn)。

WatiGroup是sync包中的一個struct類型，用來收集需要等待執(zhí)行完成的goroutine。下面是它的定義：

type WaitGroup struct {
        // Has unexported fields.
}
    A WaitGroup waits for a collection of goroutines to finish. The main
    goroutine calls Add to set the number of goroutines to wait for. Then each
    of the goroutines runs and calls Done when finished. At the same time, Wait
    can be used to block until all goroutines have finished.
A WaitGroup must not be copied after first use.
func (wg *WaitGroup) Add(delta int)
func (wg *WaitGroup) Done()
func (wg *WaitGroup) Wait()

waitGroup有三個方法：

Add()：每次激活想要被等待完成的goroutine之前，先調(diào)用Add()，用來設(shè)置或添加要等待完成的goroutine數(shù)量。例如Add(2)或者兩次調(diào)用Add(1)都會設(shè)置等待計數(shù)器的值為2，表示要等待2個goroutine完成
Done()：每次需要等待的goroutine在真正完成之前，應(yīng)該調(diào)用該方法來人為表示goroutine完成了，該方法會對等待計數(shù)器減1。
Wait()：在等待計數(shù)器減為0之前，Wait()會一直阻塞當(dāng)前的goroutine也就是說，Add()用來增加要等待的goroutine的數(shù)量，Done()用來表示goroutine已經(jīng)完成了，減少一次計數(shù)器，Wait()用來等待所有需要等待的goroutine完成。

比如：

var wg sync.WaitGroup // 創(chuàng)建同步等待組對象
func main() {
	//設(shè)置等待組中，要執(zhí)行的goroutine的數(shù)量
	wg.Add(2)
	go fun1()
	go fun2()
	fmt.Println("main進入阻塞狀態(tài),等待wg中的子goroutine結(jié)束")
	wg.Wait() //表示main goroutine進入等待，意味著阻塞
	fmt.Println("main解除阻塞")
}
func fun1() {
	for i := 1; i <= 10; i++ {
		fmt.Println("fun1.i:", i)
	}
	wg.Done() //給wg等待中的執(zhí)行的goroutine數(shù)量減1.同Add(-1)
}
func fun2() {
	defer wg.Done()
	for j := 1; j <= 10; j++ {
		fmt.Println("\tfun2.j,", j)
	}
}

運行結(jié)果：

main進入阻塞狀態(tài),等待wg中的子goroutine結(jié)束 fun1.i: 1

fun2.j, 1

fun2.j, 2

fun2.j, 3

fun2.j, 4

fun2.j, 5

fun1.i: 2

fun1.i: 3

fun1.i: 4

…

main解除阻塞

可以看到起到了很好的控制效果。

如果用第一個例子來說明，效果更好：

func main() {
	var wg sync.WaitGroup // 創(chuàng)建同步等待組對象
	wg.Add(1)
	go loop(&wg)
	wg.Wait()
	fmt.Println("hello,world!")
}
func loop(wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < 100; i++ {
		fmt.Println(i)
	}
}

運行結(jié)果：

0

1

…

99

hello,world!

爬蟲

爬蟲的功能是爬取豆瓣top250的電影的數(shù)據(jù)，并將爬到的數(shù)據(jù)永久花存儲。

思路是首先爬取所有的鏈接，這個鏈接的提取通過10 個并行的goroutine處理，然后存儲到 channel 中。然后立即創(chuàng)建 250個 goroutine，每一個協(xié)程分別爬取一個鏈接。再將爬到的數(shù)據(jù)存儲到本地。

爬蟲配置

type SpiderConfig struct {
	InitialURL   string // 初始 URL
	MaxDepth     int    // 最大深度
	MaxGoroutine int    // 最大并發(fā)數(shù)
}

爬蟲數(shù)據(jù)

type SpiderData struct {
	URL       string // 鏈接
	FilmName  string // 電影名
	Director  string // 導(dǎo)員
	Actors    Actor  // 演員列表
	Year      string // 年份
	Score     string // 評分
	Introduce string // 簡介
}
type Actor struct {
	actor1 string
	actor2 string
	actor3 string
	actor4 string
	actor5 string
	actor6 string
}

開啟并行

func spider(config SpiderConfig, chLinks chan string, wg *sync.WaitGroup) {
	for i := 0; i < 10; i++ {
		fmt.Println("正在爬取第", i, "個信息")
		go Spider(strconv.Itoa(i*25), chLinks, wg)
	}
}

爬取某個鏈接

func Spider(page string, chLinks chan string, wg *sync.WaitGroup) {
	// client
	client := http.Client{}
	URL := "https://movie.douban.com/top250?start=" + page + "&filter="
	req, err := http.NewRequest("GET", URL, nil)
	if err != nil {
		fmt.Println("req err", err)
	}
	// UA偽造，明顯比 python復(fù)雜得多
	req.Header.Set("Connection", "keep-alive")
	req.Header.Set("Cache-Control", "max-age=0")
	req.Header.Set("sec-ch-ua-mobile", "?0")
	req.Header.Set("Upgrade-Insecure-Requests", "1")
	req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36")
	req.Header.Set("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9")
	req.Header.Set("Sec-Fetch-Site", "same-origin")
	req.Header.Set("Sec-Fetch-Mode", "navigate")
	req.Header.Set("Sec-Fetch-User", "?1")
	req.Header.Set("Sec-Fetch-Dest", "document")
	req.Header.Set("Referer", "https://movie.douban.com/chart")
	req.Header.Set("Accept-Language", "zh-CN,zh;q=0.9")
	resp, err := client.Do(req)
	if err != nil {
		fmt.Println("resp err", err)
	}
	defer func(Body io.ReadCloser) {
		err := Body.Close()
		if err != nil {
			fmt.Println("close err", err)
		}
	}(resp.Body)
	// 網(wǎng)頁解析
	docDetail, err := goquery.NewDocumentFromReader(resp.Body)
	if err != nil {
		fmt.Println("解析失?。?, err)
	}
	// 選擇器
	// #content > div > div.article > ol > li:nth-child(1) > div > div.info > div.hd > a
	title := docDetail.Find("#content > div > div.article > ol > li").
		Each(func(i int, s *goquery.Selection) { // 繼續(xù)找
			link := s.Find("div > div.pic > a")
			linkTemp, OK := link.Attr("href")
			if OK {
				chLinks <- linkTemp
			}
		})
	title.Text()
	wg.Done()
}

爬取某個鏈接的電影數(shù)據(jù)

func crawl(url string, wg *sync.WaitGroup) {
	client := http.Client{}
	URL := url
	req, err := http.NewRequest("GET", URL, nil)
	if err != nil {
		fmt.Println("req err", err)
	}
	// UA偽造，明顯比 python復(fù)雜得多
	req.Header.Set("Connection", "keep-alive")
	req.Header.Set("Cache-Control", "max-age=0")
	req.Header.Set("sec-ch-ua-mobile", "?0")
	req.Header.Set("Upgrade-Insecure-Requests", "1")
	req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36")
	req.Header.Set("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9")
	req.Header.Set("Sec-Fetch-Site", "same-origin")
	req.Header.Set("Sec-Fetch-Mode", "navigate")
	req.Header.Set("Sec-Fetch-User", "?1")
	req.Header.Set("Sec-Fetch-Dest", "document")
	req.Header.Set("Referer", "https://movie.douban.com/chart")
	req.Header.Set("Accept-Language", "zh-CN,zh;q=0.9")
	resp, err := client.Do(req)
	if err != nil {
		fmt.Println("resp err", err)
	}
	defer func(Body io.ReadCloser) {
		err := Body.Close()
		if err != nil {
		}
	}(resp.Body)
	// 網(wǎng)頁解析
	docDatail, err := goquery.NewDocumentFromReader(resp.Body)
	if err != nil {
		fmt.Println("解析失??！", err)
	}
	var data SpiderData
	// 選擇器
	// #content > h1 > span:nth-child(1)	movie_name
	movie_name := docDatail.Find("#content > h1 > span:nth-child(1)").Text()
	// #info > span:nth-child(1) > span.attrs > a	director
	director := docDatail.Find("#info > span:nth-child(1) > span.attrs > a").Text()
	// #info > span.actor > span.attrs > span:nth-child(1) > a
	actor01, OK1 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(1) > a").Attr("hel")
	if OK1 {
	}
	actor02 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(2) > a").Text()
	actor03 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(3) > a").Text()
	actor04 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(4) > a").Text()
	actor05 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(5) > a").Text()
	actor06 := docDatail.Find("#info > span.actor > span.attrs > span:nth-child(6) > a").Text()
	// #content > h1 > span.year
	year := docDatail.Find("#content > h1 > span.year").Text()
	// #interest_sectl > div.rating_wrap.clearbox > div.rating_self.clearfix > strong
	score := docDatail.Find("#interest_sectl > div.rating_wrap.clearbox > div.rating_self.clearfix > strong").Text()
	//#link-report-intra > span.all.hidden
	introduce := docDatail.Find("#link-report-intra > span.all.hidden").Text()
	data.URL = URL
	data.FilmName = movie_name
	data.Director = director
	data.Actors.actor1 = actor01
	data.Actors.actor2 = actor02
	data.Actors.actor3 = actor03
	data.Actors.actor4 = actor04
	data.Actors.actor5 = actor05
	data.Actors.actor6 = actor06
	data.Year = year
	data.Score = score
	data.Introduce = introduce
	result := data2string(data)
	filename := strconv.Itoa(rand.Int()) + ".txt"
	f, err := os.Create(filename)
	if err != nil {
		fmt.Println(err)
	}
	_, err = f.Write([]byte(result))
	if err != nil {
		return
	}
	err = f.Close()
	if err != nil {
		return
	}
	defer wg.Done()
}
func data2string(data SpiderData) string {
	result := data.FilmName + data.Score + data.Director + data.Year + data.Introduce
	return result
}

main 函數(shù)開啟爬蟲

func main() {
	// 先爬取初始URL,將爬來的 links放在 chan中
	// 定義一個 chLinks,這里面將會放置 250 個links
	chLinks := make(chan string, 1000) // 有緩沖的 chan
	config := SpiderConfig{
		InitialURL:   "https://movie.douban.com/top250",
		MaxDepth:     1,
		MaxGoroutine: 10,
	}
	wg := sync.WaitGroup{}
	wg.Add(10)
	spider(config, chLinks, &wg) // 主線程,并發(fā)爬取所有的 href
	wg.Wait()
	//for i := 0; i < 250; i++ {
	//	fmt.Println(i, <-chLinks)
	//}
	// 爬完了所有的鏈接,250 個鏈接放在 chLinks
	//建立250 個協(xié)程來爬每個鏈接
	wg.Add(250)
	for i := 0; i < 250; i++ {
		go crawl(<-chLinks, &wg)
	}
	wg.Wait()
}

爬取結(jié)果：