5.13. 并發 · Go 中文文檔

## 5.13\. 并發 ### 5.13.1\. 交流來分享并發編程是很大的主題，此處只夠講 Go 方面的要點。很多環境的并發編程變得困難出自于實現正確讀寫共享變量的微妙性。Go 鼓勵一種不一樣的方式，這里，共享變量在信道是傳遞，并且事實上，從來未被獨立的執行序列所共享。每一特定時間只有一個夠程在存取該值。從設計上數據競爭就不會發生。為鼓勵這種思考方式我們把它縮減成一句口號： > **別靠共享內存來通信，要靠通信來分享內存!** 此方式可扯的太遠。例如，引用計數最好靠整型變量外加一個互斥。但最為高層方式，使用信道控制存取可以更容易寫成清楚正確的程序。思考這種模型一種方式是考慮在單一 CPU 上運行的典型的單線程程序，不需用到同步原語。現在多運行一份；也同樣不需同步。現在讓兩者通信；如果通信是同步者，還是不需其它的同步。例如，Unix 的管道完美的適合這個模型。盡管 Go 的并行方式源自 Hoare 的通信順序進程（CSP），它也可視為泛化的類型安全的 Unix 管道。 ### 5.13.2\. Goroutines(Go程) 它們叫做夠程，是因為現有的術語 -- 線程，協程和進程等 -- 傳達的含義不夠精確。夠程的模式很簡單：它是在同一地址空間和其它夠程并列執行的函數。它輕盈，只比分配堆棧空間多費一點兒。因為堆棧開始時很小，所以它們很便宜，只在需要時分配（和釋放）堆庫存。夠程在多個 OS 線程間復用，所以如果某個需要阻塞，例如在等待IO，其它的可繼續執行。它們的設計隱藏了許多線程生成和管理的復雜性。在某個函數或方法前加上 go 鍵字則在新夠程中執行此調用。當調用完成，夠程安靜的退出。（效果類似 Unix shell 的 & -- 在后臺執行命令。） ``` go list.Sort() // run list.Sort in parallel; don't wait for it. ``` 函數字面在實施夠程上很順手： ``` func Announce(message string, delay int64) { go func() { time.Sleep(delay) fmt.Println(message) }() // Note the parentheses - must call the function. } ``` 在 Go 里，函數字面是閉包：實現保證函數引用的變量可以活到它們不在用了。這些例子沒什么用處，因為函數無法通知其結束。那需用到信道。 ### 5.13.3\. Channels(信道) 類型映射，信道是引用類型，使用 make 分配。如果提供了可選的整型參量，它會設置信道的緩沖大小。默認是0，即無緩沖的或同步的信道。 ``` ci := make(chan int) // unbuffered channel of integers cj := make(chan int, 0) // unbuffered channel of integers cs := make(chan *os.File, 100) // buffered channel of pointers to Files ``` 信道結合了通信 -- 即值的交換 -- 和同步 -- 確保兩個計算（夠程）處于某個已知狀態。信道有很多慣用語。先從一個開始。上節我們啟動了個后臺的排序。信道使啟動夠程能等待排序完成。 ``` c := make(chan int) // Allocate a channel. // Start the sort in a goroutine; when it completes, signal on the channel. go func() { list.Sort() c <- 1 // Send a signal; value does not matter. }() doSomethingForAWhile() <-c // Wait for sort to finish; discard sent value. ``` 接收者阻塞到有數據可以接收。如果信道是非緩沖的，發送者阻塞到接收者收到其值。如果信道有緩沖，發送者只需阻塞到值拷貝到緩沖里；如果緩沖滿，則等待直到某個接收者取走一值。一個緩沖信道可以用作信號燈，比如用來限速。下例中，到來請求傳遞給 handle，來發送一值到信道，處理請求，以及才信道接收值。信道的容量決定了可同時調用 process 的數量。 ``` var sem = make(chan int, MaxOutstanding) func handle(r *Request) { sem <- 1 // Wait for active queue to drain. process(r) // May take a long time. <-sem // Done; enable next request to run. } func Serve(queue chan *Request) { for { req := <-queue go handle(req) // Don't wait for handle to finish. } } ``` 同樣的概念，我們可以啟動一定數量的 handle 夠程，全都讀取請求信道。夠程的數量限制同時調用 process 的數量。此 Serve 函數也接受一個信道來告知它退出；夠程啟動后會接收阻塞在此信道。 ``` func handle(queue chan *Request) { for r := range queue { process(r) } } func Serve(clientRequests chan *clientRequests, quit chan bool) { // Start handlers for i := 0; i < MaxOutstanding; i++ { go handle(clientRequests) } <-quit // Wait to be told to exit. } ``` ### 5.13.4\. Channels of channels(信道的信道) Go 的一個最重要的特色是信道作為一等值可以被分配被傳遞，正如其它的值。此特色常用來實現安全并行的分路器。上節的例子里，handle 是個理想化的請求經手者，但我們并未定義其經手的類型。如果此類型包括一個可回發的信道，每個客戶都可提供自身的回答途徑。下面是類型 Request 的語義定義。 ``` type Request struct { args []int f func([]int) int resultChan chan int } ``` 客戶提供一個函數及其參量，以及在請求物件里的一個信道，用來接收答案。 ``` func sum(a []int) (s int) { for _, v := range a { s += v } return } request := &Request{[]int{3, 4, 5}, sum, make(chan int)} // Send request clientRequests <- request // Wait for response. fmt.Printf("answer: %d\n", <-request.resultChan) ``` 服務器端，經手函數是唯一需要改變的。 ``` func handle(queue chan *Request) { for req := range queue { req.resultChan <- req.f(req.args) } } ``` 當然還要很多工作使其實際，但此代碼是個速率限制、并行、非阻塞的 RPC 系統的框架，而且看不到一個互斥。 ### 5.13.5\. 并發這些概念的另一應用是在多 CPU 核上并發計算。如果一個運算可以分解為獨立片段，則可并發，用一信道通知每個片段的結束。比如我們有個很花時間的運算執行在一列項上，每個項的運算值都是獨立的，如下例： ``` type Vector []float64 // Apply the operation to v[i], v[i+1] ... up to v[n-1]. func (v Vector) DoSome(i, n int, u Vector, c chan int) { for ; i < n; i++ { v[i] += u.Op(v[i]) } c <- 1 // signal that this piece is done } ``` 我們在循環里單獨啟動片段，每個 CPU 一個。它們誰先完成都沒關系；我們只是啟動全部夠程前清空信道，再數數結束通知即可。 ``` const NCPU = 4 // number of CPU cores func (v Vector) DoAll(u Vector) { c := make(chan int, NCPU) // Buffering optional but sensible. for i := 0; i < NCPU; i++ { go v.DoSome(i*len(v)/NCPU, (i+1)*len(v)/NCPU, u, c) } // Drain the channel. for i := 0; i < NCPU; i++ { <-c // wait for one task to complete } // All done. } ``` 現在的 gc 實現（6g 等）不會默認的并發此代碼。它只投入一個核給用戶層的運算。任意多的夠程可以阻塞在系統調用上，但默認的每個時刻只有一個可以執行用戶層的代碼。它本該更聰明些，某天它會變聰明，但那之前如果你要并發 CPU 就必須告知運行態你要同時執行代碼的夠程的數量。有兩種相關的辦法，或者你把環境變量GOMAXPROCS 設為你要用到的核數（默認1）；或者導入 runtime 包調用 runtime.GOMAXPROCS(NCPU)。再提一遍，此要求會隨著調度及運行態的進步而退休。 ### 5.13.6\. 漏水緩沖并發編程的工具也可用來使非并發的概念更容易表達。下例是從某個RPC 包里提取的。客戶夠程循環接收數據自某源，可能是網絡。為免分配釋放緩沖，它保有一個自由列，并由一個緩沖的信道代表。如果信道空，則新緩沖被分配。當消息緩沖好時，它在 serverChan 上發給服務器。 ``` var freeList = make(chan *Buffer, 100) var serverChan = make(chan *Buffer) func client() { for { b, ok := <-freeList // grab a buffer if available if !ok { // if not, allocate a new one b = new(Buffer) } load(b) // read next message from the net serverChan <- b // send to server } } ``` 服務器循環讀消息自客戶，處理，返回緩沖到自由列。 ``` func server() { for { b := <-serverChan // wait for work process(b) _ = freeList <- b // reuse buffer if room } } ``` 客戶無阻的從 freeList 得到一個緩沖，如還沒有則客戶分配個新的緩沖。服務器無阻的發送給 freeList，放 b 回自由列，除非列滿，此時緩沖掉到地板上被垃圾收集器回收。（發送操作賦值給空白標識使其無阻但會忽略操作是否成功。）此實現僅用幾行就打造了個漏水緩沖，靠緩沖信道和垃圾收集器記賬。