go語言中協程的實現機制

今天就跟大家聊聊有關go語言中協程的實現機制，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

協程（coroutine）是Go語言中的輕量級線程實現，由Go運行時（runtime）管理。

在一個函數調用前加上go關鍵字，這次調用就會在一個新的goroutine中并發執行。當被調用的函數返回時，這個goroutine也自動結束。需要注意的是，如果這個函數有返回值，那么這個返回值會被丟棄。

先看下面的例子：

func Add(x, y int) {
    z := x + y
    fmt.Println(z)
}

func main() {
    for i:=0; i<10; i++ {
        go Add(i, i)
    }
}

執行上面的代碼，會發現屏幕什么也沒打印出來，程序就退出了。

對于上面的例子，main()函數啟動了10個goroutine，然后返回，這時程序就退出了，而被啟動的執行Add()的goroutine沒來得及執行。我們想要讓main()函數等待所有goroutine退出后再返回，但如何知道goroutine都退出了呢？這就引出了多個goroutine之間通信的問題。

在工程上，有兩種最常見的并發通信模型：共享內存和消息。

來看下面的例子，10個goroutine共享了變量counter，每個goroutine執行完成后，將counter值加1.因為10個goroutine是并發執行的，所以我們還引入了鎖，也就是代碼中的lock變量。在main()函數中，使用for循環來不斷檢查counter值，當其值達到10時，說明所有goroutine都執行完畢了，這時main()返回，程序退出。

package main
import (
    "fmt"
    "sync"
    "runtime"
)

var counter int = 0

func Count(lock *sync.Mutex) {
    lock.Lock()
    counter++
    fmt.Println("counter =", counter)
    lock.Unlock()
}


func main() {

    lock := &sync.Mutex{}

    for i:=0; i<10; i++ {
        go Count(lock)
    }

    for {
        lock.Lock()

        c := counter

        lock.Unlock()

        runtime.Gosched()　　　　// 出讓時間片

        if c >= 10 {
            break
        }
    }
}

上面的例子，使用了鎖變量（屬于一種共享內存）來同步協程，事實上Go語言主要使用消息機制（channel）來作為通信模型。

channel

消息機制認為每個并發單元是自包含的、獨立的個體，并且都有自己的變量，但在不同并發單元間這些變量不共享。每個并發單元的輸入和輸出只有一種，那就是消息。

channel是Go語言在語言級別提供的goroutine間的通信方式，我們可以使用channel在多個goroutine之間傳遞消息。channel是進程內的通信方式，因此通過channel傳遞對象的過程和調用函數時的參數傳遞行為比較一致，比如也可以傳遞指針等。
channel是類型相關的，一個channel只能傳遞一種類型的值，這個類型需要在聲明channel時指定。

channel的聲明形式為：

var chanName chan ElementType

舉個例子，聲明一個傳遞int類型的channel：

var ch chan int

使用內置函數make()定義一個channel：

ch := make(chan int)

在channel的用法中，最常見的包括寫入和讀出：

// 將一個數據value寫入至channel，這會導致阻塞，直到有其他goroutine從這個channel中讀取數據
ch <- value

// 從channel中讀取數據，如果channel之前沒有寫入數據，也會導致阻塞，直到channel中被寫入數據為止
value := <-ch

默認情況下，channel的接收和發送都是阻塞的，除非另一端已準備好。

我們還可以創建一個帶緩沖的channel：

c := make(chan int, 1024)

// 從帶緩沖的channel中讀數據
for i:=range c {
　　...
}

此時，創建一個大小為1024的int類型的channel，即使沒有讀取方，寫入方也可以一直往channel里寫入，在緩沖區被填完之前都不會阻塞。

可以關閉不再使用的channel：

close(ch)

應該在生產者的地方關閉channel，如果在消費者的地方關閉，容易引起panic；

在一個已關閉 channel 上執行接收操作(<-ch)總是能夠立即返回，返回值是對應類型的零值。

現在利用channel來重寫上面的例子：

func Count(ch chan int) {
    ch <- 1
    fmt.Println("Counting")
}

func main() {

    chs := make([] chan int, 10)

    for i:=0; i<10; i++ {
        chs[i] = make(chan int)
        go Count(chs[i])
    }

    for _, ch := range(chs) {
        <-ch
    }
}

