💢歡迎來到張胤塵的開源技術站
💥開源如江河，匯聚眾志成。代碼似星辰，照亮行征程。開源精神長，傳承永不忘。攜手共前行，未來更輝煌💥

通道

在傳統的并發編程中，多個線程或進程之間通常通過共享內存來通信。這種模型雖然高效，但容易引發 競爭條件 和 死鎖 等問題。為了避免這些問題，程序員在開發時需要使用復雜的同步機制（例如：鎖、信號量等）來保護共享數據。但是 golang 采用了不同的思路：避免共享內存，通過通信來實現并發。

示意圖所下所示：

通道就是 golang 中實現 Goroutine 之間通信的機制。它是一種 類型化的通道，允許多個 Goroutine 之間安全地傳遞數據。通道是 Golang 并發模型的核心，它解決了傳統并發編程中共享內存帶來的復雜性和風險。

本篇文章主要介紹的是通道，有關于協程、并發編程等相關知識點請關注后續文章《Golang 協程》、《Golang 并發編程》。

通道聲明

使用 var 關鍵字聲明通道變量。如下所示：

var ch chan int

聲明后，ch 是一個未初始化的通道，其默認值為 nil。

初始化

使用 make 函數初始化通道，如下所示：

ch = make(chan int)

也可以使用 make 函數創建一個帶緩沖區的通道，如下所示：

ch = make(chan int, 10)

緩沖機制

通道分為 無緩沖通道 和 帶緩沖通道，它們在行為和使用場景上有一些關鍵區別。

無緩沖通道

在初始化的小結中，使用 make 函數進行初始化通道，當沒有指定通道的容量，或者說通道的容量大小為 0 時，稱為無緩沖通道。

無緩沖通道在發送數據和接收數據時存在如下的特點：

• 發送操作：發送數據時，發送方會阻塞，直到有接收方準備接收數據。
• 接收操作：接收方會阻塞，直到有發送方發送數據。
• 無緩沖通道的發送和接收操作是同步的，必須有發送方和接收方同時準備好，才能完成通信。

代碼示例

package mainimport ("fmt""sync""time"
)func main() {unbufferedChan := make(chan int) // 創建無緩沖通道var wg sync.WaitGroupwg.Add(1)go func() {defer wg.Done()fmt.Println("receiver is waiting...") // receiver is waiting...time.Sleep(time.Second * 3)	// 模擬3秒的準備時間data := <-unbufferedChan // 在這行函數執行之前，發送方處于阻塞狀態fmt.Println("received:", data)}()unbufferedChan <- 42 	// 發送方阻塞，直到接收方準備好close(unbufferedChan)	// 關閉通道wg.Wait()
}

帶緩沖通道

當指定了通道的容量，例如 make(chan int, 10)，則稱為帶緩沖通道。

帶緩沖通道在發送數據和接收數據時存在如下的特點：

• 發送操作：發送數據時，如果緩沖區未滿，數據會被放入緩沖區，發送方不會阻塞；如果緩沖區已滿，發送方會阻塞，直到緩沖區有空間。
• 接收操作：接收方從緩沖區中取出數據，如果緩沖區為空，接收方會阻塞，直到有數據可用。
• 帶緩沖通道的發送和接收操作是異步的，發送方和接收方不需要同時準備好。緩沖區的存在允許數據在發送方和接收方之間暫時存儲。

代碼示例

package mainimport ("fmt""sync""time"
)func main() {bufferedChan := make(chan int, 2) // 創建帶緩沖通道，容量大小為2var wg sync.WaitGroupwg.Add(1)go func() {defer wg.Done()fmt.Println("receiver is waiting...")// for {// 	select {// 	case data, ok := <-bufferedChan:// 		if ok {// 			fmt.Println("received:", data)// 		} else {// 			fmt.Println("bufferedChan closed")// 			return// 		}// 	}// }for {time.Sleep(time.Second * 5) // 模擬接收方不從緩沖通道中接收數據}}()bufferedChan <- 42bufferedChan <- 43bufferedChan <- 44 // 如果接收方一直未接收數據，則在此處會發送阻塞fmt.Println("send data over ...")close(bufferedChan) // 關閉通道wg.Wait()
}

通道操作

發送數據

使用 <- 操作符將數據發送到通道，如下所示：

package mainfunc main() {ch := make(chan int)ch <- 10 // 向通道中發送一個 10close(ch) // 關閉通道
}

接收數據

使用 <- 操作符從通道中讀取數據，如下所示：

package mainimport "fmt"func main() {ch := make(chan int, 1)ch <- 10 // 向通道中發送一個 10data := <-ch      // 通道中讀取一個數據fmt.Println(data) // 10close(ch) // 關閉通道
}

單向通道

golang 中提供了單向通道這么一種特殊的通道類型，它只能用于發送或接收數據，而不能同時進行發送和接收操作。單向通道在類型聲明上與普通通道（雙向通道）有所不同，主要用于限制通道的使用方式，從而提高代碼的可讀性和安全性。

單向通道有如下兩種類型：

• 只發送通道：用于發送數據，但不能接收數據。

chan<- Type

• 只接收通道：用于接收數據，但不能發送數據。

<-chan Type

單向通道的用途

單向通道的主要用途是限制通道的使用范圍，避免在函數或方法中濫用通道的發送和接收功能。例如：

• 當一個函數只需要從通道中讀取數據時，使用只接收通道可以明確表示該函數的意圖。
• 當一個函數只需要向通道中寫入數據時，使用只發送通道可以避免意外讀取通道中的數據。

