Golang —— goroutine（協程）和channel（管道）
協程（goroutine） 是Go中應用程式併發處理的部分，它可以進行高效的併發運算。
- 協程是輕量的，比執行緒更廉價。使用4K的棧記憶體就可以在記憶體中建立。
- 能夠對棧進行分割，動態地增加或縮減記憶體的使用。棧的管理會在協程退出後自動釋放。
- 協程的棧會根據需要進行伸縮，不出現棧溢位。
協程的使用
package main
import (
"fmt"
"time"
)
func main() {
fmt.Println("In main()")
go longWait()
go shortWait()
fmt.Println("About to sleep in main()")
//time.Sleep(4 * 1e9)
time.Sleep(10 * 1e9)
fmt.Println("At the end of main()")
}
func longWait() {
fmt.Println("Beginning longWait()")
time.Sleep(5 * 1e9)
fmt.Println("End of longWait()")
}
func shortWait() {
fmt.Println("Beginning shortWait()")
time.Sleep(2 * 1e9)
fmt.Println("End of shortWait()")
}
複製程式碼
Go中用 go 關鍵字來開啟一個協程，其中 main 函式也可以看做是一個協程。
不難理解，上述程式碼的輸出為：
In main() About to sleep in main() Beginning shortWait() Beginning longWait() End of shortWait() End of longWait() At the end of main() 複製程式碼
但是，當我們將 main 的睡眠時間設定成4s時，輸出發生了改變。
In main() About to sleep in main() Beginning shortWait() Beginning longWait() End of shortWait() At the end of main() 複製程式碼
程式並沒有輸出 End of longWait() ，原因在於， longWait() 和 main() 執行在不同的協程中，兩者是非同步的。也就是說，早在 longWait() 結束之前，main已經退出，自然也就看不到輸出了。
通道（channel）
通道（channel） 是Go中一種特殊的資料型別，可以通過它們傳送型別化的資料在協程之間通訊，避開記憶體共享導致的問題。
通道的通訊方式保證了同步性，並且同一時間只有一個協程能夠訪問資料，不會出現 資料競爭 。
以工廠的傳輸帶為例，一個機器放置物品（生產者協程），經過傳送帶，到達下一個機器打包裝箱（消費者協程）。
通道的使用
在學習使用管道之前，我們先來看一個“悲劇”。
package main
import (
"fmt"
"time"
)
func main() {
fmt.Println("Reveal romantic feelings...")
go sendLove()
go responseLove()
waitFor()
fmt.Println("Leaving ☠️....")
}
func waitFor() {
for i := 0; i < 5; i++ {
fmt.Println("Keep waiting...")
time.Sleep(1 * 1e9)
}
}
func sendLove() {
fmt.Println("Love you, mm :heart:")
}
func responseLove() {
time.Sleep(6 * 1e9)
fmt.Println("Love you, too")
}
複製程式碼
用上面學習的知識，不難看出。。。真的慘啊
Reveal romantic feelings... Love you, mm :heart: Keep waiting... Keep waiting... Keep waiting... Keep waiting... Keep waiting... Leaving ☠️.... 複製程式碼
明明收到了暗戀女孩的迴應，然而卻以為對方不接受自己的情感，含淚離去。【TAT】
可見，協程之間沒有互相通訊將會引起多麼大的誤解。幸好，我們有了 channel ，現在就來一起改寫故事的結局吧~
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan string)
var answer string
fmt.Println("Reveal fomantic feelings...")
go sendLove()
go responseLove(ch)
waitFor()
answer = <-ch
if answer != "" {
fmt.Println(answer)
} else {
fmt.Println("Dead ☠️....")
}
}
func waitFor() {
for i := 0; i < 5; i++ {
fmt.Println("Keep waiting...")
time.Sleep(1 * 1e9)
}
}
func sendLove() {
fmt.Println("Love you, mm :heart:")
}
func responseLove(ch chan string) {
time.Sleep(6 * 1e9)
ch <- "Love you, too"
}
複製程式碼
輸出為：
Reveal fomantic feelings... Love you, mm :heart: Keep waiting... Keep waiting... Keep waiting... Keep waiting... Keep waiting... Love you, too 複製程式碼
皆大歡喜。
這裡我們用 ch := make(chan string) 建立了一個string型別的管道，當然我們還可以構建其他型別比如 ch := make(chan int) ，甚至一個函式管道 funcChan := chan func() 。
我們還用到了一個通訊操作符 <- 。
-
流向通道：
ch <- content，用管道ch傳送變數content。 -
從通道流出：
answer := <- ch，變數answer從通道ch接收資料。 -
<- ch可以單獨呼叫，以獲取通道的下一個值，當前值會被丟棄，但是可以用來驗證，比如：if <- ch != 100 { /* do something */ } 複製程式碼
通道阻塞
- 對於同一通道，傳送操作在接受者準備好之前是不會結束的。這就意味著，如果一個無緩衝通道在沒有空間接收資料的時候，新的輸入資料無法輸入，即傳送者處於阻塞狀態。
- 對於同一通道，接收操作是阻塞的，直到傳送者可用。如果通道中沒有資料，接收者會保持阻塞。
以上兩條性質，反映了 無緩衝通道 的特性： 同一時間只允許至多一個數據存在於通道中 。
我們通過例子來感受一下：
package main
import "fmt"
func main() {
ch1 := make(chan int)
go pump(ch1)
fmt.Println(<-ch1)
}
func pump(ch chan int) {
for i := 0; ; i++ {
ch <- i
}
}
複製程式碼
程式輸出：
0 複製程式碼
這裡的 pump() 函式被稱為 生產者 。
解除通道阻塞
package main
import "fmt"
import "time"
func main() {
ch1 := make(chan int)
go pump(ch1)
go suck(ch1)
time.Sleep(1e9)
}
func pump(ch chan int) {
for i := 0; ; i++ {
ch <- i
}
}
func suck(ch chan int) {
for {
fmt.Println(<-ch)
}
}
複製程式碼
這裡我們定義了一個 suck 函式，作為 接收者 ，並給 main 協程一個1s的執行時間，於是，便產生了70W+的輸出【TAT】。
通道死鎖
通道兩段互相阻塞對方，會形成死鎖狀態。Go執行時會檢查並panic，停止程式。無緩衝通道會被阻塞。
package main
import "fmt"
func main() {
out := make(chan int)
out <- 2
go f1(out)
}
func f1(in chan int) {
fmt.Println(<-in)
}
複製程式碼
fatal error: all goroutines are asleep - deadlock! 複製程式碼
顯然在 out <- 2 的時候，由於沒有接受者，主執行緒被阻塞。
同步通道
除了普通的無快取通道外，還有一種特殊的帶快取通道—— 同步通道 。
buf := 100 ch1 := make(chan string, buf) 複製程式碼
buf 是通道可以同時容納的元素個數，即 ch1 的緩衝區大小，在 buf 滿之前，通道都不會阻塞。
如果容量大於0，通道就是非同步的：在緩衝滿載或邊控之前通訊不會阻塞，元素會按照發送的順序被接收。
同步： ch := make(chan type, value)
- value ==0 --> synchronous, unbuffered（阻塞）
- value > 0 --> asynchronous, buffered（非阻塞）取決於value元素
使用通道緩衝能使程式更具有伸縮性（scalable）。
儘量在首要位置使用無緩衝通道，只在不確定的情況下使用緩衝。
package main
import "fmt"
import "time"
func main() {
c := make(chan int, 50)
go func() {
time.Sleep(15 * 1e9)
x := <-c
fmt.Println("received", x)
}()
fmt.Println("sending", 10)
c <- 10
fmt.Println("send", 10)
}
複製程式碼
訊號量模式
func compute(ch chan int) {
ch <- someComputation()
}
func main() {
ch := make(chan int)
go compute(ch)
doSomethingElaseForAWhile()
result := <-ch
}
複製程式碼
協程通過在通道 ch 中放置一個值來處理結束訊號。 main 執行緒等待 <-ch 直到從中獲取到值。
我們可以用它來處理切片排序：
done := make(chan bool)
doSort := func(s []int) {
sort(s)
done <- true
}
i := pivot(s)
go doSort(s[:i])
go doSort(s[i:])
<-done
<-done
複製程式碼
帶緩衝通道實現訊號量
訊號量時實現互斥鎖的常用同步機制，限制對資源的訪問，解決讀寫問題。
- 帶緩衝通道的容量要和同步的資源容量相同
- 通道的長度（當前存放的元素個數）與當前資源被使用的數量相同
- 容量減去通道的長度等於未處理的資源個數
//建立一個長度可變但容量為0的通道
type Empty interface {}
type semaphore chan Empty
複製程式碼
初始化訊號量
sem = make(semaphore, N) 複製程式碼
對訊號量進行操作，建立互斥鎖
func (s semaphore) P (n int) {
e := new(Empty)
for i := 0; i < n; i++ {
s <- e
}
}
func (a semaphore) V (n int) {
for i := 0; i < n; i++ {
<- s
}
}
/* mutexes */
func (s semaphore) Lock() {
s.P(1)
}
func (s semaphore) Unlock(){
s.V(1)
}
/* signal-wait */
func (s semaphore) Wait(n int) {
s.P(n)
}
func (s semaphore) Signal() {
s.V(1)
}
複製程式碼
通道工廠模式
不將通道作為引數傳遞，而是在函式內生成一個通道，並返回。
package main
import (
"fmt"
"time"
)
func main() {
stream := pump()
go suck(stream)
time.Sleep(1e9)
}
func pump() chan int {
ch := make(chan int)
go func() {
for i := 0; ; i++ {
ch <- i
}
}()
return ch
}
func suck(ch chan int) {
for {
fmt.Println(<-ch)
}
}
複製程式碼
通道使用for迴圈
for 迴圈可以從 ch 中持續獲取值，直到通道關閉。（這意味著必須有另一個協程寫入 ch ，並且在寫入完成後關閉）
for v := range ch {
fmt.Println("The value is", v)
}
複製程式碼
package main
import (
"fmt"
"time"
)
func main() {
suck(pump())
time.Sleep(1e9)
}
func pump() chan int {
ch := make(chan int)
go func() {
for i := 0; ; i++ {
ch <- i
}
}()
return ch
}
func suck(ch chan int) {
go func() {
for v := range ch {
fmt.Println(v)
}
}()
}
複製程式碼
通道的方向
通道可以表示它只發送或者只接受：
var send_only chan<- int// channel can only send data var recv_only <-chan int// channel can only receive data 複製程式碼
只接收的通道（<-chan T）無法關閉，因為關閉通道是傳送者用來表示不再給通道傳送值，所以對只接收通道是沒有意義的。
管道和選擇器模式
借鑑一個經典的例子 篩法求素數 來學習這一內容。
這個演算法的主要思想是，引入 篩法 （一種時間複雜度為O(x * ln(lnx))的演算法），對一個給定返回的正整數從大到小排序，然後從中篩選掉所有的非素數，那麼剩下的數中最小的就是素數，再去掉該數的倍數，以此類推。
假設一個範圍為1~30的正整數集，已經從大到小排序。
第一遍篩掉非素數1，然後剩餘數中最小的是2。
由於2是一個素數，將其取出，然後去掉所有2的倍數，那麼剩下的數為：
3 5 7 9 11 13 15 17 19 21 23 25 27 29
剩下的數中3最小，且為素數，取出並去除所有3的倍數，迴圈直至所有數都篩完。
程式碼如下：
// 一般寫法
package main
import (
"fmt"
)
func generate(ch chan int) {
for i := 2; i < 100; i++ {
ch <- i
}
}
func filter(in, out chan int, prime int) {
for {
i := <-in
if i%prime != 0 {
out <- i
}
}
}
func main() {
ch := make(chan int)
go generate(ch)
for {
prime := <-ch
fmt.Print(prime, " ")
ch1 := make(chan int)
go filter(ch, ch1, prime)
ch = ch1
}
}
複製程式碼
// 習慣寫法
package main
import (
"fmt"
)
func generate() chan int {
ch := make(chan int)
go func() {
for i := 2; ; i++ {
ch <- i
}
}()
return ch
}
func filter(in chan int, prime int) chan int {
out := make(chan int)
go func() {
for {
if i := <-in; i%prime != 0 {
out <- i
}
}
}()
return out
}
func sieve() chan int {
out := make(chan int)
go func() {
ch := generate()
for {
prime := <-ch
ch = filter(ch, prime)
out <- prime
}
}()
return out
}
func main() {
primes := sieve()
for {
fmt.Println(<-primes)
}
}
複製程式碼