你一定聽説過 Go 語言所倡導的這個核心併發原則：“不要通過共享內存來通信，而要通過通信來共享內存 (Don't communicate by sharing memory; instead, share memory by communicating)”。這一理念深刻影響了 Go 的併發設計。
本文將具體討論 Go 中的 併發控制機制 (concurrency control mechanisms) ，特別是基於 CSP (Communicating Sequential Processes) 的實現，包括 chan 和 select 等關鍵要素的設計思路及核心實現細節。理解這些內容，對於編寫出高效、安全的 Go 併發程序至關重要。本文假設讀者已經對 Go 的 GPM 調度模型 (GPM scheduling model) 有了比較深入的瞭解。
CSP, Communicating Sequential Processes
令我頗感驚訝的是，CSP 這個併發模型是由計算機科學家 託尼·霍爾 (Tony Hoare) 在 1978 年提出的。在那個個人計算機尚未普及、多核處理器更是遙不可及的年代，學術界和工業界普遍關注的重點是如何在單核處理器上實現有效的任務併發與切換，以及如何管理共享資源帶來的複雜性。
CSP 的核心思想是將獨立的、順序執行的進程作為基本的計算單元。這些進程之間不共享內存，而是通過顯式的 通道 (channels) 來進行通信和同步。一個進程向通道發送消息，另一個進程從該通道接收消息。這種通信方式是同步的，即發送方會阻塞直到接收方準備好接收，或者接收方會阻塞直到發送方發送了消息（對於無緩衝通道而言）。
Go 語言在原生層面通過 chan 的設計，為 CSP 模型提供了強大的支持。這樣做的好處顯而易見：
- 簡化併發邏輯 ：通過將數據在不同
goroutine之間傳遞，而不是共享狀態，極大地降低了併發編程中數據競爭的風險。開發者可以將注意力更多地放在消息的流動和處理上，而不是複雜的鎖機制。 - 清晰的關係 ：在任意時刻，數據要麼屬於某個
goroutine，要麼正在通過chan進行傳遞。這種清晰的關係使得推理程序的行為變得更加容易。 - 可組合性 ：基於
chan的組件更容易組合起來構建更復雜的併發系統。
與主流的併發模型相比，Go 的 CSP 實現展現出其獨特性。
- 對比 Java/pthread 的共享內存模型 ：Java 和 C++ (pthread) 等語言主要依賴共享內存和鎖（如
mutex、semaphore）進行併發控制。這種模型下，開發者需要非常小心地管理對共享數據的訪問，否則極易出現 死鎖 (deadlock) 和 競態條件 (race condition) 。Go 的 CSP 模型通過chan將數據在goroutine間傳遞，避免了直接的內存共享，從而在設計上減少了這類問題。內存同步由chan的操作隱式完成。 - 對比 Actor 模型 ：Actor 模型（如 Akka、Erlang OTP 中的
gen_server）與 CSP 有相似之處，都強調通過消息傳遞進行通信，避免共享狀態。主要區別在於 Actor 通常擁有自己的狀態，並且 Actor 之間的通信是異步的，每個 Actor 一般都有一個郵箱 (mailbox) 來存儲傳入的消息。而 Go 的chan通信可以是同步的（無緩衝chan）或異步的（有緩衝chan）。Go 的goroutine比 Actor 更輕量。 - 對比 JavaScript 的異步回調/Promise ：JavaScript (尤其是在 Node.js 環境中) 採用單線程事件循環和異步回調（或
Promise/async/await）來處理併發。這種方式避免了多線程帶來的複雜性，但在回調層級很深（回調地獄 callback hell）時，代碼可讀性和維護性會下降。Promise和async/await改善了這一點，但其併發的本質仍然是協作式的單任務切換，而非像 Go 那樣可以利用多核進行並行計算的搶佔式調度。
在調度方面，Go 的 goroutine 由 Go 運行時進行調度，是用户態的輕量級線程，切換成本遠低於操作系統線程。chan 的操作天然地與調度器集成，可以高效地掛起和喚醒 goroutine。在公平性方面，select 語句在處理多個 chan 操作時，會通過一定的隨機化策略來避免飢餓問題。Go 的併發原語設計精良，易於組合，使得構建複雜的併發模式成為可能。
關於併發模型的更多更詳細的對比，讀者可以參考 Paul Butcher 的《七週七併發模型 (Seven Concurrency Models in Seven Weeks: When Threads Unravel) 》。雖已在我的書單中，但我也還未完全讀完，歡迎互相交流學習。
chan 具體是什麼
chan 是 Go 語言中用於在不同 goroutine 之間傳遞數據和同步執行的核心類型。它是一種類型化的管道，你可以通過它發送和接收特定類型的值。
我們從一個簡單的 chan 用法開始：
package main
import (
"fmt"
"time"
)
func main() {
// 創建一個字符串類型的無緩衝 channel
messageChannel := make(chan string)
go func() {
// 向 channel 發送數據
messageChannel <- "Hello from goroutine!"
fmt.Println("Sender: Message sent.")
}()
go func() {
// 從 channel 接收數據
time.Sleep(1 * time.Second) // 模擬耗時操作，確保接收者後準備好
receivedMessage := <-messageChannel
fmt.Println("Receiver: Received message:", receivedMessage)
}()
// 等待 goroutine 執行完畢
time.Sleep(2 * time.Second)
fmt.Println("Main: Finished.")
}在這個例子中，make(chan string) 創建了一個可以傳遞 string 類型數據的 chan。messageChannel <- "Hello" 是發送操作，它會將字符串發送到 chan 中。receivedMessage := <-messageChannel 是接收操作，它會從 chan 中讀取數據。對於無緩衝的 chan，發送操作會阻塞，直到另一個 goroutine 對同一個 chan 執行接收操作；反之亦然，接收操作也會阻塞，直到有數據被髮送。
這些簡潔的 chan 操作符實際上是 Go 語言提供的 語法糖 (syntactic sugar) 。在底層，它們會轉換為運行時的內部函數調用。
- 向
chan發送數據ch <- v大致對應於運行時函數runtime.chansend1(ch, v)（具體函數可能因版本和場景略有不同，如chansend）。 - 從
chan接收數據v := <-ch或v, ok := <-ch大致對應於運行時函數runtime.chanrecv1(ch, &v)或runtime.chanrecv2(ch, &v)（返回第二個bool值表示chan是否關閉且已空）。 for v := range ch循環，在底層會持續嘗試從chan接收數據，直到chan被關閉並且緩衝區為空。
要理解 chan 的行為，瞭解其內部數據結構至關重要。在 Go 的運行時中，chan 的內部表示是 runtime.hchan 結構體（位於 src/runtime/chan.go）。其核心字段包括：
// src/runtime/chan.go
type hchan struct {
qcount uint // 當前隊列中剩餘元素個數 (current number of elements in the queue)
dataqsiz uint // 環形隊列的大小，即緩衝區大小 (size of the circular queue, i.e., buffer size)
buf unsafe.Pointer // 指向環形隊列的指針 (pointer to the circular queue buffer)
elemsize uint16 // channel 中元素的大小 (size of an element in the channel)
closed uint32 // 標記 channel 是否關閉 (marks if the channel is closed)
timer *timer // 可能與內部調試或計時器相關的 select 優化有關
elemtype *_type // channel 中元素的類型 (type of an element in the channel)
sendx uint // 發送操作處理到的位置 (index for send operations)
recvx uint // 接收操作處理到的位置 (index for receive operations)
recvq waitq // 等待接收的 goroutine 隊列 (list of goroutines waiting to receive)
sendq waitq // 等待發送的 goroutine 隊列 (list of goroutines waiting to send)
bubble *synctestBubble // 此字段通常僅在開啓了競爭檢測 (`-race`) 或特定的同步測試構建 (`synctest`) 中出現。
// 用於輔助競爭檢測器跟蹤 channel 操作的同步事件，幫助發現潛在的 data race。
// 對於常規的 channel 理解和使用，可以不必關注此字段。
lock mutex // 保護 hchan 中所有字段的鎖 (lock protecting all fields in hchan)
}
type waitq struct { // 是一個雙向鏈表
first *sudog
last *sudog
}qcount：表示當前chan緩衝區中實際存儲的元素數量。dataqsiz：表示chan的緩衝區大小。如果為 0，則該chan是無緩衝的。buf：一個指針，指向底層存儲元素的環形緩衝區。只有在dataqsiz > 0時（即有緩衝chan），這個字段才有意義。closed：一個標誌位，表示chan是否已經被關閉。sendq和recvq：分別是等待發送數據的goroutine隊列和等待接收數據的goroutine隊列。它們是sudog結構體（代表一個阻塞的goroutine）組成的鏈表。lock：一個互斥鎖，用於保護hchan結構體內部字段的併發訪問，確保chan操作的原子性。
當創建一個 chan 時，make(chan T, N)，如果 N 為 0 或省略，則創建的是無緩衝 chan (dataqsiz 為 0，buf 為 nil)。如果 N 大於 0，則創建的是有緩衝 chan (dataqsiz 為 N，並分配相應大小的 buf)。
chan 的併發控制
chan 的併發控制能力是其設計的核心，它緊密地與 Go 的 goroutine 調度器協同工作，以實現高效的同步和通信。
當一個 goroutine 嘗試對 chan 進行操作（發送或接收）時，會首先獲取 hchan 結構體中的 lock 互斥鎖，以保證操作的原子性和數據一致性。
發送操作 (ch <- v) 的邏輯
- 嘗試直接喚醒接收者 ：如果
recvq(等待接收的goroutine隊列) 不為空，説明有goroutine因為嘗試從該chan接收數據而被阻塞。這時，發送操作會直接將數據從發送方goroutine的棧（或堆，取決於數據）複製到該等待的接收方goroutine的指定內存位置，然後喚醒這個接收方goroutine(將其標記為可運行狀態，等待調度器調度執行)。這對於無緩衝chan和緩衝chan空閒時是常見路徑。發送方goroutine通常可以繼續執行。 - 嘗試放入緩衝區 ：如果
recvq為空，但chan有緩衝區 (dataqsiz > 0) 且緩衝區未滿 (qcount < dataqsiz)，發送操作會將數據從發送方複製到buf環形緩衝區中的下一個可用槽位，並增加qcount。發送方goroutine繼續執行。 - 阻塞發送者 ：如果
recvq為空，並且chan是無緩衝的 (dataqsiz == 0)，或者chan是有緩衝的但緩衝區已滿 (qcount == dataqsiz)，那麼發送操作無法立即完成。此時，發送方goroutine會被封裝成一個sudog結構，包含要發送的數據的指針，並加入到hchan的sendq(等待發送的goroutine隊列) 中。隨後，該發送方goroutine會調用gopark函數，釋放P(處理器)，進入 阻塞 (waiting) 狀態，等待被接收方喚醒。
接收操作 (v := <-ch 或 v, ok := <-ch) 的邏輯
嘗試直接從發送者獲取或喚醒發送者 ：如果 sendq (等待發送的 goroutine 隊列) 不為空，説明有 goroutine 因為嘗試向該 chan 發送數據而被阻塞。
- 對於無緩衝
chan：接收操作會直接從sendq中隊首的sudog(阻塞的發送者) 獲取數據，將其複製到接收方goroutine的指定內存位置，然後喚醒這個發送方goroutine。接收方goroutine繼續執行。 - 對於有緩衝
chan(但緩衝區此時為空) ：如果sendq（等待發送的goroutine隊列）不為空，這表明此前因為緩衝區已滿而有發送者goroutine(GS) 被阻塞。現在一個接收者goroutine(GR) 來了，並且緩衝區是空的 (qcount == 0)。此時，接收操作會從sendq中取出第一個等待的發送者GS，將其數據直接複製給當前接收者GR（或者複製到GR預期的內存位置）。然後，發送者GS會被喚醒並可以繼續執行。這個過程可以看作是一次“直接的數據交接”，儘管它是在緩衝chan的上下文中發生的。緩衝區hchan.buf在此特定交互中可能不直接存儲這個傳遞中的數據，或者數據只是邏輯上“通過”了一個緩衝區槽位以保持sendx和recvx索引的一致性。關鍵在於，一個等待的發送者被匹配並喚醒，其數據被成功傳遞。
嘗試從緩衝區獲取 ：如果 sendq 為空，但 chan 有緩衝區 (dataqsiz > 0) 且緩衝區不為空 (qcount > 0)，接收操作會從 buf 環形緩衝區中取出一個元素，複製到接收方 goroutine 的指定內存位置，減少 qcount，並相應地移動 recvx 指針。接收方 goroutine 繼續執行。
處理已關閉的 chan ：如果 chan 已經被關閉 (closed > 0) 並且緩衝區為空 (qcount == 0)：
v := <-ch會立即返回該chan元素類型的零值。v, ok := <-ch會立即返回元素類型的零值和false給ok。
這使得 for v := range ch 循環能夠在 chan 關閉且數據取完後優雅退出。
阻塞接收者 ：如果 sendq 為空，chan 未關閉，並且 chan 是無緩衝的，或者 chan 是有緩衝的但緩衝區為空 (qcount == 0)，那麼接收操作無法立即完成。此時，接收方 goroutine 會被封裝成一個 sudog 結構，並加入到 hchan 的 recvq (等待接收的 goroutine 隊列) 中。隨後，該接收方 goroutine 調用 gopark 進入阻塞狀態，等待被髮送方喚醒。
喚醒機制 ：goroutine 的阻塞 (gopark) 和喚醒 (goready) 是由 Go 運行時調度器核心管理的。當一個 goroutine 因為 chan 操作需要阻塞時，它會釋放當前佔用的 P，其狀態被標記為 _Gwaiting。當條件滿足（例如，數據被髮送到 chan，或有 goroutine 準備好從 chan 接收）時，另一個 goroutine (執行對應 chan 操作的 goroutine) 會調用 goready 將阻塞的 goroutine 的狀態改為 _Grunnable，並將其放入運行隊列，等待調度器分配 P 來執行。
有緩衝 vs 無緩衝舉例
-
無緩衝
chan(make(chan int))- 發送者
ch <- 1會阻塞，直到接收者<-ch準備好。它們必須“握手”。 - 這常用於強同步，確保消息被處理。
- 發送者
-
有緩衝
chan(make(chan int, 1))- 發送者
ch <- 1可以立即完成（只要緩衝區未滿），不需要等待接收者。 - 如果緩衝區滿了，比如
ch <- 1之後再ch <- 2(假設容量為1)，第二個發送者會阻塞。 - 這允許一定程度的解耦和流量削峯。
- 發送者
chan 通信的本質 ： chan 通信的本質仍然是 內存複製 。無論是直接在發送者和接收者 goroutine 之間傳遞，還是通過緩衝區中轉，元素的值都會從源位置複製到目標位置。對於指針或包含指針的複雜類型，複製的是指針值本身，而不是指針指向的數據。這意味着如果傳遞的是一個大數據結構的指針，實際複製的開銷很小，但需要注意共享數據帶來的併發問題（儘管 CSP 的理念是避免共享）。
關閉一個有數據的 chan
當一個有數據的 chan 被 close(ch) 時：
- 後續的發送操作
ch <- v會引發panic。 - 接收操作
<-ch會繼續從緩衝區讀取剩餘的值，直到緩衝區為空。 - 當緩衝區為空後，接收操作
v := <-ch會立即返回元素類型的零值。 - 接收操作
v, ok := <-ch會返回元素類型的零值和false。
Go 通過 hchan 的 closed 標誌和 qcount 來精確控制這些行為，確保 for v := range ch 循環在 chan 關閉且緩衝區耗盡後能夠自動、優雅地退出，因為此時 chanrecv 操作會返回 (zeroValue, false)，range 機制檢測到 ok 為 false 就會終止循環。
原子操作 ：hchan 內部的關鍵字段（如 qcount, closed, sendx, recvx 以及對 sendq 和 recvq 鏈表的操作）的訪問和修改，都受到 hchan.lock 這個互斥鎖的保護。因此，從外部視角看，對 chan 的發送、接收和關閉操作都可以認為是 原子性的 (atomic) ，它們要麼完整執行，要麼不執行（例如，在嘗試獲取鎖時被阻塞）。這種原子性是由 Go 運行時的鎖機制來保證的，而非硬件層面的原子指令直接作用於整個 chan 操作（儘管鎖的實現本身會用到硬件原子操作）。
select 語言層面原生的多路複用
select 語句是 Go 語言中實現併發控制的另一個強大工具，它允許一個 goroutine 同時等待多個通信操作。select 會阻塞，直到其中一個 case（通信操作）可以執行，然後執行該 case。如果多個 case 同時就緒，select 會 偽隨機地 (pseudo-randomly) 選擇一個執行，以保證公平性，避免某些 chan 總是優先得到處理。
基本用法
ch1 := make(chan int)
ch2 := make(chan string)
// ... goroutines to send to ch1 and ch2
select {
case val1 := <-ch1:
fmt.Printf("Received from ch1: %d\n", val1)
case str2 := <-ch2:
fmt.Printf("Received from ch2: %s\n", str2)
case ch1 <- 10: // 也可以包含發送操作
fmt.Println("Sent 10 to ch1")
default: // 可選的 default case
fmt.Println("No communication was ready.")
// default 會在沒有任何 case 就緒時立即執行，使 select 非阻塞
}底層實現 ：當 Go 代碼執行到一個 select 語句時，編譯器和運行時會協同工作。
- 收集
case：編譯器會生成代碼，將select語句中的所有case（每個case對應一個chan的發送或接收操作）收集起來，形成一個scase(select case) 結構數組。每個scase包含了操作的類型（發送/接收）、目標chan以及用於接收/發送數據的內存地址。 - 亂序處理 ：為了保證公平性，運行時會先對這些
scase進行一個隨機的排序（通過select_order數組）。 -
輪詢檢查 ：按照亂序後的順序，運行時會遍歷所有的
case，檢查對應的chan是否已經就緒（即是否可以立即執行發送或接收操作而不會阻塞）。- 發送操作 ：檢查
chan是否有等待的接收者，或者其緩衝區是否有空間。 - 接收操作 ：檢查
chan是否有等待的發送者，或者其緩衝區是否有數據，或者chan是否已關閉。
- 發送操作 ：檢查
- 立即執行 ：如果在此輪詢過程中發現有任何一個
case可以立即執行，運行時會選擇第一個（按照亂序後的順序）就緒的case，執行相應的chan操作（發送或接收數據），然後跳轉到該case對應的代碼塊執行。select語句結束。 default處理 ：如果在輪詢所有case後沒有發現任何一個可以立即執行，並且select語句包含default子句，那麼default子句的代碼塊會被執行。select語句結束。default使得select可以成為一種非阻塞的檢查機制。-
阻塞與喚醒 ：如果輪詢後沒有
case就緒，且沒有default子句，那麼當前goroutine就需要阻塞。- 對於每一個
case中的chan，運行時會將當前goroutine（表示為一個sudog）加入到該chan的sendq或recvq等待隊列中，並記錄下是哪個case把它加入的。 - 然後，當前
goroutine調用gopark進入阻塞狀態，等待被喚醒。 - 當任何一個被
select監聽的chan發生狀態變化（例如，有數據發送進來，或有goroutine嘗試接收，或chan被關閉），並且這個變化使得某個case的條件滿足時，操作該chan的goroutine會負責喚醒因select而阻塞的goroutine。 - 被喚醒的
goroutine會再次檢查哪個case導致了喚醒（通過sudog中記錄的hchan信息），然後執行該case。在執行選中的case之前，一個關鍵步驟是 將該goroutine的sudog從所有其他未被選中的case所對應的chan的等待隊列 (sendq或recvq) 中移除 。
- 對於每一個
但是，移除操作時間複雜度是怎樣的？
實際上，hchan 中的 sendq 和 recvq (即 waitq 結構) 都是 雙向鏈表 (doubly linked lists) 。sudog 結構體自身包含了指向其在鏈表中前一個和後一個 sudog 的指針 (prev 和 next)。當 select 語句決定喚醒一個 goroutine 時，它已經擁有了指向該 goroutine 的 sudog 的指針。對於那些未被選中的 case，select 機制會遍歷這些 case，並針對每個 case 對應的 chan，利用已知的 sudog 指針以及其 prev 和 next 指針，在 O(1) 時間複雜度內將其從該 chan 的等待隊列中移除（unlinking 操作）。因此，整個清理過程的複雜度與 select 語句中 case 的數量成正比（即 O(N_cases)，其中 N_cases 是 select 中的 case 數量），而不是與等待隊列的實際長度成正比，這保證了 select 機制在處理多個 case 時的效率。
核心算法流程 ：select 的核心可以概括為 runtime.selectgo 函數（位於 src/runtime/select.go）。這個函數實現了上述的收集、亂序、輪詢、阻塞和喚醒邏輯。
它首先嚐試一個“非阻塞”的輪詢，看是否有 case 能夠立即成功。如果找不到，並且沒有 default，它會將當前 goroutine 註冊到所有相關 chan 的等待隊列中，然後 gopark。當其他 goroutine 對這些 chan 操作並喚醒當前 goroutine 時，selectgo 會被重新調度執行，確定哪個 case 被觸發，完成數據交換，並從其他 chan 的等待隊列中清理當前 goroutine。
公平性 ：select 的公平性主要通過兩方面保證：
- 隨機輪詢順序 ：在檢查哪些
case可以執行時，select並不是固定地從第一個case檢查到最後一個，而是引入了一個隨機化的順序。這意味着如果同時有多個case就緒，它們被選中的概率是均等的，避免了排在前面的case總是優先響應。 - 喚醒機制 ：當一個
goroutine因select阻塞後，任何一個使其case成立的chan操作都可以將其喚醒。
這種設計使得 select 在處理多個併發事件源時，能夠公平地響應，而不會因為 case 的書寫順序導致某些事件被餓死。
select 中多個 chan 與死鎖
select 語句本身是一種避免在多個通道操作中選擇時發生死鎖的機制。它會選擇一個 可以立即執行 的 case（發送或接收），如果多個 case 同時就緒，它會偽隨機選擇一個。如果沒有 case 就緒且沒有 default 子句，則執行 select 的 goroutine 會阻塞，直到至少一個 case 變得可以執行。
然而，雖然 select 本身旨在處理多路通道的就緒選擇，但它並不能完全阻止整個程序級別的死鎖。死鎖的發生通常是由於程序中 goroutine 之間形成了循環等待依賴關係，而 select 語句可能成為這種循環依賴的一部分：
所有通信方均阻塞
如果一個 select 語句等待的多個 chan，其對應的發送方或接收方 goroutine 也都因為其他原因被阻塞，並且無法再對這些 chan 進行操作，那麼這個 select 語句可能會永久阻塞。如果這種情況導致程序中所有 goroutine 都無法繼續執行，Go 運行時會檢測到這種全局死鎖，並通常會 panic，打印出 "fatal error: all goroutines are asleep - deadlock!"。
循環依賴
假設有兩個 goroutine，G1 和 G2，以及兩個 chan，chA 和 chB。
- G1 執行
select，其中一個case是從 chA 接收，另一個case是向 chB 發送。 - G2 執行
select，其中一個case是從 chB 接收，另一個case是向 chA 發送。
如果 G1 選擇了等待從 chA 接收，它就需要 G2 向 chA 發送。同時，如果 G2 選擇了等待從 chB 接收，它就需要 G1 向 chB 發送。如果它們都做出了這樣的選擇（或者沒有其他路徑可以走），並且沒有其他 goroutine 來打破這個僵局，那麼 G1 和 G2 就會相互等待，形成死鎖。
基於 hchan.lock 地址排序加鎖
這個策略用在 runtime.selectgo 函數（位於 src/runtime/select.go）中。
背景與問題 ：select 語句可能涉及多個 chan。每個 hchan 結構體內部都有一個互斥鎖 lock，用於保護其內部狀態（如緩衝區、等待隊列 sendq 和 recvq 等）的併發訪問。
當一個 goroutine 執行 select 語句並且沒有 case能立即執行（也沒有 default），它需要將自己（表示為一個 sudog 結構）掛載到所有相關 case 對應的 chan 的等待隊列上。這個掛載操作以及後續可能的摘除操作，都需要獲取相應 hchan 的 lock。
如果 selectgo 在嘗試獲取多個 hchan 的鎖時，沒有一個固定的、全局一致的順序，就可能發生死鎖。例如：
goroutine1 的select涉及chanA和chanB，它嘗試先鎖chanA再鎖chanB。goroutine2 的select（或對這些chan的其他併發操作）也涉及chanA和chanB，但它嘗試先鎖chanB再鎖chanA。
如果 G1 成功鎖定了 chanA 並等待 chanB，同時 G2 成功鎖定了 chanB 並等待 chanA，那麼 G1 和 G2 之間就會因為爭奪這些 hchan.lock 而發生死鎖。這與經典的哲學家就餐問題中的死鎖場景類似。
解決方案：按鎖地址排序。 為了防止這種因獲取 hchan.lock 順序不一致而導致的死鎖，selectgo 函數在需要同時操作多個 hchan（比如，將 goroutine 註冊到它們的等待隊列，或者從等待隊列中移除）時，會執行以下步驟：
- 收集
hchan：首先，它會收集select語句中所有case涉及的hchan指針。 - 排序
hchan：然後，它會根據這些hchan結構體的 內存地址 對它們進行排序。通常是按地址從小到大的順序。由於每個hchan內部的lock字段是其一部分，按hchan地址排序等效於按hchan.lock的地址排序（只要lock字段在hchan結構中的偏移是固定的）。 - 順序加鎖 ：
selectgo會嚴格按照這個排好序的順序來依次獲取每個hchan的lock。 - 執行操作 ：在所有需要的鎖都成功獲取後，再執行相應的操作（如修改等待隊列）。
- 順序解鎖 ：操作完成後，通常以與加鎖相反的順序釋放這些鎖。
通過確保所有需要同時鎖定多個 hchan 的代碼路徑（主要是 selectgo）都遵循相同的“按地址排序後加鎖”的規則，Go 運行時避免了在 hchan 鎖這個層級上發生死鎖。這是一種經典的資源分級（resource hierarchy）或鎖排序（lock ordering）死鎖預防技術。
這個機制確保了 select 在管理其與多個通道的複雜交互時，不會因為內部鎖的爭奪順序問題而陷入困境。
類型系統做到“讀寫分離”
Go 語言的類型系統為 chan 提供了一種優雅的方式來實現“讀寫分離”，即限制對 chan 的操作權限。這是通過 單向 chan (unidirectional channels) 實現的。
一個普通的 chan T 是雙向的，既可以發送數據，也可以接收數據。但我們可以將其轉換為單向 chan：
chan<- T(send-only channel) ：表示一個只能發送T類型數據的chan。你不能從一個chan<- T類型的chan中接收數據。<-chan T(receive-only channel) ：表示一個只能接收T類型數據的chan。你不能向一個<-chan T類型的chan發送數據。
本質與實現
單向 chan 並不是一種全新的 chan 類型。它們本質上是對同一個底層雙向 chan 的不同“視圖”或“接口”。當你將一個 chan T 賦值給一個 chan<- T 或 <-chan T 類型的變量時，並沒有創建新的 chan 結構，只是限制了通過該變量可以對 chan 進行的操作。
這種限制是在 編譯期 (compile-time) 由 Go 的類型檢查器強制執行的。如果你嘗試對一個 chan<- T 進行接收操作，或者對一個 <-chan T 進行發送操作，編譯器會報錯。
例如：
package main
import "fmt"
// sender 函數接受一個只能發送的 chan
func sender(ch chan<- string, message string) {
ch <- message
// msg := <-ch // 編譯錯誤: invalid operation: cannot receive from send-only channel ch (variable of type chan<- string)
}
// receiver 函數接受一個只能接收的 chan
func receiver(ch <-chan string) {
msg := <-ch
fmt.Println("Received:", msg)
// ch <- "pong" // 編譯錯誤: invalid operation: cannot send to receive-only channel ch (variable of type <-chan string)
}
func main() {
myChannel := make(chan string, 1)
// 傳遞給 sender 時，myChannel 被隱式轉換為 chan<- string
go sender(myChannel, "ping")
// 傳遞給 receiver 時，myChannel 被隱式轉換為 <-chan string
receiver(myChannel)
// 也可以顯式轉換
var sendOnlyChan chan<- string = myChannel
var recvOnlyChan <-chan string = myChannel
sendOnlyChan <- "hello again"
fmt.Println(<-recvOnlyChan)
}技巧與注意事項
- API 設計 ：在設計函數或方法時，如果一個
chan參數僅用於發送數據，應將其類型聲明為chan<- T；如果僅用於接收數據，則聲明為<-chan T。這使得函數的意圖更加清晰，並能在編譯期防止誤用。這是 Go 語言中一種重要的封裝和抽象手段。 - 所有權 ：通常，創建
chan的goroutine擁有其“寫”端，並將“讀”端（或雙向chan）傳遞給其他goroutine。或者，一個生產者goroutine創建chan，並將其作為<-chan T返回給消費者，這樣生產者負責寫入和關閉，消費者只負責讀取。 - 關閉
chan：一個重要的規則是：只應該由發送者關閉chan，而不應該由接收者關閉 。因為接收者無法知道是否還有其他發送者會向該chan發送數據。如果一個chan被關閉，而發送者仍然嘗試向其發送數據，會導致panic。將chan的寫端權限（chan T或chan<- T）限定在負責發送和關閉的goroutine中，有助於遵守這一規則。 - 類型轉換 ：一個雙向
chan T可以被隱式或顯式地轉換為chan<- T或<-chan T。但是，單向chan不能被轉換回雙向chan，也不能在不同方向的單向chan之間直接轉換（例如，chan<- T不能直接轉為<-chan T）。
通過這種方式，Go 的類型系統在編譯階段就幫助開發者構建更安全、更易於理解的併發程序，有效地體現了最小權限原則。
常見併發模式參考
利用 chan 和 select，Go 語言可以優雅地實現許多經典的併發模式。
首先，關於 for v := range ch 循環，它確實是處理 chan 接收的一種便捷的語法糖。其本質等價於：
for {
v, ok := <-ch
if !ok { // 如果 chan 被關閉且已空, ok 會是 false
break // 退出循環
}
// ... 使用 v ...
}range 循環會自動處理檢查 ok 狀態的邏輯，使得代碼更簡潔。
接下來介紹一些常見的基於 chan 和 select 的併發模式：
1. 扇入 (Fan-in)
扇入模式是將多個輸入 chan 合併到一個輸出 chan 中。這常用於將多個生產者產生的數據彙總給一個消費者。
package main
import (
"fmt"
"sync"
"time"
)
func produce(id int, ch chan<- string) {
for i := 0; i < 3; i++ {
msg := fmt.Sprintf("Producer %d: Message %d", id, i)
ch <- msg
time.Sleep(time.Millisecond * time.Duration(id*100)) // 模擬不同生產速度
}
}
func fanIn(inputs ...<-chan string) <-chan string {
out := make(chan string)
var wg sync.WaitGroup
for _, inputChan := range inputs {
wg.Add(1)
go func(ch <-chan string) {
defer wg.Done()
for val := range ch {
out <- val
}
}(inputChan)
}
go func() {
wg.Wait() // 等待所有輸入 goroutine 完成
close(out) // 然後關閉輸出 channel
}()
return out
}
func main() {
ch1 := make(chan string)
ch2 := make(chan string)
ch3 := make(chan string)
go produce(1, ch1)
go produce(2, ch2)
go produce(3, ch3)
// 啓動後立即關閉，因為 produce 函數內部會發送數據然後 producer goroutine 結束
// fanIn 需要知道何時停止，這裏通過關閉輸入 ch 實現
// 實際應用中，關閉時機需要仔細設計
go func() { time.Sleep(1 * time.Second); close(ch1) }()
go func() { time.Sleep(1 * time.Second); close(ch2) }()
go func() { time.Sleep(1 * time.Second); close(ch3) }()
mergedOutput := fanIn(ch1, ch2, ch3)
for msg := range mergedOutput {
fmt.Println("Main received:", msg)
}
fmt.Println("All messages processed.")
}Main received: Producer 3: Message 0
Main received: Producer 1: Message 0
Main received: Producer 2: Message 0
Main received: Producer 1: Message 1
Main received: Producer 2: Message 1
Main received: Producer 1: Message 2
Main received: Producer 3: Message 1
Main received: Producer 2: Message 2
Main received: Producer 3: Message 2
All messages processed.在 fanIn 函數中，為每個輸入 chan 啓動一個 goroutine，將接收到的數據轉發到統一的 out 通道。使用 sync.WaitGroup 來確保在所有輸入 chan 都被處理完畢（通常是它們的生產者關閉了它們，導致 range 循環退出）後，再關閉 out 通道。
2. 工作池 (Worker Pool)
工作池模式通過啓動固定數量的 goroutine (workers) 來處理來自一個任務 chan 的任務，並將結果發送到一個結果 chan。這可以控制併發數量，防止資源耗盡。
package main
import (
"fmt"
"sync"
"time"
)
type Task struct {
ID int
Input int
}
type Result struct {
TaskID int
Output int
}
func worker(id int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
fmt.Printf("Worker %d started\n", id)
for task := range tasks {
fmt.Printf("Worker %d processing task %d with input %d\n", id, task.ID, task.Input)
time.Sleep(time.Millisecond * 100) // 模擬工作
results <- Result{TaskID: task.ID, Output: task.Input * 2}
}
fmt.Printf("Worker %d finished\n", id)
}
func main() {
numTasks := 10
numWorkers := 3
tasks := make(chan Task, numTasks)
results := make(chan Result, numTasks)
var wg sync.WaitGroup
// 啓動 workers
for i := 1; i <= numWorkers; i++ {
wg.Add(1)
go worker(i, tasks, results, &wg)
}
// 分發任務
for i := 1; i <= numTasks; i++ {
tasks <- Task{ID: i, Input: i}
}
close(tasks) // 所有任務已發送，關閉 tasks channel，worker 會在處理完後退出
// 等待所有 worker 完成
// 需要一個 goroutine 來等待 wg.Wait() 然後關閉 results channel
// 否則主 goroutine 在收集結果時會死鎖
go func() {
wg.Wait()
close(results)
}()
// 收集結果
for result := range results {
fmt.Printf("Main: Received result for task %d -> %d\n", result.TaskID, result.Output)
}
fmt.Println("All tasks processed.")
}Worker 3 started
Worker 3 processing task 1 with input 1
Worker 2 started
Worker 2 processing task 2 with input 2
Worker 1 started
Worker 1 processing task 3 with input 3
Worker 2 processing task 5 with input 5
Worker 3 processing task 6 with input 6
Worker 1 processing task 4 with input 4
Main: Received result for task 3 -> 6
Main: Received result for task 2 -> 4
Main: Received result for task 1 -> 2
Worker 2 processing task 8 with input 8
Worker 3 processing task 9 with input 9
Worker 1 processing task 7 with input 7
Main: Received result for task 4 -> 8
Main: Received result for task 5 -> 10
Main: Received result for task 6 -> 12
Worker 3 processing task 10 with input 10
Worker 2 finished
Worker 1 finished
Main: Received result for task 9 -> 18
Main: Received result for task 8 -> 16
Main: Received result for task 7 -> 14
Worker 3 finished
Main: Received result for task 10 -> 20
All tasks processed.3. 超時與取消 (Timeout and Cancellation)
select 語句非常適合處理操作超時。可以使用 time.After 創建一個在指定時間後發送信號的 chan。
package main
import (
"fmt"
"time"
)
func longOperation(done chan<- bool) {
time.Sleep(3 * time.Second) // 模擬耗時操作
done <- true
}
func main() {
operationDone := make(chan bool)
go longOperation(operationDone)
select {
case <-operationDone:
fmt.Println("Operation completed successfully!")
case <-time.After(2 * time.Second): // 設置2秒超時
fmt.Println("Operation timed out!")
}
// Cancellation example using a done channel
// (More complex cancellation often uses context.Context)
quit := make(chan struct{}) // struct{} 作為信號，不佔用額外內存
worker := func(q <-chan struct{}) {
for {
select {
case <-q:
fmt.Println("Worker: told to quit. Cleaning up.")
// Do cleanup
fmt.Println("Worker: finished.")
return
default:
// Do work
fmt.Println("Worker: working...")
time.Sleep(500 * time.Millisecond)
}
}
}
go worker(quit)
time.Sleep(2 * time.Second)
fmt.Println("Main: Signaling worker to quit.")
close(quit) // 關閉 quit channel 作為取消信號
time.Sleep(1 * time.Second) // 給 worker 一點時間退出
fmt.Println("Main: Exiting.")
}Operation timed out!
Worker: working...
Worker: working...
Worker: working...
Worker: working...
Main: Signaling worker to quit.
Worker: told to quit. Cleaning up.
Worker: finished.
Main: Exiting.對於更復雜的取消場景，尤其是涉及多個 goroutine 協作時，Go 推薦使用 context.Context 包，它提供了更結構化的方式來傳遞取消信號、截止時間等。
4. 節流 (Throttling) 與 背壓 (Backpressure)
節流 ：限制操作的速率。可以使用 time.Ticker 或一個帶緩衝的 chan 作為令牌桶。
package main
import (
"fmt"
"time"
)
func main() {
requests := make(chan int, 5) // 假設有5個請求要處理
for i := 1; i <= 5; i++ {
requests <- i
}
close(requests)
limiter := time.NewTicker(500 * time.Millisecond) // 每500ms允許一個操作
defer limiter.Stop()
for req := range requests {
<-limiter.C // 等待 limiter 發送信號
fmt.Printf("Processing request %d at %v\n", req, time.Now().Format("15:04:05.000"))
}
fmt.Println("All requests processed.")
}Processing request 1 at 22:44:20.729
Processing request 2 at 22:44:21.227
Processing request 3 at 22:44:21.728
Processing request 4 at 22:44:22.227
Processing request 5 at 22:44:22.732
All requests processed.背壓 ：當消費者處理不過來時，通過阻塞生產者或減少生產速率來反向施加壓力。有緩衝 chan 本身就提供了一種簡單的背壓機制：當緩衝區滿時，發送者會阻塞。更復雜的背壓可能需要監控隊列長度並動態調整。
5. 令牌桶算法 (Token Bucket)
使用一個帶緩衝的 chan 來實現令牌桶，控制對某個資源的訪問速率。
package main
import (
"fmt"
"time"
)
type TokenLimiter struct {
tokenBucket chan struct{}
}
func NewTokenLimiter(capacity int, fillInterval time.Duration) *TokenLimiter {
bucket := make(chan struct{}, capacity)
// Initially fill the bucket
for i := 0; i < capacity; i++ {
bucket <- struct{}{}
}
limiter := &TokenLimiter{
tokenBucket: bucket,
}
// Goroutine to refill tokens periodically
go func() {
ticker := time.NewTicker(fillInterval)
defer ticker.Stop()
for range ticker.C {
select {
case limiter.tokenBucket <- struct{}{}:
// Token added
default:
// Bucket is full, do nothing
}
}
}()
return limiter
}
func (tl *TokenLimiter) Allow() bool {
select {
case <-tl.tokenBucket:
return true // Got a token
default:
return false // No token available
}
}
func (tl *TokenLimiter) WaitAndAllow() {
<-tl.tokenBucket // Wait for a token
}
func main() {
// Allow 2 operations per second, bucket capacity 5
limiter := NewTokenLimiter(5, 500*time.Millisecond) // capacity, fill one token every 500ms
for i := 1; i <= 10; i++ {
// Non-blocking attempt
// if limiter.Allow() {
// fmt.Printf("Request %d allowed at %s\n", i, time.Now().Format("15:04:05.000"))
// } else {
// fmt.Printf("Request %d denied at %s\n", i, time.Now().Format("15:04:05.000"))
// }
// Blocking attempt
limiter.WaitAndAllow()
fmt.Printf("Request %d processed at %s\n", i, time.Now().Format("15:04:05.000"))
// Simulate some work so the timing is observable
// If no work, all will seem to pass quickly after initial burst
if i < 5 { // First 5 might go through quickly due to initial capacity
time.Sleep(100 * time.Millisecond)
} else {
time.Sleep(600 * time.Millisecond) // Make it slower than fill rate to see blocking
}
}
fmt.Println("All operations attempted.")
}// Non-blocking attempt
Request 1 allowed at 22:53:00.261
Request 2 allowed at 22:53:00.265
Request 3 allowed at 22:53:00.265
Request 4 allowed at 22:53:00.265
Request 5 allowed at 22:53:00.265
Request 6 denied at 22:53:00.265
Request 7 denied at 22:53:00.265
Request 8 denied at 22:53:00.265
Request 9 denied at 22:53:00.265
Request 10 denied at 22:53:00.265
All operations attempted.// Blocking attempt
Request 1 processed at 22:51:00.763
Request 2 processed at 22:51:00.868
Request 3 processed at 22:51:00.968
Request 4 processed at 22:51:01.073
Request 5 processed at 22:51:01.175
Request 6 processed at 22:51:01.775
Request 7 processed at 22:51:02.377
Request 8 processed at 22:51:02.979
Request 9 processed at 22:51:03.583
Request 10 processed at 22:51:04.185
All operations attempted.這些模式只是冰山一角，Go 的 chan 和 select 提供了構建各種複雜併發系統的基礎模塊。理解它們的行為和組合方式是掌握 Go 併發編程的關鍵。