代碼示例

package mainimport ("fmt""sync"
)func producer(ch chan<- int, wg *sync.WaitGroup) { // ch 參數：一個只發送通道defer wg.Done()for i := 0; i < 5; i++ {ch <- i // 向通道發送數據fmt.Printf("sent: %d\n", i)}close(ch) // 關閉通道
}func consumer(ch <-chan int, wg *sync.WaitGroup) { // ch 參數：一個只接收通道defer wg.Done()for {select {case data, ok := <-ch:	// 讀取通道數據if ok {fmt.Printf("received: %d\n", data)} else {fmt.Println("chan closed ...")return}}}
}func main() {var wg sync.WaitGroupch := make(chan int) // 創建一個雙向通道wg.Add(2)// received: 0// sent: 0// sent: 1// received: 1// received: 2// sent: 2// sent: 3// received: 3// received: 4// sent: 4// chan closed ...go consumer(ch, &wg) // 接收者go producer(ch, &wg) // 發送者wg.Wait()
}

多路復用

通道的多路復用機制是一種用于處理多個通道操作的技術，它允許程序同時等待多個通道的讀寫操作。這種機制的核心是 select...case 語句，它提供了對多個通道操作的并發處理能力。

select {
case <-ch1:// 處理 ch1 的讀操作
case data := <-ch2:// 處理 ch2 的讀操作，并將讀取到的數據賦值給 data
case ch3 <- value:// 向 ch3 中發送值
default:// 如果沒有通道準備好，則執行默認邏輯
}

另外需要強調的是，如果多個通道同時準備好，select...case 會隨機選擇一個通道執行操作；如果沒有任何一個通道準備好，則 select 會陷入阻塞，除非有 default 分支執行。

需要注意的是，上面提到的 通道準備好 是個口語，對于接收操作來說需要滿足以下兩個條件中的任意一個：

• 通道中有數據可讀：如果通道的緩沖區中有數據，或者有發送操作正在等待發送數據到通道中，那么接收操作就準備好。
• 通道已關閉：即使通道中沒有數據，如果通道已經被關閉，接收操作也會立即返回零值，并且 ok 為 false（如果使用了 data, ok := <-ch 的形式）。

對發送操作來說也需要滿足以下兩個條件中的任意一個：

• 通道中有空間可寫：如果通道是無緩沖的，并且有等待接收的協程，或者通道是緩沖的且緩沖區中有空閑位置，那么發送操作就準備好。
• 通道已關閉：如果通道已經被關閉，發送操作會觸發 panic("send on closed channel")。

代碼示例

通過 select 同時處理多個通道，如下所示：

package mainimport ("fmt""time"
)func main() {ch1 := make(chan string) // 創建通道 ch1ch2 := make(chan string) // 創建通道 ch2go func() { // 每間隔 1 秒，向通道 ch1 中發送數據，一共發送 10 條數據defer close(ch1)for i := 1; i <= 10; i++ {ch1 <- "message from ch1"time.Sleep(1 * time.Second)}}()go func() { // 每間隔 2 秒，向通道 ch2 中發送數據，一共發送 10 條數據defer close(ch2)for i := 1; i <= 10; i++ {ch2 <- "message from ch2"time.Sleep(2 * time.Second)}}()// no data reception ...// recived:  message from ch2// recived:  message from ch1// no data reception ...// no data reception ...// recived:  message from ch1// no data reception ...// recived:  message from ch2// recived:  message from ch1// no data reception ...// recived:  message from ch1// no data reception ...// no data reception ...// recived:  message from ch1// recived:  message from ch2// no data reception ...// recived:  message from ch1// no data reception ...// recived:  message from ch1// recived:  message from ch2// no data reception ...// recived:  message from ch1// no data reception ...// no data reception ...// recived:  message from ch1// recived:  message from ch2// no data reception ...// recived:  message from ch1// no data reception ...// recived:  message from ch2// ch1 chan closed ...// recived:  message from ch2// recived:  message from ch2// recived:  message from ch2// recived:  message from ch2// ch2 chan closed ...// both channels closed. exiting...ch1_s, ch2_s := true, truefor ch1_s || ch2_s {select {case data, ok := <-ch1:if ok {fmt.Println("recived: ", data)} else if ch1_s {fmt.Println("ch1 chan closed ...")ch1_s = false}case data, ok := <-ch2:if ok {fmt.Println("recived: ", data)} else if ch2_s {fmt.Println("ch2 chan closed ...")ch2_s = false}default:fmt.Println("no data reception ...")time.Sleep(time.Second)}}fmt.Println("both channels closed. exiting...")
}

通道復用器的實現

在某些復雜場景中，可能需要手動實現通道復用器。例如，可以將多個通道的輸出合并到一個通道中，從而簡化后續的處理邏輯。

下面是一個簡單的通道復用器實現思路：

package mainimport ("fmt""sync""time"
)func send(ch chan<- int) { // 模擬數據發送，每秒發送一個數字defer close(ch)for i := 1; i <= 10; i++ {ch <- itime.Sleep(time.Second)}
}func multiplexer(channels ...<-chan int) <-chan int { // 合并多個通道數據，函數返回最終的一個只讀通道out := make(chan int)go func() {defer close(out)var wg sync.WaitGroupfor _, ch := range channels {wg.Add(1)go func(ch <-chan int) {defer wg.Done()for {select {case data, ok := <-ch:if ok {out <- data // 數據發送到最終的只讀通道} else {return}}}}(ch)}wg.Wait()}()return out
}func main() {ch1 := make(chan int) // 創建通道 ch1ch2 := make(chan int) // 創建通道 ch2ch3 := make(chan int) // 創建通道 ch3go send(ch1)go send(ch2)go send(ch3)ret_ch := multiplexer(ch1, ch2, ch3) // 接收返回的通道for data := range ret_ch { // 從只讀通道中獲取合并后的數據fmt.Printf("ret received: %d\n", data)}fmt.Println("ret chan closed ...")
}

超時機制

在 golang 中，通道的超時機制可以通過 time.After 或 context 兩種方式來實現。

time.After

time.After 是一個返回通道的函數，會在指定的超時時間后向通道發送一個時間值。結合 select...case 語句，可以實現超時邏輯。如下所示：

package mainimport ("fmt""time"
)func main() {ch := make(chan string) // 創建通道go func() {time.Sleep(3 * time.Second) // 模擬耗時操作ch <- "任務完成"}()select {case res := <-ch:fmt.Println(res)case <-time.After(2 * time.Second): // 設置超時為2秒fmt.Println("超時退出") // 最終打印 超時退出}
}

time.After 是一種非常簡潔且易于理解的超時機制，特別適合簡單的超時場景。但是需要注意的是，time.After 會在后臺啟動一個定時器，即使 select 提前退出，定時器也不會立刻回收，可能導致輕微的資源泄漏。

context

context 是 golang 中用于傳遞可取消信號、超時時間等的工具。通過 context.WithTimeout 創建一個帶有超時功能的上下文，其 Done() 方法返回一個通道，用于超時控制，如下所示：

package mainimport ("context""fmt""time"
)func main() {ch := make(chan string) // 創建通道ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)defer cancel() // cancel() 會釋放與上下文相關的資源，避免內存泄漏go func() {time.Sleep(3 * time.Second) // 模擬耗時操作ch <- "任務完成"}()select {case res := <-ch:fmt.Println(res)case <-ctx.Done(): // 監聽超時信號fmt.Println("超時退出", ctx.Err())}
}

在上面的這個代碼中：

• context.WithTimeout 創建了一個帶有超時的上下文，并設置超時時間為 2 秒。
• defer cancel() 確保在函數返回時調用 cancel()，釋放資源。
• 如果任務在超時前完成，cancel() 會被調用，終止所有監聽 ctx.Done() 的協程。

總的來說，context 更適用于復雜的并發場景，例如多個任務的超時控制、任務取消等；time.After 更適用于簡單的超時控制，例如單個任務的超時。

關閉通道

使用 close 函數關閉通道，如下所示：

package mainfunc main() {ch := make(chan int, 5)close(ch) // 關閉通道
}

需要注意的是，當關閉通道后，不能再向通道發送數據，但可以繼續從通道中接收數據。

檢查通道是否關閉

接收數據時會有兩個返回值，一個是數據另一個是布爾值，用于判斷通道是否關閉，如下所示：

package mainimport ("fmt""sync"
)func main() {ch := make(chan int, 5)var wg sync.WaitGroupwg.Add(1)go func() {defer wg.Done()for {select {case data, ok := <-ch: // data 表示數據，ok 表示通道是否關閉if ok {// received: 11// received: 10// received: 15fmt.Println("received:", data)} else {fmt.Println("ch closed") // ch closedreturn}}}}()ch <- 11ch <- 10ch <- 15close(ch) // 關閉通道wg.Wait()
}

獲取通道長度

使用 len 函數獲取通道的當前長度，如下所示：

package mainimport "fmt"func main() {ch := make(chan int, 5)fmt.Println(len(ch)) // 0ch <- 10fmt.Println(len(ch)) // 1close(ch) // 關閉通道
}

獲取通道容量

使用 cap 函數獲取通道的當前容量，如下所示：

package mainimport "fmt"func main() {ch := make(chan int, 5)fmt.Println(cap(ch)) // 5ch1 := make(chan int)fmt.Println(cap(ch1)) // 0close(ch)  // 關閉通道 chclose(ch1) // 關閉通道 ch1
}

源碼解析

針對通道的源代碼進行解析，從以下幾個方面：

• 創建通道
• 發送數據
• 接收數據
• 關閉通道

通道結構體

源碼位置：src/runtime/chan.go

type hchan struct {qcount   uint           // total data in the queuedataqsiz uint           // size of the circular queuebuf      unsafe.Pointer // points to an array of dataqsiz elementselemsize uint16closed   uint32timer    *timer // timer feeding this chanelemtype *_type // element typesendx    uint   // send indexrecvx    uint   // receive indexrecvq    waitq  // list of recv waiterssendq    waitq  // list of send waiters// lock protects all fields in hchan, as well as several// fields in sudogs blocked on this channel.//// Do not change another G's status while holding this lock// (in particular, do not ready a G), as this can deadlock// with stack shrinking.lock mutex
}

• qcount：當前通道中存儲的數據數量，對于無緩沖通道，qcount 的值通常為 0 或 1。
• dataqsiz：對于緩沖通道，dataqsiz 表示緩沖區可以存儲的最大元素數量。對于無緩沖通道，dataqsiz 為 0。
• buf：指向緩沖區的指針。緩沖區是一個數組，用于存儲通道中的數據。僅對緩沖通道有效。無緩沖通道的 buf 為 nil。
• elemsize：通道中每個元素的大小（以字節為單位）。
• closed：標記通道是否已關閉。關閉的通道不能再次發送數據，但可以繼續接收數據直到緩沖區為空。
• timer：指向一個定時器，該定時器與通道相關聯（例如，用于超時操作）。
• elemtype：指向通道中元素的類型信息，用于在運行時檢查通道中存儲的數據類型是否正確。
• sendx：用于管理緩沖區的環形隊列，記錄下一次發送操作在緩沖區中的索引。
• recvx：用于管理緩沖區的環形隊列，記錄下一次接收操作在緩沖區中的索引。
• recvq：存儲等待接收的協程隊列，在發送操作中，如果緩沖區已滿且沒有等待接收的協程，則發送協程會被加入到 sendq。
• sendq：存儲等待發送的協程隊列，在接收操作中，如果緩沖區為空且沒有等待發送的協程，則接收協程會被加入到 recvq。
• lock：保護 hchan 結構體中所有字段的互斥鎖，確保通道操作的線程安全性。在發送、接收和關閉操作中，lock 用于防止多個協程同時修改通道的狀態。

創建通道

在 golang 的運行時中，創建通道的代碼會被編譯為對 makechan 的調用。如下所示：

package mainfunc main() {ch := make(chan int, 1)
}

編譯成匯編代碼如下所示：

0x001a 00026        CALL    runtime.makechan(SB)

以上匯編代碼只是部分截取，請注意甄別。

makechan 函數是運行時中用于創建通道的核心函數。它初始化了一個 hchan 結構體，并根據指定的類型和緩沖區大小分配內存。

源碼位置：src/runtime/chan.go

函數原型

func makechan(t *chantype, size int) *hchan {// ...
}

• t *chantype：通道的類型信息，包含通道中元素的類型。
• size int：通道的緩沖區大小。如果為 0，則創建無緩沖通道；如果大于 0，則創建有緩沖通道。
• 返回一個初始化后的 hchan 結構體指針。

函數內容

func makechan(t *chantype, size int) *hchan {elem := t.Elem	// 從通道類型 t 中獲取通道中元素的類型信息// compiler checks this but be safe.// 檢查通道中元素的大小是否超過 64KB，如果超過，拋出異常if elem.Size_ >= 1<<16 {throw("makechan: invalid channel element type")}// 確保 hchan 的大小是最大對齊單位的倍數，并且元素的對齊要求不超過最大對齊單位// maxAlign = 8if hchanSize%maxAlign != 0 || elem.Align_ > maxAlign {throw("makechan: bad alignment")}// 計算緩沖區的總大小：elem.Size_ * sizemem, overflow := math.MulUintptr(elem.Size_, uintptr(size))// 檢查是否發生溢出，或者緩沖區大小超過最大分配限制，或者緩沖區大小小于0，拋出異常if overflow || mem > maxAlloc-hchanSize || size < 0 {panic(plainError("makechan: size out of range"))}// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.// buf points into the same allocation, elemtype is persistent.// SudoG's are referenced from their owning thread so they can't be collected.// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.// 內存分配var c *hchanswitch {case mem == 0:// 如果緩沖區大小為 0（無緩沖通道）或元素大小為 0，僅分配 hchan 的內存// Queue or element size is zero.c = (*hchan)(mallocgc(hchanSize, nil, true))// Race detector uses this location for synchronization.// 初始化通道的緩沖區的指針// buf 指向 hchan 結構體本身，表示沒有獨立的緩沖區c.buf = c.raceaddr()case !elem.Pointers():// 如果元素類型不包含指針（如基本類型或數組），將 hchan 和緩沖區分配在同一塊內存中// Elements do not contain pointers.// Allocate hchan and buf in one call.c = (*hchan)(mallocgc(hchanSize+mem, nil, true))// 初始化通道的緩沖區的指針// 將 c.buf 指向 hchan 結構體之后的內存區域，該區域用于存儲緩沖區數據c.buf = add(unsafe.Pointer(c), hchanSize)default:// 如果元素類型包含指針（如結構體或切片），分別分配 hchan 和緩沖區的內存// 因為 gc 需要跟蹤指針類型的內存分配// Elements contain pointers.c = new(hchan)// 初始化通道的緩沖區的指針c.buf = mallocgc(mem, elem, true)}// 設置通道的元素大小c.elemsize = uint16(elem.Size_)// 設置通道的元素類型c.elemtype = elem// 設置通道的緩沖區大小c.dataqsiz = uint(size)// 初始化通道的互斥鎖lockInit(&c.lock, lockRankHchan)// 如果啟用了通道調試模式，打印調試信息if debugChan {print("makechan: chan=", c, "; elemsize=", elem.Size_, "; dataqsiz=", size, "\n")}// 返回通道指針return c
}

具體分配內存 mallocgc 函數原型如下所示：

func mallocgc(size uintptr, typ *_type, needzero bool) unsafe.Pointer {// ...
}

• size：分配的內存大小。
• typ：分配內存的類型信息。
• needzero：是否需要將分配的內存清零。

更多關于 mallocgc 函數的內容本文章不再贅述，感興趣的同學請關注后續文章《Golang 內存模型》。

發送數據

在 golang 的運行時中，發送操作的代碼會被編譯為對 chansend1 的調用。如下所示：

package mainfunc main() {ch := make(chan int, 1)ch <- 1
}

編譯成匯編代碼如下所示：

0x001a 00026        CALL    runtime.makechan(SB)
# ...
0x0026 00038        CALL    runtime.chansend1(SB)

以上匯編代碼只是部分截取，請注意甄別。

而在 chansend1 的函數內部，又是對 chansend 的調用，如下所示：

源碼位置：src/runtime/chan.go

// entry point for c <- x from compiled code.
//
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {chansend(c, elem, true, getcallerpc())
}

函數原型

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// ...
}

• c *hchan：指向通道的指針。
• ep unsafe.Pointer：指向數據的指針。
• block bool：是否阻塞發送。
  • 如果 block == true，發送操作會在通道緩沖區滿或沒有接收方時阻塞，直到可以發送數據。
  • 如果 block == false，發送操作是非阻塞的。如果當前無法發送數據（緩沖區滿或沒有接收方），函數會立即返回 false。
• callerpc uintptr：程序計數器（PC）的值，主要用于調試和競態檢測。
• 發送操作是否成功。

函數內容

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// 檢查通道指針 c 是否為 nilif c == nil {if !block {// 如果通道為 nil 且是非阻塞發送，直接返回 falsereturn false}// 如果是阻塞發送，則協程會阻塞并拋出異常gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2)throw("unreachable")}// 如果啟用了通道調試模式，打印調試信息if debugChan {print("chansend: chan=", c, "\n")}// 如果啟用了競態檢測器，記錄對通道的讀操作if raceenabled {racereadpc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(chansend))}// Fast path: check for failed non-blocking operation without acquiring the lock.//// After observing that the channel is not closed, we observe that the channel is// not ready for sending. Each of these observations is a single word-sized read// (first c.closed and second full()).// Because a closed channel cannot transition from 'ready for sending' to// 'not ready for sending', even if the channel is closed between the two observations,// they imply a moment between the two when the channel was both not yet closed// and not ready for sending. We behave as if we observed the channel at that moment,// and report that the send cannot proceed.//// It is okay if the reads are reordered here: if we observe that the channel is not// ready for sending and then observe that it is not closed, that implies that the// channel wasn't closed during the first observation. However, nothing here// guarantees forward progress. We rely on the side effects of lock release in// chanrecv() and closechan() to update this thread's view of c.closed and full().// 如果是非阻塞發送并且通道未關閉但緩沖區已滿，直接返回 false// 前置判斷，避免進入鎖的開銷if !block && c.closed == 0 && full(c) {return false}// 如果啟用了阻塞剖析，記錄當前時間戳var t0 int64if blockprofilerate > 0 {t0 = cputicks()}// 鎖定通道的互斥鎖，確保操作的原子性lock(&c.lock)// 如果通道已關閉，解鎖并拋出異常// 閱讀到此處時，考慮是否改為使用原子操作，代替 lock/unlock？// 可？不可？// 記錄吧！！！if c.closed != 0 {unlock(&c.lock)panic(plainError("send on closed channel"))}// 如果有等待接收的協程，直接將數據發送給接收協程，跳過緩沖區if sg := c.recvq.dequeue(); sg != nil {// Found a waiting receiver. We pass the value we want to send// directly to the receiver, bypassing the channel buffer (if any).// 調用 send 函數完成數據傳遞，并解鎖send(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}// 如果沒有等待接收的協程，則判斷緩沖區是否有空間if c.qcount < c.dataqsiz {// Space is available in the channel buffer. Enqueue the element to send.// 計算緩沖區中下一個寫入位置的指針qp := chanbuf(c, c.sendx)if raceenabled {// 通知競態檢測器當前操作的上下文信息，對緩沖區的寫入操作被正確跟蹤racenotify(c, c.sendx, nil)}// 將要發送的數據從 ep 復制到緩沖區的 qp 位置typedmemmove(c.elemtype, qp, ep)// 更新發送索引為下一個寫入位置c.sendx++// 如果 c.sendx 達到緩沖區大小，則將其重置為 0，實現環形緩沖區的效果if c.sendx == c.dataqsiz {c.sendx = 0}// 每次成功寫入數據后，c.qcount 增加 1c.qcount++// 解鎖通道的互斥鎖unlock(&c.lock)// 寫入成功，返回 truereturn true}// 如果緩沖區已滿且是非阻塞發送，解鎖并返回 falseif !block {unlock(&c.lock)return false}// 下面是阻塞發送的代碼邏輯// Block on the channel. Some receiver will complete our operation for us.// 獲取當前正在運行的協程的指針gp := getg()// 分配一個 sudog 對象，sudog 是 golang 運行時中用于表示協程在通道操作中等待的結構體mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}// No stack splits between assigning elem and enqueuing mysg// on gp.waiting where copystack can find it.// 下面這些是初始化 sudog 對象屬性mysg.elem = ep	// 指向要發送的數據指針mysg.waitlink = nilmysg.g = gp	// 指向當前協程mysg.isSelect = false	// 標記是否是 select 操作mysg.c = c	// 指向當前通道gp.waiting = mysg	// 將當前協程的 sudog 對象設置為等待狀態gp.param = nil// 將 sudog 對象加入通道的發送等待隊列c.sendq.enqueue(mysg)// Signal to anyone trying to shrink our stack that we're about// to park on a channel. The window between when this G's status// changes and when we set gp.activeStackChans is not safe for// stack shrinking.// 標記當前協程即將阻塞在通道操作gp.parkingOnChan.Store(true)// 阻塞當前協程，直到被接收操作喚醒gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2)// Ensure the value being sent is kept alive until the// receiver copies it out. The sudog has a pointer to the// stack object, but sudogs aren't considered as roots of the// stack tracer.// 確保發送的數據在接收協程拷貝之前不會被回收，因為 ep 指向的數據有可能是棧上的數據// 而棧上的數據可能在協程阻塞后被回收KeepAlive(ep)// 下面是協程被喚醒后的處理代碼// someone woke us up.if mysg != gp.waiting {throw("G waiting list is corrupted")}// 清空協程的等待狀態gp.waiting = nil	// 標記協程不再阻塞在通道操作上gp.activeStackChans = false// 發送操作是否成功，mysg.success 表示 true 成功，false 失敗closed := !mysg.successgp.param = nilif mysg.releasetime > 0 {// 記錄協程阻塞的時間blockevent(mysg.releasetime-t0, 2)}// 清空 sudog 對象的通道指針mysg.c = nilreleaseSudog(mysg)if closed {// 如果通道未關閉但協程被喚醒，拋出異常if c.closed == 0 {throw("chansend: spurious wakeup")}// 通道在發送操作完成前被關閉，報錯 ！！！// 這也就是為什么說向一個已經關閉通道寫數據會報錯，原因就在這里panic(plainError("send on closed channel"))}// 發送操作成功完成return true
}

send 函數的作用是將數據直接發送給等待接收的協程，并喚醒等待接收的協程。

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {// 競態檢測if raceenabled {// 無緩沖通道if c.dataqsiz == 0 {// 通知競態檢測器當前操作的上下文racesync(c, sg)} else {// 有緩沖通道// Pretend we go through the buffer, even though// we copy directly. Note that we need to increment// the head/tail locations only when raceenabled.// 即使數據是直接發送的，競態檢測器也會模擬數據通過緩沖區的流程// 調用 racenotify，記錄接收索引（c.recvx）的變化。racenotify(c, c.recvx, nil)racenotify(c, c.recvx, sg)// 更新接收索引 c.recvx 和發送索引 c.sendx，模擬環形緩沖區的行為c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz}}// 如果 sg.elem 不為 nil，調用 sendDirect 將數據從發送方 ep 直接發送到接收方 sg.elemif sg.elem != nil {sendDirect(c.elemtype, sg, ep)// 清空 sg.elem，表示數據已發送sg.elem = nil}// 獲取等待接收的協程gp := sg.g// 調用解鎖函數，釋放通道的鎖unlockf()// 將 sudog 對象傳遞給接收協程，用于后續處理gp.param = unsafe.Pointer(sg)// 標記發送操作成功sg.success = trueif sg.releasetime != 0 {sg.releasetime = cputicks()}// 將 gp 協程標記為可運行狀態，并將其加入調度隊列中等待執行goready(gp, skip+1)
}

sendDirect 函數的作用是將數據直接從發送方的內存位置發送到等待接收的協程 sudog 對象 。

func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {// src is on our stack, dst is a slot on another stack.// Once we read sg.elem out of sg, it will no longer// be updated if the destination's stack gets copied (shrunk).// So make sure that no preemption points can happen between read & use.// 獲取接收方的內存位置，以便將數據直接發送到該位置dst := sg.elem// 觸發寫屏障// 確保垃圾回收器能夠正確追蹤目標內存區域中的指針typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.Size_)// No need for cgo write barrier checks because dst is always// Go memory.// 這是一個底層函數，用于在內存中安全地移動數據// dst 目標地址，src 源地址memmove(dst, src, t.Size_)
}

關于更多內存管理的知識點，請關注后續文章《Golang 內存模型》。

接收數據

在 golang 的運行時中，接收數據操作的代碼會被編譯為對 chanrecv1 的調用。如下所示：

package mainimport "fmt"func main() {ch := make(chan int, 1)ch <- 1data := <-chfmt.Println(data)
}

編譯成匯編代碼如下所示：

0x001a 00026        CALL    runtime.makechan(SB)
# ...
0x0043 00067        CALL    runtime.chanrecv1(SB)

以上匯編代碼只是部分截取，請注意甄別。

而在 chanrecv1 的函數內部，又是對 chanrecv 的調用，如下所示：

源碼位置：src/runtime/chan.go

// entry points for <- c from compiled code.
//
//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {chanrecv(c, elem, true)
}

函數原型

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {// ...
}

• c *hchan：指向目標通道的指針。
• ep unsafe.Pointer：指向接收數據的內存位置。如果為 nil，表示僅檢查通道狀態而不接收數據。
• block bool：是否允許阻塞接收。如果為 false，則在無法立即接收數據時返回。
• selected bool：表示是否成功接收數據。在 select 語句中，用于標記是否選擇了當前通道。
• received bool：表示是否實際接收到數據。如果通道關閉且緩沖區為空，received 為 false。

函數內容

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {// raceenabled: don't need to check ep, as it is always on the stack// or is new memory allocated by reflect.// 如果啟用了通道調試模式，打印調試信息if debugChan {print("chanrecv: chan=", c, "\n")}// 如果通道為 nilif c == nil {if !block {// 非阻塞接收，直接返回return}// 如果是阻塞接收，協程會阻塞并拋出異常gopark(nil, nil, waitReasonChanReceiveNilChan, traceBlockForever, 2)throw("unreachable")}// 如果通道關聯了一個定時器,則調用 maybeRunChan 來處理定時器事件if c.timer != nil {c.timer.maybeRunChan()}// 如果是非阻塞接收且通道為空// Fast path: check for failed non-blocking operation without acquiring the lock.if !block && empty(c) {// After observing that the channel is not ready for receiving, we observe whether the// channel is closed.//// Reordering of these checks could lead to incorrect behavior when racing with a close.// For example, if the channel was open and not empty, was closed, and then drained,// reordered reads could incorrectly indicate "open and empty". To prevent reordering,// we use atomic loads for both checks, and rely on emptying and closing to happen in// separate critical sections under the same lock.  This assumption fails when closing// an unbuffered channel with a blocked send, but that is an error condition anyway.// 檢查通道是否關閉if atomic.Load(&c.closed) == 0 {// Because a channel cannot be reopened, the later observation of the channel// being not closed implies that it was also not closed at the moment of the// first observation. We behave as if we observed the channel at that moment// and report that the receive cannot proceed.// 如果通道未關閉，直接返回return}// The channel is irreversibly closed. Re-check whether the channel has any pending data// to receive, which could have arrived between the empty and closed checks above.// Sequential consistency is also required here, when racing with such a send.// 如果通道已關閉并且為非阻塞并且緩沖區為空if empty(c) {// The channel is irreversibly closed and empty.if raceenabled {raceacquire(c.raceaddr())}// ep 不是空，則清空目標內存 epif ep != nil {typedmemclr(c.elemtype, ep)}// 返回成功接收數據(true)，未實際接收到數據(false)return true, false}}var t0 int64if blockprofilerate > 0 {t0 = cputicks()}// 鎖定通道的互斥鎖，確保操作的原子性lock(&c.lock)// 如果通道已關閉if c.closed != 0 {// 緩沖區為空if c.qcount == 0 {if raceenabled {raceacquire(c.raceaddr())}// 這里是通道已經關閉了，而且緩沖區已經為空// 考慮是否可以采用原子操作來代替這里的 lock/unlock 操作的操作？// ...// 和通道關閉時，發送者的檢測同理// 雖然理論上可以通過原子操作來避免加鎖，但在實際實現中，鎖的使用是為了確保線程安全和一致性，另外，雖然原子操作避免了鎖的開銷，但它們仍然有一定的性能開銷// 即使使用原子操作，也需要確保在檢查 c.closed 和 c.qcount 時不會出現競態條件。例如，如果一個協程在檢查 c.closed 后修改了 c.qcount，可能會導致不一致的行為// 記錄吧！！！// 解鎖unlock(&c.lock)// 清空目標內存 epif ep != nil {typedmemclr(c.elemtype, ep)}// 返回成功接收數據(true)，未實際接收到數據(false)return true, false}// The channel has been closed, but the channel's buffer have data.} else {// 通道未關閉，并且有等待發送的協程// Just found waiting sender with not closed.if sg := c.sendq.dequeue(); sg != nil {// Found a waiting sender. If buffer is size 0, receive value// directly from sender. Otherwise, receive from head of queue// and add sender's value to the tail of the queue (both map to// the same buffer slot because the queue is full).// 調用 recv 函數直接從發送協程接收數據，跳過緩沖區// 解鎖通道recv(c, sg, ep, func() { unlock(&c.lock) }, 3)// 返回成功接收數據(true)，實際接收到數據(true)return true, true}}// 如果緩沖區中有數據if c.qcount > 0 {// Receive directly from queue// 得到緩沖區地址qp := chanbuf(c, c.recvx)if raceenabled {racenotify(c, c.recvx, nil)}// 從緩沖區中讀取數據到目標內存if ep != nil {typedmemmove(c.elemtype, ep, qp)}// 清空緩沖區中的數據typedmemclr(c.elemtype, qp)// 更新接收索引 c.recvxc.recvx++// 如果接收索引 == 緩沖區大小，則從 0 重新開始if c.recvx == c.dataqsiz {c.recvx = 0}// 更新緩沖區計數 c.qcountc.qcount--// 解鎖unlock(&c.lock)// 返回成功接收數據(true)，實際接收到數據(true)return true, true}// 如果是非阻塞接收且緩沖區為空，解鎖通道并返回未成功接收數據(false)，未實際接收到數據(false)if !block {unlock(&c.lock)return false, false}// 下面是進行阻塞接收數據代碼邏輯// no sender available: block on this channel.// 獲取當前協程gp := getg()// 分配一個 sudog 對象mysg := acquireSudog()// 下面是針對 mysg 對象屬性的初始化mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}// No stack splits between assigning elem and enqueuing mysg// on gp.waiting where copystack can find it.mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.isSelect = falsemysg.c = cgp.param = nil// 將協程加入接收等待隊列c.recvq.enqueue(mysg)// 如果通道關聯了定時器，調用 blockTimerChanif c.timer != nil {blockTimerChan(c)}// Signal to anyone trying to shrink our stack that we're about// to park on a channel. The window between when this G's status// changes and when we set gp.activeStackChans is not safe for// stack shrinking.// 標記協程即將阻塞在通道操作上gp.parkingOnChan.Store(true)// 調用 gopark 阻塞當前協程，直到被發送操作喚醒gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2)// 下面是協程被喚醒之后的操作流程// someone woke us up// 檢查協程是否被正確喚醒if mysg != gp.waiting {throw("G waiting list is corrupted")}// 如果通道關聯了定時器，調用 unblockTimerChanif c.timer != nil {unblockTimerChan(c)}// 清理協程狀態gp.waiting = nilgp.activeStackChans = falseif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}// 獲取接收操作的結果success := mysg.successgp.param = nil// 釋放 sudog 對象mysg.c = nilreleaseSudog(mysg)// 返回接收操作結果return true, success
}

recv 函數用于從通道中讀取數據，并將其傳遞給等待接收的協程。

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {// 如果通道無緩沖區if c.dataqsiz == 0 {if raceenabled {racesync(c, sg)}// 并且目標內存位置不是 nil， 則直接拷貝到目標位置if ep != nil {// copy data from senderrecvDirect(c.elemtype, sg, ep)}} else {// 通道有緩沖區// Queue is full. Take the item at the// head of the queue. Make the sender enqueue// its item at the tail of the queue. Since the// queue is full, those are both the same slot.// 計算緩沖區中接收位置的指針qp := chanbuf(c, c.recvx)if raceenabled {racenotify(c, c.recvx, nil)racenotify(c, c.recvx, sg)}// copy data from queue to receiverif ep != nil {// 如果目標內存位置不是 nil， 將數據從緩沖區復制到接收方的內存位置typedmemmove(c.elemtype, ep, qp)}// copy data from sender to queue// 將發送方的數據復制到緩沖區的尾部typedmemmove(c.elemtype, qp, sg.elem)// 增加接收索引并處理環形緩沖區的邊界條件c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}// 更新發送索引，使其指向下一個可用位置c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz}// 清空 sudog 對象中的數據指針sg.elem = nil// 獲取發送協程的指針gp := sg.g// 調用解鎖函數，釋放通道的鎖unlockf()gp.param = unsafe.Pointer(sg)// 標記接收操作成功sg.success = trueif sg.releasetime != 0 {sg.releasetime = cputicks()}// 將 gp 協程標記為可運行狀態，并將其加入調度隊列中等待執行goready(gp, skip+1)
}

recvDirect 函數的作用是從發送方直接接收數據，并將其復制到接收方的目標內存位置。

func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {// dst is on our stack or the heap, src is on another stack.// The channel is locked, so src will not move during this// operation.// 指向發送方的內存位置src := sg.elem// 寫屏障// 確保目標內存位置中的指針被垃圾回收器正確追蹤typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.Size_)// 將數據從發送方的內存位置復制到接收方的內存位置memmove(dst, src, t.Size_)
}

關于更多內存管理的知識點，請關注后續文章《Golang 內存模型》。

關閉通道

在 golang 的運行時中，關閉通道操作的代碼會被編譯為對 closechan 的調用。如下所示：

package mainimport "fmt"func main() {ch := make(chan int, 1)close(ch)
}

編譯成匯編代碼如下所示：

0x001a 00026        CALL    runtime.makechan(SB)
# ...
0x0020 00032        CALL    runtime.closechan(SB)

以上匯編代碼只是部分截取，請注意甄別。

函數原型

源碼位置：src/runtime/chan.go

func closechan(c *hchan) {// ...
}

• c *hchan：指向要關閉的通道的指針。

函數內容

func closechan(c *hchan) {// 如果通道為 nil，直接拋出異常if c == nil {panic(plainError("close of nil channel"))}// 鎖定通道的互斥鎖，確保關閉操作的原子性lock(&c.lock)if c.closed != 0 {// 如果通道已經關閉，解鎖，拋出異常unlock(&c.lock)panic(plainError("close of closed channel"))}if raceenabled {callerpc := getcallerpc()racewritepc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(closechan))racerelease(c.raceaddr())}// 將通道的 closed 標志設置為 1，表示通道已關閉c.closed = 1// 需要被喚醒的協程集合var glist gList// release all readersfor {// 遍歷通道的接收等待隊列，釋放所有等待接收的協程sg := c.recvq.dequeue()if sg == nil {break}// 如果 sg.elem 不為 nilif sg.elem != nil {// 清空接收方的內存位置typedmemclr(c.elemtype, sg.elem)sg.elem = nil}if sg.releasetime != 0 {sg.releasetime = cputicks()}// 獲取到接收者協程gp := sg.ggp.param = unsafe.Pointer(sg)// 標記獲取操作失敗sg.success = falseif raceenabled {raceacquireg(gp, c.raceaddr())}// 將協程加入到 glist 中，稍后喚醒glist.push(gp)}// release all writers (they will panic)for {// 遍歷通道的發送等待隊列，釋放所有等待發送的協程sg := c.sendq.dequeue()if sg == nil {break}// 清空 sg.elem，避免內存泄漏sg.elem = nilif sg.releasetime != 0 {sg.releasetime = cputicks()}// 獲取到發送者協程gp := sg.ggp.param = unsafe.Pointer(sg)// 標記發送操作失敗sg.success = falseif raceenabled {raceacquireg(gp, c.raceaddr())}// 將協程加入到 glist 中，稍后喚醒glist.push(gp)}// 解鎖通道的互斥鎖unlock(&c.lock)// Ready all Gs now that we've dropped the channel lock.// 遍歷 glist，喚醒所有等待的協程for !glist.empty() {gp := glist.pop()gp.schedlink = 0// 調用 goready 將協程標記為可運行狀態goready(gp, 3)}
}

🌺🌺🌺撒花！

如果本文對你有幫助，就點關注或者留個👍
如果您有任何技術問題或者需要更多其他的內容，請隨時向我提問。