在這個例子中，定義了一個包含10個channel的數組，并把數組中的每個channel分配給10個不同的goroutine。在每個goroutine完成后，向goroutine寫入一個數據，在這個channel被讀取前，這個操作是阻塞的。

在所有的goroutine啟動完成后，依次從10個channel中讀取數據，在對應的channel寫入數據前，這個操作也是阻塞的。這樣，就用channel實現了類似鎖的功能，并保證了所有goroutine完成后main()才返回。

另外，我們在將一個channel變量傳遞到一個函數時，可以通過將其指定為單向channel變量，從而限制該函數中可以對此channel的操作。

單向channel變量的聲明：

var ch2 chan int  　　　　// 普通channel
var ch3 chan <- int 　　 // 只用于寫int數據
var ch4 <-chan int 　　 // 只用于讀int數據

可以通過類型轉換，將一個channel轉換為單向的：

ch5 := make(chan int)
ch6 := <-chan int(ch5)   // 單向讀
ch7 := chan<- int(ch5)  //單向寫

單向channel的作用有點類似于c++中的const關鍵字，用于遵循代碼“最小權限原則”。

例如在一個函數中使用單向讀channel：

func Parse(ch <-chan int) {
    for value := range ch {
        fmt.Println("Parsing value", value) 
    }
}

channel作為一種原生類型，本身也可以通過channel進行傳遞，例如下面這個流式處理結構：

type PipeData struct {
    value int
    handler func(int) int
    next chan int
}

func handle(queue chan *PipeData) {
    for data := range queue {
        data.next <- data.handler(data.value)
    }
}

select

在UNIX中，select()函數用來監控一組描述符，該機制常被用于實現高并發的socket服務器程序。Go語言直接在語言級別支持select關鍵字，用于處理異步IO問題，大致結構如下：

select {
    case <- chan1:
    // 如果chan1成功讀到數據
    
    case chan2 <- 1:
    // 如果成功向chan2寫入數據

    default:
    // 默認分支
}

select默認是阻塞的，只有當監聽的channel中有發送或接收可以進行時才會運行，當多個channel都準備好的時候，select是隨機的選擇一個執行的。

Go語言沒有對channel提供直接的超時處理機制，但我們可以利用select來間接實現，例如：

timeout := make(chan bool, 1)

go func() {
    time.Sleep(1e9)
    timeout <- true
}()

switch {
    case <- ch:
    // 從ch中讀取到數據

    case <- timeout:
    // 沒有從ch中讀取到數據，但從timeout中讀取到了數據
}

這樣使用select就可以避免永久等待的問題，因為程序會在timeout中獲取到一個數據后繼續執行，而無論對ch的讀取是否還處于等待狀態。

并發

早期版本的Go編譯器并不能很智能的發現和利用多核的優勢，即使在我們的代碼中創建了多個goroutine，但實際上所有這些goroutine都允許在同一個CPU上，在一個goroutine得到時間片執行的時候其它goroutine都會處于等待狀態。

實現下面的代碼可以顯式指定編譯器將goroutine調度到多個CPU上運行。

import "runtime"...
runtime.GOMAXPROCS(4)

PS：runtime包中有幾個處理goroutine的函數，

調度

Go調度的幾個概念：

M：內核線程；

G：go routine，并發的最小邏輯單元，由程序員創建；

P：處理器，執行G的上下文環境，每個P會維護一個本地的go routine隊列；

除了每個P擁有一個本地的go routine隊列外，還存在一個全局的go routine隊列。

具體調度原理：

1、P的數量在初始化由GOMAXPROCS決定；

2、我們要做的就是添加G；

3、G的數量超出了M的處理能力，且還有空余P的話，runtime就會自動創建新的M；

4、M拿到P后才能干活，取G的順序：本地隊列>全局隊列>其他P的隊列，如果所有隊列都沒有可用的G，M會歸還P并進入休眠；

一個G如果發生阻塞等事件會進行阻塞，如下圖：

G發生上下文切換條件：

系統調用；

讀寫channel；

gosched主動放棄，會將G扔進全局隊列；

如上圖，一個G發生阻塞時，M0讓出P，由M1接管其任務隊列；當M0執行的阻塞調用返回后，再將G0扔到全局隊列，自己則進入睡眠（沒有P了無法干活）；

看完上述內容，你們對go語言中協程的實現機制有進一步的了解嗎？如果還想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀。