一、goroutine簡介
Golang被極度贊揚的是它的異步機制,也就是goroutine。goroutine使用方式非常的簡單,只需使用go關鍵字即可啟動一個協程, 並且它是處於異步方式運行,你不需要等它運行完成以后再執行以后的代碼。
go func()//通過go關鍵字啟動一個協程來運行函數
除去語法上的簡潔,goroutine是一個協程,也就是比線程更節省資源,一個線程中可以有多個協程,而且goroutine被分配到多個CPU上運行,是真正意義上的並發。
go func()//通過go關鍵字啟動一個協程來運行函數
二、goroutine內部原理
在介紹goroutine原理之前,先對一些關鍵概念進行介紹:
關鍵概念
並發
一個cpu上能同時執行多項任務,在很短時間內,cpu來回切換任務執行(在某段很短時間內執行程序a,然后又迅速得切換到程序b去執行),有時間上的重疊(宏觀上是同時的,微觀仍是順序執行),這樣看起來多個任務像是同時執行,這就是並發。
並行
當系統有多個CPU時,每個CPU同一時刻都運行任務,互不搶占自己所在的CPU資源,同時進行,稱為並行。
簡單理解
你吃飯吃到一半,電話來了,你一直到吃完了以后才去接,這就說明你不支持並發也不支持並行。
你吃飯吃到一半,電話來了,你停了下來接了電話,接完后繼續吃飯,這說明你支持並發。
你吃飯吃到一半,電話來了,你一邊打電話一邊吃飯,這說明你支持並行。並發的關鍵是你有處理多個任務的能力,不一定要同時。
並行的關鍵是你有同時處理多個任務的能力。在計算機中就是:
所以我認為它們最關鍵的點就是:是否是『同時』。
進程
cpu在切換程序的時候,如果不保存上一個程序的狀態(也就是我們常說的context--上下文),直接切換下一個程序,就會丟失上一個程序的一系列狀態,於是引入了進程這個概念,用以划分好程序運行時所需要的資源。
因此進程就是一個程序運行時候的所需要的基本資源單位(也可以說是程序運行的一個實體)。
線程
cpu切換多個進程的時候,會花費不少的時間,因為切換進程需要切換到內核態,而每次調度需要內核態都需要讀取用戶態的數據,進程一旦多起來,cpu調度會消耗一大堆資源,因此引入了線程的概念,線程本身幾乎不占有資源,他們共享進程里的資源,內核調度起來不會那么像進程切換那么耗費資源。
線程是進程的一個執行實體,是CPU調度和分派的基本單位,它是比進程更小的能獨立運行的基本單位。
NOTE:線程包括三大類,而且goroutine也並非真正地協程。(請查看:《線程那些事兒》)
有時候為了方便理解可以簡單把goroutine類比成協程,但心里一定要有個清晰的認知 — goroutine並不等同於協程。
協程
協程擁有自己的寄存器上下文和棧。協程調度切換時,將寄存器上下文和棧保存到其他地方,在切回來的時候,恢復先前保存的寄存器上下文和棧。因此,協程能保留上一次調用時的狀態(即所有局部狀態的一個特定組合),每次過程重入時,就相當於進入上一次調用的狀態,換種說法:進入上一次離開時所處邏輯流的位置。線程和進程的操作是由程序觸發系統接口,最后的執行者是系統;協程的操作執行者則是用戶自身程序,goroutine也是協程。
G-P-M調度模型簡介
groutine能擁有強大的並發實現是通過GPM調度模型實現,下面就來解釋下goroutine的調度模型。

M:M代表內核級線程,一個M就是一個線程,goroutine就是跑在M之上的;M是一個很大的結構,里面維護小對象內存cache(mcache)、當前執行的goroutine、隨機數發生器等等非常多的信息
G:代表一個goroutine,它有自己的棧,instruction pointer和其他信息(正在等待的channel等等),用於調度。
P:P全稱是Processor,處理器,它的主要用途就是用來執行goroutine的,所以它也維護了一個goroutine隊列,里面存儲了所有需要它來執行的goroutine
NOTE:G-P-M模型詳解,請查看該篇博文。
調度實現

