Go 深入多線程

 

Golang

介紹

線程是cpu調度的最小單位，只有不同的線程才能同時在多核cpu上同時運行。但線程太占資源，線程調度開銷大。go中的goroutine是一個輕量級的線程，執行時只需要4-5k的內存，比線程更易用，更高效，更輕便，調度開銷比線程小，可同時運行上千萬個並發。
go語言中開啟一個goroutine非常簡單，go函數名（），就開啟了個線程。

默認情況下，調度器僅使用單線程，要想發揮多核處理器的並行處理能力，必須調用runtine.GOMAXPROCS(n)來設置可並發的線程數，也可以通過設置環境變量GOMAXPROCS打到相同的目的。

goroutine

Runtime包中提供了幾個與goroutine相關的函數。Gosched()讓當前正在執行的goroutine放棄CPU執行權限。調度器安排其他正在等待的線程運行。
請看以下例子：

package main import ( "runtime" "fmt" ) func main(){ go sayHello() go sayWorld() var str string fmt.Scan(&str) } func sayHello(){ for i := 0; i < 10; i++{ fmt.Print("hello ") runtime.Gosched() } } func sayWorld(){ for i := 0; i < 10; i++ { fmt.Println("world") runtime.Gosched() } } 

 

運行結果

從上面輸出結果可知，我們啟動了兩個線程，其中一個線程輸出一句后調用Gosched函數，釋放CPU權限；之后另一個線程獲得CPU權限。這樣兩個線程交替獲得cpu權限，才輸出了以上結果。

runtime.NumCPU()返回了cpu核數，runtime.NumGoroutine()返回當前進程的goroutine線程數。即便我們沒有開啟新的goroutine。

package main import ( "runtime" "fmt" ) func main(){ fmt.Println(runtime.NumCPU()) fmt.Println(runtime.NumGoroutine()) } 

 

運行結果

runtime.Goexit()函數用於終止當前的goroutine，單defer函數將會繼續被調用。

package main import ( "runtime" "fmt" ) func test(){ defer func(){ fmt.Println(" in defer") }() for i := 0; i < 10; i++{ fmt.Print(i) if i > 5{ runtime.Goexit() } } } func main(){ go test() var str string fmt.Scan(&str) } 

 

運行結果

在這里大家或許有個疑問，下面這兩句代碼干嘛的呢

var str string fmt.Scan(&str) 

這兩句代碼是等待輸入的意思，在這里用來阻止主線程關閉的。如果沒有這兩句的話，會發現我們的程序瞬間就結束了，而且什么都沒有輸出。這是因為主線程關閉之后，所有開啟的goroutine都會強制關閉，他還沒有來得及輸出，就結束了。
但是這樣感覺怪怪的。如果有一種機制，在子線程結束的時候通知一下主線程，然后主線程再關閉，豈不是更好，這樣就不用無休止的等待了。於是就有了channel。

channel

goroutine之間通過channel來通訊，可以認為channel是一個管道或者先進先出的隊列。你可以從一個goroutine中向channel發送數據，在另一個goroutine中取出這個值。
使用make創建

var channel chan int = make(chan int) // 或 channel := make(chan int) 

生產者/消費者是最經典的使用示例。生產者goroutine負責將數據放入channel，消費者goroutine從channel中取出數據進行處理。

package main import ( "fmt" ) func main(){ buf:=make(chan int) flg := make(chan int) go producer(buf) go consumer(buf, flg) <-flg //等待接受完成 } func producer(c chan int){ defer close(c) // 關閉channel for i := 0; i < 10; i++{ c <- i // 阻塞，直到數據被消費者取走后，才能發送下一條數據 } } func consumer(c, f chan int){ for{ if v, ok := <-c; ok{ fmt.Print(v) // 阻塞，直到生產者放入數據后繼續讀取數據 }else{ break } } f<-1 //發送數據，通知main函數已接受完成 } 

 

 

運行結果

可以將channel指定為單向通信。比如 <-chan int僅能接收， chan<-int僅能發送。之前的生產者消費者可以改為一下方式：

 

func producer(c chan<-int){ defer close(c) // 關閉channel for i := 0; i < 10; i++{ c <- i // 阻塞，直到數據被消費者取走后，才能發送下一條數據 } } func consumer(c <-chan int, f chan<-int){ for{ if v, ok := <-c; ok{ fmt.Print(v) // 阻塞，直到生產者放入數據后繼續讀取數據 }else{ break } } f<-1 //發送數據，通知main函數已接受完成 } 

channle可以是帶緩沖的。make的第二個參數作為緩沖長度來初始化一個帶緩沖的channel:

c := make(chan int, 5) 

向帶緩沖的channel發送數據時，只有緩沖區滿時，發送操作才會被阻塞。當緩沖區空時，接收才會阻塞。
可以通過以下程序調整發送和接收的順序調試

package main import ( "fmt" ) func main(){ c := make(chan int, 2) c <- 1 c <- 2 fmt.Println(<-c) fmt.Println(<-c) } 

select

如果有多個channel需要監聽，可以考慮用select,隨機處理一個可用的channel

package main import ( "fmt" ) func main(){ c := make(chan int) quit := make(chan int) go func(){ for i := 0; i < 10; i++{ fmt.Printf("%d ", <-c) } quit <- 1 }() testMuti(c, quit) } func testMuti(c, quit chan int){ x, y := 0, 1 for { select{ case c<-x: x, y = y, x+y case <-quit: fmt.Print("\nquit") return } } } 

 

運行結果

channle超時機制

當一個channel被read/write阻塞時，會被一直阻塞下去，直到channel關閉。產生一個異常退出程序。channel內部沒有超時的定時器。但我們可以用select來實現channel的超時機制

package main import ( "time" "fmt" ) func main(){ c := make(chan int) select{ case <- c: fmt.Println("沒有數據") case <-time.After(5* time.Second): fmt.Println("超時退出") } } 

 

運行結果

線程同步

假設現在我們有兩個線程，一個線程寫文件，一個線程讀文件。如果在讀文件的同時，寫文件的線程向文件中寫數據，就會出現問題。為了保證能夠正確的讀寫文件，在讀文件的時候，不能進行寫入文件的操作，在寫入時，不能進行讀的操作。這就需要互斥鎖。互斥鎖是線程間同步的一種機制，用了保證在同一時刻只用一個線程訪問共享資源。go中的互斥鎖在sync包中。下面是個線程安全的map:

package main import ( "errors" "sync" "fmt" ) func main(){ m := &MyMap{mp:make(map[string]int), mutex:new(sync.Mutex)} go SetValue(m) go m.Display() var str string fmt.Scan(&str) } type MyMap struct{ mp map[string]int mutex *sync.Mutex } func (this *MyMap)Get(key string)(int, error){ this.mutex.Lock() i, ok := this.mp[key] this.mutex.Unlock() if !ok{ return i, errors.New("不存在") } return i, nil } func (this *MyMap)Set(key string, val int){ this.mutex.Lock() defer this.mutex.Unlock() this.mp[key] = val } func (this *MyMap)Display(){ this.mutex.Lock() defer this.mutex.Unlock() for key, val := range this.mp{ fmt.Println(key, "=", val) } } func SetValue(m *MyMap){ var a rune a = 'a' for i := 0; i< 10; i++{ m.Set(string(a+rune(i)), i) } } 

 

運行結果

完