Go 語言的并發(fā)
并發(fā)
前言
在學(xué)習(xí) Go 的并發(fā)之前，先復(fù)習(xí)一下操作系統(tǒng)的基礎(chǔ)知識(shí)。
并發(fā)與并行
先來理一理并發(fā)與并行的區(qū)別。
并行：指的是在同一時(shí)間，多個(gè)程序在不同的 CPU 上共同運(yùn)行，互相之間并沒有對(duì) CPU 資源進(jìn)行競(jìng)爭(zhēng)。比如，我在看書的時(shí)候，左手用來翻書，右手做筆記，兩者可以同時(shí)進(jìn)行。
并發(fā)：如果系統(tǒng)只有一個(gè) CPU，有多個(gè)程序要運(yùn)行，系統(tǒng)只能將 CPU 的時(shí)間劃分為多個(gè)時(shí)間片，然后分配給不同的程序。比如，我看書的時(shí)候，只能用右手翻完書之后，才能騰出手來做筆記。
可是明確的是并發(fā)≠并行，但是只要 CPU 運(yùn)行足夠快，每個(gè)時(shí)間片劃分足夠小，就會(huì)給人們?cè)斐梢环N假象，認(rèn)為計(jì)算機(jī)在同一時(shí)刻做了多個(gè)事情。
進(jìn)程、線程、協(xié)程
進(jìn)程是一個(gè)程序執(zhí)行的過程，也是系統(tǒng)進(jìn)行資源分配和調(diào)度的基本單位。簡(jiǎn)單來說，一個(gè)進(jìn)程就是我們電腦上某個(gè)獨(dú)立運(yùn)行的程序。
而線程是系統(tǒng)能夠調(diào)度的最小單位，它被包含在進(jìn)程里面，是進(jìn)程中的實(shí)際運(yùn)作單位，一個(gè)進(jìn)程可以包含多個(gè)線程?？梢詫⑦M(jìn)程理解為一個(gè)工廠，而工廠里面的工人就是線程。就像工廠里面必須要有一個(gè)工人才能工作一樣，每個(gè)進(jìn)程里面也必須有一個(gè)線程才能工作。比如，JavaScript 就被成為單線程的語言，說明 JavaScript 工廠里面只有一個(gè)打工人，這個(gè)打工人就是工頭，稱為主線程。多線程的進(jìn)程中也會(huì)有一個(gè)主線程，主線程一般隨著進(jìn)程一起創(chuàng)建和銷毀。
進(jìn)程與線程都是操作系統(tǒng)上的概念，程序中如果要進(jìn)行進(jìn)程或者線程的切換，在切換的過程中，需要先保存當(dāng)線程的狀態(tài)，然后恢復(fù)另一個(gè)線程的狀態(tài)，這是需要耗費(fèi)時(shí)間的，如果是進(jìn)程的切換還可能跨 CPU，無法利用 CPU 緩存，導(dǎo)致進(jìn)程比線程的切換成本更加高昂。
所以，除了系統(tǒng)級(jí)別的內(nèi)核線程外，一些程序中創(chuàng)建了用戶線程這一說，這么做可以減少與操作系統(tǒng)交互，將線程的切換控制在程序內(nèi)，這種用戶態(tài)的線程被稱為協(xié)程。用戶線程的切換完全由程序控制，實(shí)際上使用的內(nèi)核線程就只存在一個(gè)，內(nèi)核線程與用戶線程之間的關(guān)系為一對(duì)多。雖然這樣做可以減少線程上下文切換帶來的開銷，但是，無法避免阻塞的問題。一旦某個(gè)用戶線程被阻塞會(huì)導(dǎo)致內(nèi)核線程的阻塞，無法進(jìn)行用戶線程進(jìn)行切換，從而整個(gè)進(jìn)程都被掛起，
協(xié)程
Go 語言中的線程模型既不是使用內(nèi)核線程，也不是完全的用戶線程，而是一種混合型的線程模型。用戶線程與內(nèi)核線程的對(duì)應(yīng)關(guān)系為多對(duì)多，用戶線程與內(nèi)核線程動(dòng)態(tài)關(guān)聯(lián)，當(dāng)某個(gè)線程出現(xiàn)阻塞的時(shí)候，可以動(dòng)態(tài)切換到另外的內(nèi)核線程上。
G-P-M模型
上面只是 Go 語言中抽象層面的線程模型，具體是如何進(jìn)行線程調(diào)度的，還是看看 Go 語言的代碼。
func log(msg string) {
fmt.Println(msg)
}
func main() {
log("hello")
go log("world")
}
之前的文章介紹過，Go 程序在運(yùn)行時(shí)，默認(rèn)以 main 函數(shù)為入口，main 函數(shù)中運(yùn)行的代碼會(huì)到一個(gè) goroutine 中運(yùn)行。如果我們?cè)谡{(diào)用的函數(shù)前，加上一個(gè) go 關(guān)鍵詞，那么這個(gè)函數(shù)就放到另外一個(gè) goroutine 中運(yùn)行。
這里說的 goroutine 就是 Go 語言中的用戶線程，也就是協(xié)程。Go 語言在運(yùn)行時(shí)，會(huì)建立一個(gè) G-P-M 模型，這個(gè)模型專門負(fù)責(zé) goroutine 的調(diào)度。
G：gotoutine（用戶線程）； P：processor（邏輯處理器）； M：machine（機(jī)器資源）；
每個(gè) goroutine 都會(huì)放到一個(gè) goroutine 隊(duì)列中，由于是用戶自主創(chuàng)建，上下文的切換成本極低。P（processor）的主要作用是管理用戶線程，將 goroutine 合理的安排到內(nèi)核線程上，也就是這個(gè)模型的 M。通常情況下，G 的數(shù)量遠(yuǎn)遠(yuǎn)多于 M。
Goroutine
如果你有運(yùn)行過上面的代碼，你會(huì)發(fā)現(xiàn)，go 關(guān)鍵詞后的函數(shù)并沒有真正執(zhí)行。
func log(msg string) {
fmt.Println(msg)
}
func main() {
log("hello")
go log("world")
}
運(yùn)行后，終端只輸出了 hello，并沒有輸出 world。
這是因?yàn)?main 函數(shù)會(huì)在主 goroutine 中運(yùn)行，類似于主線程，而每個(gè) go 語句會(huì)啟動(dòng)一個(gè)新的 goroutine，啟動(dòng)后的 goroutine 并不會(huì)直接執(zhí)行，而是會(huì)放入一個(gè) G 隊(duì)列中，等待 P 的分配。但是主 goroutine 結(jié)束后，就意味著程序結(jié)束了，G 隊(duì)列中的 goroutine 還沒有等到執(zhí)行時(shí)間。所以，go 語句后的函數(shù)是一個(gè)異步的函數(shù)，go 語句調(diào)用后，會(huì)立即去執(zhí)行后面的語句，而不會(huì)等待 go 語句后的函數(shù)執(zhí)行。
如果要 world 輸出，我們可以在 main 函數(shù)后面加一個(gè)休眠，延長(zhǎng)主 goroutine 的執(zhí)行時(shí)間。
import (
"fmt"
"time"
)
func log(msg string) {
fmt.Println(msg)
}
func main() {
fmt.Println()
log("hello")
go log("world")
time.Sleep(time.Millisecond * 500)
}
通道
多線程編程中，由于各個(gè)線程之間需要共享數(shù)據(jù)，一般采用的是共享內(nèi)存的方案。但是這么做，勢(shì)必會(huì)出現(xiàn)多個(gè)線程同時(shí)修改同一份數(shù)據(jù)情況，為了保證數(shù)據(jù)的安全性，需要為數(shù)據(jù)加鎖，處理起來就比較麻煩。
所以在 Go 語言社區(qū)有一句名言：
不要通過共享內(nèi)存來通信，而應(yīng)該通過通信來共享內(nèi)存。
創(chuàng)建通道
這里說的通信的方式，就是 Go 語言中的通道（channel）。通道是 Go 語言中的一種特殊類型，需要通過 make 方法創(chuàng)建一個(gè)通道。
ch := make(chan int) // 創(chuàng)建一個(gè) int 類型的通道
創(chuàng)建通道的時(shí)候，需要加上一個(gè)類型，表示該通道傳輸數(shù)據(jù)的類型。也可以通過指定一個(gè)空接口的方式，創(chuàng)建一個(gè)可以傳送任意數(shù)據(jù)的通道。
ch := make(chan interface{})
創(chuàng)建的通道分為無緩存通道和有緩存通道，make 方法的第二個(gè)參數(shù)表示可緩存的數(shù)量（如果傳入 0，效果和不傳一樣）。
ch := make(chan string, 0) // 無緩存通道，傳入
ch := make(chan string, 1)
發(fā)送和接收數(shù)據(jù)
通道創(chuàng)建后，通過 <- 符號(hào)來接收和發(fā)送數(shù)據(jù)。
ch := make(chan string)
ch <- "hello world" // 發(fā)送一個(gè)字符串
msg := <- ch // 接收之前發(fā)送的字符串
實(shí)際在這個(gè)代碼運(yùn)行的時(shí)候，會(huì)提示一個(gè)錯(cuò)誤。
fatal error: all goroutines are asleep - deadlock!
表明當(dāng)前的 goroutine 處于掛起狀態(tài)，并且后續(xù)不會(huì)有響應(yīng)，只能直接中斷程序。因?yàn)檫@里創(chuàng)建的是無緩存通道，發(fā)送數(shù)據(jù)后通道不會(huì)將數(shù)據(jù)緩存在通道中，導(dǎo)致后面要找通道要數(shù)據(jù)的時(shí)候無法正常從通道中獲取數(shù)據(jù)。我們可以將通道的緩存設(shè)置為 1，讓通道可以緩存一個(gè)數(shù)據(jù)在里面。
ch := make(chan string, 1)
ch <- "hello world" // 發(fā)送一個(gè)字符串
msg := <- ch // 接收之前發(fā)送的字符串
fmt.Println(msg)
但是如果發(fā)送的數(shù)據(jù)超出了緩存數(shù)量，或者接受數(shù)據(jù)時(shí)，緩存里面已經(jīng)沒有數(shù)據(jù)了，依然會(huì)報(bào)錯(cuò)。
ch := make(chan string, 1)
ch <- "hello world"
ch <- "hello world"
// fatal error: all goroutines are asleep - deadlock!
ch := make(chan string, 1)
ch <- "hello world"
<- ch
<- ch
// fatal error: all goroutines are asleep - deadlock!
協(xié)程中使用通道
那么無緩存的通道中，應(yīng)該怎么發(fā)送和接收數(shù)據(jù)呢？這就需要將通道與協(xié)程進(jìn)行結(jié)合，也就是 Go 語言中常用的并發(fā)的開發(fā)模式。
無緩存的通道在收發(fā)數(shù)據(jù)時(shí)，由于一次只能同步的發(fā)送一個(gè)數(shù)據(jù)，會(huì)在兩個(gè) goroutine 間反復(fù)橫跳，通道在接受數(shù)據(jù)時(shí)，會(huì)阻塞當(dāng)前 goroutine，直到通道在另一個(gè) goroutine 發(fā)送了數(shù)據(jù)。
ch := make(chan string) // 創(chuàng)建一個(gè)無緩存通道
temp := "我在地球"
go func () {
// 接收一個(gè)字符串
ch <- "hello world"
temp = "進(jìn)入了異次元"
}()
// 運(yùn)行到這里會(huì)被阻塞
// 直到通道在另一個(gè) goroutine 發(fā)送了數(shù)據(jù)
msg := <- ch
fmt.Println(msg)
fmt.Println("temp =>", temp)
為了證明通道在接收數(shù)據(jù)時(shí)會(huì)被阻塞，我們可以在前面加上一個(gè) temp 變量，然后在另外的 goroutine 中修改這個(gè)變量，看最后輸出的值是否被修改，以此證明通道在接受數(shù)據(jù)時(shí)是否發(fā)生了阻塞。
運(yùn)行結(jié)果已經(jīng)證明，當(dāng)通道接收數(shù)據(jù)時(shí)，阻塞了主 goroutine 的執(zhí)行。除了主動(dòng)的從通道里面一條條的獲取數(shù)據(jù)，還可以通過 range 的方式循環(huán)獲取數(shù)據(jù)。
ch := make(chan string)
go func() {
for i := 0; i < 5; i++ {
ch <- fmt.Sprintf("數(shù)據(jù) %d", i)
}
close(ch)
}()
for data := range ch {
fmt.Println("接收 =>", data)
}
如果使用 range 循環(huán)讀取通道中的數(shù)據(jù)時(shí)，在數(shù)據(jù)發(fā)送完畢時(shí)，需要調(diào)用 close(ch) ，將通道關(guān)閉。
實(shí)戰(zhàn)
在了解了前面的基礎(chǔ)知識(shí)之后，我們可以通過協(xié)程 + 通道的寫一段爬蟲，來實(shí)戰(zhàn)一下 Go 語言的并發(fā)能力。
首先確定爬蟲需要爬取的網(wǎng)站，由于個(gè)人比較喜歡看電影，所以決定爬一爬豆瓣的電影 TOP 榜單。
其域名為 https://movie.douban.com/top250，翻到第二頁后，域名為 https://movie.douban.com/top250?start=25 ，第三頁的域名為 https://movie.douban.com/top250?start=50，說明每次這個(gè) TOP 榜單每頁會(huì)有 25 部電影，每次翻頁就給 start 參數(shù)加上 25。
const limit = 25 // 每頁的數(shù)量為 25
const total = 100 // 爬取榜單的前 100 部電影
const page = total / limit // 需要爬取的頁數(shù)
func main() {
var start int
var url string
for i :=0; i < page; i++ {
start := i * limit
// 計(jì)算得到所有的域名
url := "https://movie.douban.com/top250?start=" + strconv.Itoa(start)
}
}
然后，我們可以構(gòu)造一個(gè) fetch 函數(shù)，用于請(qǐng)求對(duì)應(yīng)的頁面。
func fetch(url string) {
// 構(gòu)造請(qǐng)求體
req, _ := http.NewRequest("GET", url, nil)
// 由于豆瓣會(huì)校驗(yàn)請(qǐng)求的 Header
// 如果沒有 User-Agent，http code 會(huì)返回 418
req.Header.Add("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36")
// 發(fā)送請(qǐng)求
client := &http.Client{}
rsp, _ := client.Do(req)
// 斷開連接
defer rsp.Body.Close()
}
func main() {
for i :=0; i < page; i++ {
url := ……
go fetch(url, ch)
}
}
然后使用 goquery 來解析 HTML，提取電影的排名以及電影名。
// 第二個(gè)參數(shù)為與主goroutine 溝通的通道
func fetch(url string, ch chan string) {
// 省略部分代碼 ……
rsp, _ := client.Do(req)
// 斷開連接
defer rsp.Body.Close()
// 解析 HTML
doc, _ := goquery.NewDocumentFromReader(rsp.Body)
// 提取 HTML 中的電影排行與電影名稱
doc.Find(".item").Each(func(_ int, s *goquery.Selection) {
num := s.Find(".pic em").Text()
title := s.Find(".title::first-child").Text()
// 將電影排行與名稱寫入管道中
ch <- fmt.Sprintf("top %s: %s\n", num, title)
})
}
最后，在主 goroutine 中創(chuàng)建通道，以及接收通道中的數(shù)據(jù)。
func main() {
ch := make(chan string)
for i :=0; i < page; i++ {
url := ……
go fetch(url, ch)
}
for i :=0; i < total; i++ {
top := <- ch // 接收數(shù)據(jù)
fmt.Println(top)
}
}
最后的執(zhí)行結(jié)果如下：
可以看到由于是并發(fā)執(zhí)行，輸出的順序是亂序。
完整代碼
package main
import (
"fmt"
"github.com/PuerkitoBio/goquery"
"net/http"
"strconv"
)
const limit = 25
const total = 100
const page = total / limit
func fetch(url string, ch chan string) {
req, _ := http.NewRequest("GET", url, nil)
req.Header.Add("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36")
client := &http.Client{}
rsp, _ := client.Do(req)
defer rsp.Body.Close()
doc, _ := goquery.NewDocumentFromReader(rsp.Body)
doc.Find(".item").Each(func(_ int, s *goquery.Selection) {
num := s.Find(".pic em").Text()
title := s.Find("span.title::first-child").Text()
ch <- fmt.Sprintf("top %s: %s\n", num, title)
})
}
func main() {
ch := make(chan string)
for i :=0; i < page; i++ {
start := i * limit
url := "https://movie.douban.com/top250?start=" + strconv.Itoa(start)
go fetch(url, ch)
}
for i :=0; i < total; i++ {
top := <- ch
fmt.Println(top)
}
}