P的數量可以通過GOMAXPROCS()來設置,它其實也就代表了真正的並發度,即有多少個goroutine可以同時運行。
圖中灰色的那些goroutine並沒有運行,而是出於ready的就緒態,正在等待被調度。P維護着這個隊列(稱之為runqueue),
Go語言里,啟動一個goroutine很容易:go function 就行,所以每有一個go語句被執行,runqueue隊列就在其末尾加入一個
goroutine,在下一個調度點,就從runqueue中取出(如何決定取哪個goroutine?)一個goroutine執行。
當一個OS線程M0陷入阻塞時(如下圖),P轉而在運行M1,圖中的M1可能是正被創建,或者從線程緩存中取出。
如果沒有拿到的話,它就把goroutine放在一個global runqueue里,然后自己睡眠(放入線程緩存里)。所有的P也會周期性的檢查global runqueue並運行其中的goroutine,否則global runqueue上的goroutine永遠無法執行。

三、使用goroutine
基本使用
設置goroutine運行的CPU數量,最新版本的go已經默認已經設置了。
num := runtime.NumCPU() //獲取主機的邏輯CPU個數 runtime.GOMAXPROCS(num) //設置可同時執行的最大CPU數
使用示例

package main import ( "fmt" "time" ) func cal(a int , b int ) { c := a+b fmt.Printf("%d + %d = %d\n",a,b,c) } func main() { for i :=0 ; i<10 ;i++{ go cal(i,i+1) //啟動10個goroutine 來計算 } time.Sleep(time.Second * 2) // sleep作用是為了等待所有任務完成 } //結果 //8 + 9 = 17 //9 + 10 = 19 //4 + 5 = 9 //5 + 6 = 11 //0 + 1 = 1 //1 + 2 = 3 //2 + 3 = 5 //3 + 4 = 7 //7 + 8 = 15 //6 + 7 = 13
goroutine異常捕捉
當啟動多個goroutine時,如果其中一個goroutine異常了,並且我們並沒有對進行異常處理,那么整個程序都會終止,所以我們在編寫程序時候最好每個goroutine所運行的函數都做異常處理,異常處理采用recover

package main import ( "fmt" "time" ) func addele(a []int ,i int) { defer func() { //匿名函數捕獲錯誤 err := recover() if err != nil { fmt.Println("add ele fail") } }() a[i]=i fmt.Println(a) } func main() { Arry := make([]int,4) for i :=0 ; i<10 ;i++{ go addele(Arry,i) } time.Sleep(time.Second * 2) } //結果 add ele fail [0 0 0 0] [0 1 0 0] [0 1 2 0] [0 1 2 3] add ele fail add ele fail add ele fail add ele fail add ele fail
同步的goroutine
由於goroutine是異步執行的,那很有可能出現主程序退出時還有goroutine沒有執行完,此時goroutine也會跟着退出。此時如果想等到所有goroutine任務執行完畢才退出,go提供了sync包和channel來解決同步問題,當然如果你能預測每個goroutine執行的時間,你還可以通過time.Sleep方式等待所有的groutine執行完成以后在退出程序(如上面的列子)。

package main import ( "fmt" "sync" ) func cal(a int , b int ,n *sync.WaitGroup) { c := a+b fmt.Printf("%d + %d = %d\n",a,b,c) defer n.Done() //goroutinue完成后, WaitGroup的計數-1 } func main() { var go_sync sync.WaitGroup //聲明一個WaitGroup變量 for i :=0 ; i<10 ;i++{ go_sync.Add(1) // WaitGroup的計數加1 go cal(i,i+1,&go_sync) } go_sync.Wait() //等待所有goroutine執行完畢 } //結果 9 + 10 = 19 2 + 3 = 5 3 + 4 = 7 4 + 5 = 9 5 + 6 = 11 1 + 2 = 3 6 + 7 = 13 7 + 8 = 15 0 + 1 = 1 8 + 9 = 17
示例二:通過channel實現goroutine之間的同步。
實現方式:通過channel能在多個groutine之間通訊,當一個goroutine完成時候向channel發送退出信號,等所有goroutine退出時候,利用for循環channe去channel中的信號,若取不到數據會阻塞原理,等待所有goroutine執行完畢,使用該方法有個前提是你已經知道了你啟動了多少個goroutine。

package main import ( "fmt" "time" ) func cal(a int , b int ,Exitchan chan bool) { c := a+b fmt.Printf("%d + %d = %d\n",a,b,c) time.Sleep(time.Second*2) Exitchan <- true } func main() { Exitchan := make(chan bool,10) //聲明並分配管道內存 for i :=0 ; i<10 ;i++{ go cal(i,i+1,Exitchan) } for j :=0; j<10; j++{ <- Exitchan //取信號數據,如果取不到則會阻塞 } close(Exitchan) // 關閉管道 }
goroutine之間的通訊
goroutine本質上是協程,可以理解為不受內核調度,而受go調度器管理的線程。goroutine之間可以通過channel進行通信或者說是數據共享,當然你也可以使用全局變量來進行數據共享。
示例:使用channel模擬消費者和生產者模式

package main import ( "fmt" "sync" ) func Productor(mychan chan int,data int,wait *sync.WaitGroup) { mychan <- data fmt.Println("product data:",data) wait.Done() } func Consumer(mychan chan int,wait *sync.WaitGroup) { a := <- mychan fmt.Println("consumer data:",a) wait.Done() } func main() { datachan := make(chan int, 100) //通訊數據管道 var wg sync.WaitGroup for i := 0; i < 10; i++ { go Productor(datachan, i,&wg) //生產數據 wg.Add(1) } for j := 0; j < 10; j++ { go Consumer(datachan,&wg) //消費數據 wg.Add(1) } wg.Wait() } //結果 consumer data: 4 product data: 5 product data: 6 product data: 7 product data: 8 product data: 9 consumer data: 1 consumer data: 5 consumer data: 6 consumer data: 7 consumer data: 8 consumer data: 9 product data: 2 consumer data: 2 product data: 3 consumer data: 3 product data: 4 consumer data: 0 product data: 0 product data: 1
四、channel
不同goroutine之間是如何進行通訊的呢?
- 方法一:全局變量和鎖同步
- 方法二:Channel
這里我們主要注重講解下go中特有的channel,其類似於UNIX中的管道(piple)。
channel概念
channel俗稱管道,用於數據傳遞或數據共享,其本質是一個先進先出的隊列,使用goroutine+channel進行數據通訊簡單高效,同時也線程安全,多個goroutine可同時修改一個channel,不需要加鎖。
channel操作
定義和聲明:
1 var 變量名 chan 類型 //channel是有類型的,一個整數的channel只能存放整數 2 3 var test chan int 4 5 var test chan map[string]string 6 7 var test chan *stu
channel可分為三種:
只讀channel:只能讀channel里面數據,不可寫入
只寫channel:只能寫數據,不可讀
一般channel:可讀可寫
var readOnlyChan <-chan int // 只讀chan var writeOnlyChan chan<- int // 只寫chan var mychan chan int //讀寫channel
mychannel = make(chan int,10) //或者 read_only := make (<-chan int,10)//定義只讀的channel write_only := make (chan<- int,10)//定義只寫的channel read_write := make (chan int,10)//可同時讀寫
定義完成以后需要make來分配內存空間,不然會deadlock!

//定義一個結構體類型的channel package main type student struct{ name string } func main() { var stuChan chan student stuChan = make(chan student, 10) stu := student{name:"syu01"} stuChan <- stu }
讀寫數據
ch <- "wd" //寫數據 a := <- ch //讀取數據 a, ok := <-ch //推薦的讀取數據方法
注意:
- 管道如果未關閉,在讀取超時會則會引發deadlock異常
- 管道如果關閉進行寫入數據會pannic
- 當管道中沒有數據時候再行讀取或讀取到默認值,如int類型默認值是0
遍歷管道
- 使用for range遍歷管道,如果管道未關閉會引發deadlock錯誤。
- 如果采用for死循環已經關閉的管道,當管道沒有數據時候,讀取的數據會是管道的默認值,並且循環不會退出。

package main import ( "fmt" "time" ) func main() { mychannel := make(chan int,10) for i := 0;i < 10;i++{ mychannel <- i } close(mychannel) //關閉管道 fmt.Println("data lenght: ",len(mychannel)) for v := range mychannel { //遍歷管道 fmt.Println(v) } fmt.Printf("data lenght: %d",len(mychannel)) }
帶緩沖區channe和不帶緩沖區channel
帶緩沖區channel:定義聲明時候制定了緩沖區大小(長度),可以保存多個數據。
不帶緩沖區channel:只能存一個數據,並且只有當該數據被取出時候才能存下一個數據。
ch := make(chan int) //不帶緩沖區 ch := make(chan int ,10) //帶緩沖區
不帶緩沖區示例:

package main import "fmt" func test(c chan int) { for i := 0; i < 10; i++ { fmt.Println("send ", i) c <- i } } func main() { ch := make(chan int) go test(ch) for j := 0; j < 10; j++ { fmt.Println("get ", <-ch) } } //結果: send 0 send 1 get 0 get 1 send 2 send 3 get 2 get 3 send 4 send 5 get 4 get 5 send 6 send 7 get 6 get 7 send 8 send 9 get 8 get 9
channel實現作業池
我們創建三個channel,一個channel用於接受任務,一個channel用於保持結果,還有個channel用於決定程序退出的時候。

package main import ( "fmt" ) func Task(taskch, resch chan int, exitch chan bool) { defer func() { //異常處理 err := recover() if err != nil { fmt.Println("do task error:", err) return } }() for t := range taskch { // 處理任務 fmt.Println("do task :", t) resch <- t // } exitch <- true //處理完發送退出信號 } func main() { taskch := make(chan int, 20) //任務管道 resch := make(chan int, 20) //結果管道 exitch := make(chan bool, 5) //退出管道 go func() { for i := 0; i < 10; i++ { taskch <- i } close(taskch) }() for i := 0; i < 5; i++ { //啟動5個goroutine做任務 go Task(taskch, resch, exitch) } go func() { //等5個goroutine結束 for i := 0; i < 5; i++ { <-exitch } close(resch) //任務處理完成關閉結果管道,不然range報錯 close(exitch) //關閉退出管道 }() for res := range resch{ //打印結果 fmt.Println("task res:",res) } }
只讀channel和只寫channel
一般定義只讀和只寫的管道意義不大,更多時候我們可以在參數傳遞時候指明管道可讀還是可寫,即使當前管道是可讀寫的。

package main import ( "fmt" "time" ) //只能向chan里寫數據 func send(c chan<- int) { for i := 0; i < 10; i++ { c <- i } } //只能取channel中的數據 func get(c <-chan int) { for i := range c { fmt.Println(i) } } func main() { c := make(chan int) go send(c) go get(c) time.Sleep(time.Second*1) } //結果 0 1 2 3 4 5 6 7 8 9
select-case實現非阻塞channel
原理通過select+case加入一組管道,當滿足(這里說的滿足意思是有數據可讀或者可寫)select中的某個case時候,那么該case返回,若都不滿足case,則走default分支。

package main import ( "fmt" ) func send(c chan int) { for i :=1 ; i<10 ;i++ { c <-i fmt.Println("send data : ",i) } } func main() { resch := make(chan int,20) strch := make(chan string,10) go send(resch) strch <- "wd" select { case a := <-resch: fmt.Println("get data : ", a) case b := <-strch: fmt.Println("get data : ", b) default: fmt.Println("no channel actvie") } } //結果:get data : wd
channel中定時器的使用
在對channel進行讀寫的時,可以對讀寫進行頻率控制,通過time.Ticke實現
示例:

package main import ( "time" "fmt" ) func main(){ requests:= make(chan int ,5) for i:=1;i<5;i++{ requests<-i } close(requests) limiter := time.Tick(time.Second*1) for req:=range requests{ <-limiter fmt.Println("requets",req,time.Now()) //執行到這里,需要隔1秒才繼續往下執行,time.Tick(timer)上面已定義 } } //結果: requets 1 2018-07-06 10:17:35.98056403 +0800 CST m=+1.004248763 requets 2 2018-07-06 10:17:36.978123472 +0800 CST m=+2.001798205 requets 3 2018-07-06 10:17:37.980869517 +0800 CST m=+3.004544250 requets 4 2018-07-06 10:17:38.976868836 +0800 CST m=+4.000533569