《Go 語言併發之道》讀後感 - 第四章
阿新 • • 發佈：2021-01-23
# 約束
約束可以減輕開發者的認知負擔以便寫出有更小臨界區的併發程式碼。確保某一資訊再併發過程中僅能被其中之一的程序進行訪問。程式中通常存在兩種可能的約束：特定約束和詞法約束。
## 特定約束
通過公約實現約束，無論是由語言社群、你所在的團隊，還是你的程式碼庫設定。在 Go 語言官方預設安裝 gofmt 去格式化你的程式碼，爭取讓大家都寫一樣的程式碼
## 詞法約束
設計使用詞法作用域僅公開用於多個併發程序的正確資料和併發原語，這使得做錯事是不可能的，例如：Go 中 goroutine 和 channel ，而不是使用 Thread 包（無論是官方，第三方）。在 Go 的世界裡作業系統執行緒不用程式設計師管理，需要併發 go 就可以了。
# for-select 迴圈
在 Go 語言中你經常看到 for-select 迴圈。它的結構類似這樣的
```go
for{ // 無限迴圈或者用 range 語句迴圈
select {
// 使用 channel 的任務
}
}
```
## 向 channel 傳送資料
```go
for _,v := range []string{"jisdf","jisdf","ier"}{
select {
case <- done:
return
case stringChan <- v:
// 做些什麼
}
}
```
## 迴圈等待停止
```go
// 第一種保持 select 語句儘可能短：
// 如果完成的 channel 未關閉，我們將退出 select 語句並繼續執行 for 迴圈
for {
select {
case <- done:
return
default:
}
// 非搶佔業務
}
// 第二種將工作嵌入 select 的 default 中
// 如果完成的 channel 尚未關閉，則執行 default 內容的任務
for {
select {
case <- done:
return
default:
// 非搶佔業務
}
}
```
# 防止 goroutine 洩露
執行緒安全，是每一個程式設計師經常討論的話題。 在 Go 中對應的是 goroutine 協程，雖然 goroutine 開銷非常小，非常廉價，但是過多的 goroutine 未得到釋放或終止，也是會消耗資源的。goroutine 有以下幾種方式被終止：
- 當它完成了它的工作。
- 因為不可恢復的錯誤，它不能繼續工作。
- 當它被告知需要終止工作。
前兩種方式非常簡單明瞭，並且隱含在你的程式中。那麼我們如何來取消工作？Go 程式在執行時預設會有一個主 goroutine （main goroutine），他會將一些沒有工作的 goroutine 設定為自旋，這會導致記憶體利用率的下降。思考下，既然 main goroutine 能夠將其他 goroutine 設定自旋，那麼它能不能通知其他 goroutine 停止或退出呢？Of sure ，首先我們需要一個 channel 輔助 main goroutine，它可以包含多種指令，例如超時、異常、特定條件等 。它通常被命名為 done，並且只讀。舉個例子：
```go
doWork := func(done <- chan int ,s <-chan string) <-chan s{
terminated := make(chan int)
go func () {
// 當前函式 return 後列印一條資訊用於驗證，for {} 死迴圈是否被終止
defer fmt.Println("doWork exited")
defer close(termainted)
for {
select {
case l := <- s:
fmt.Println(l)
case <- done: // 由於 select 會相對均勻的挑選 case ，當 done 被讀取，則 return 跳出整個併發
return
}
}
}()
return terminated
}
// 建立控制併發的 channel done
done := make(chan int)
terminated := doWork(done, "a")
// 啟動一個 goroutine 在 1s 後關閉 done channel
go func() {
time.Sleep(1 * time.Second)
fmt.Println("取消工作的 goroutine")
close(done)
}()
// main goroutine 中讀出 termainated 中的資料，驗證我們是否成功通知工作的 goroutine 終止工作
<- terminated
fmt.Println("Done")
```
當一個 goroutine 阻塞了向channel 進行寫入的請求，我們可以這樣做：
```go
newRandstream := func(done <-chan interface{}) <- chan int{
randStream := make(chan int)
go func(){
defer fmt.Println("newRanstream 關閉了")
defer close(randStream)
for{
select {
case randStream <- rand.int():
case <-done:
return
}
}
}()
return
}
done := make(chan interface{})
randStream := newRandStream(done)
fmt.Println("遍歷三次")
for i := 1; i<=3;i++{
fmt.Println("%d: %d\n",i,<-randStream)
}
close(done)
// 模擬正在進行的工作，暫停 1s
time.Sleap(1 * time.Second)
```
# or-channel
以上部分我們瞭解到單一條件下如何取消 goroutine 防止洩露。如果我們有多種條件觸發取消 goroutine ，我們要怎麼辦呢？讓我來了解下 or-channel，建立一個複合 done channel 來處理這種複雜情況。
我們以使用更多的 goroutine 為代價，實現了簡潔性。f(x)=x/2 ，其中 x 是 goroutine 的數量，但你要記住 Go 語言種的一個優點就是能夠快速建立，排程和執行 goroutine ，並且該語言積極鼓勵使用 goroutine 來正確建模問題。不必擔心在這裡建立的 goroutine 的數量可能是一個不成熟的優化。此外，如果在編譯時你不知道你正在使用多少個 done channel ，則將會沒有其他方式可以合併 done channel。
# 錯誤處理
說到錯誤處理，也許很多程式程式設計師覺得 Go 語言錯誤處理簡直太糟糕了。漫天的 `if err != nil{}` ，try catch 捕捉並列印錯誤多麼好。我要說首先我們需要注意 Go 的併發模式，與其他語言有著很大的區別。Go 專案開發者希望我們將錯誤視為一等公民，合併入我們定義的訊息體內，channel 中的資料被讀出的時候我們進行判斷，程式併發過程中是否出現錯誤。這避免了多程序多執行緒模型下，try catch 丟失一些報錯，在故障回顧的時候非常麻煩。
```go
// 建議的訊息體
type MyMessage struct{
Data string
Err error
}
```
讓錯誤成為一等公民合併進你的結構體中，程式碼也許會更易懂
```go
type MyMessage struct{
N int
Err error
}
func myfuncation(n string) MyMessage{
var mm MyMessage
mm.N,mm.Err = anotherFunc(n)
return mm
}
func anotherFunc(n string) (int,error){
i,err := strconv.Atoi(n)
if err !=nil{
return i,err
}
return i,nil
}
func main(){
mymsg := myfuncation("Concurrency In GO")
if mymsg.Err != nil{
// 這裡可以換成其他的 log 框架，部分 log 框架會自動識別 error 來源。例如：func (m *MyMessage) myfuncation() 這樣的函式就會被抓到錯誤來自於哪裡。
fmt.Println(mymsg.Err)
}
}
```
# pipeline
我曾經在祖傳程式碼中見到一個約 2000 行的函式。我希望看見這篇文章的你，不要這麼做。我們已經瞭解了資料如何在兩個或多個 goroutine 之間通過 channel 傳遞，那我我們把這樣的程式用多個 channel組合在一起，其中的每一次讀出，或寫入channel 都是這一環上的一個 stage（步），這就是 pipeline。Go 語言的併發模式，讓我們很方便，快捷，安全的在一個程序中實現了流式處理。我們來看一個官方 pipeline 的例子：
```go
package main
import (
"fmt"
"sync"
"time"
)
func gen(nums ...int) <-chan int {
genOut := make(chan int)
go func() {
for _, n := range nums {
genOut <- n
}
fmt.Println("Input gen Channel number =>", len(genOut))
close(genOut)
}()
return genOut
}
func sq(done <-chan struct{}, in <-chan int) <-chan int {
sqOut := make(chan int)
go func() {
// 這個 close(sqOut) 一定要先寫，執行的時候優先壓入棧，待函式執行完成關閉 sqOut channel
defer close(sqOut)
for n := range in {
// 利用 select {} 均衡排程 channel
select {
case sqOut <- n * n:
fmt.Printf("=> %v <= write into sqOut channel \n", n*n)
case <-done:
return
}
}
//fmt.Printf("Wait close the chan => %v\n", len(sqOut))
}()
return sqOut
}
// merge Fan-In 函式合併多個結果
func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
var wg sync.WaitGroup
mergeOut := make(chan int, 1)
output := func(c <-chan int) {
defer wg.Done()
for n := range c {
select {
case mergeOut <- n:
case <-done:
return
}
}
}
wg.Add(len(cs))
for _, c := range cs {
go output(c)
}
go func() {
wg.Wait()
close(mergeOut)
}()
return mergeOut
}
// pfnumber 計算算數平方數
func pfnumber() {
// 定義 don channel 用於終止 pipeline
don := make(chan struct{}, 3)
don <- struct{}{}
don <- struct{}{}
close(don)
// 傳入 don 通知傳送方停止傳送
for n := range sq(don, sq(don, gen(3, 4, 2))) {
fmt.Println("Last result ", n)
}
fmt.Println("============================================")
}
func fanInOut() {
don := make(chan struct{}, 3)
in := gen(2, 3)
c1 := sq(don, in)
c2 := sq(don, in)
for n := range merge(don, c1, c2) {
fmt.Println(n)
}
don <- struct{}{}
don <- struct{}{}
don <- struct{}{}
fmt.Println("Finish channel len => ", len(don))
<-don
close(don)
}
func f1(i chan int) {
fmt.Println(<-i)
}
func runf1() {
out := make(chan int)
go f1(out)
time.Sleep(2 * time.Second)
out <- 2
time.Sleep(2 * time.Second)
}
func main() {
//runf1()
pfnumber()
// FanIn and FanOut
//fanInOut()
}
```
簡單總結一下如何正確構建一個 pipeline:
- 當所有的傳送已完成，stage 應該關閉輸出 channel
- stage 應該持續從只讀 channel 中讀出資料，除非 channel 關閉或主動通知到傳送方停止傳送
[Golang Pipeline Blog 譯文](https://juejin.cn/post/6844903864462737415)
[Golang Pipeline Blog](https://blog.golang.org/pipelines)
# 扇出、扇入
扇出模式優先的場景：
- 它不依賴於之前的 stage 計算的值
- 需要執行很長時間，例如：I/O 等待，遠端呼叫，訪問 REST full API等
扇入模式優先：
扇入意味著多個數據流複用或者合併成一個流。例如：上文 pipeline 中的 merge 函式，可以通過開啟 fanInOut() 函式執行一下試試。
# or-done-channel
在防止 goroutine 洩露，pipeline 中我們都在函式執行過程中嵌入了 done channel 以便終止需要停止的 goroutine。我們可以看出他們有個統一的特點，傳入 done ,jobChannel ，返回 resultChannel 。那麼我們可以把它封裝起來，像這樣：
```go
orDone := func(done ,c <-chan interface{}) <- chan interface{}{
valStream := make(chan interface{})
go func(){
defer close(valStream)
for {
select{
case <- done:
case v,ok := <- c:
if ok == false{
return
}
select{
case valStream <- v:
case <-done:
}
}
}
}()
return valStream
}
```
# tee-channel
可能需要將同一個結果傳送給兩個接收者，這個時候就需要用到 tee-channel 的方式。
應用場景：
- 流量映象
- 操作審計
```go
tee := func(done <- chan interface{},in <-chan interface{}
)(_,_ <- chan interface{}) { <-chan interface{}) {
out1 := make(chan interface{})
out2 := make(chan interface{})
go func(){
defer close(out1)
defer close(out2)
for val := range orDone(done, in){
var out1,out2 = out1,out2
for i:=0;i<2; i++{
select{
case <- done:
case out1 <- val:
out1 = nil
case out2 <- val:
out2 = nil
}
}
}
}()
return out1,out2
}
```
# 其他的應用場景
## 橋接 channel
在 channel 中傳遞 channel 。筆者學術才淺，紙上談兵多，動手實踐少，著實想不到合適的場景，希望讀者能為我補充一下。
## 佇列
佇列可能是我們第一次看見 channel 的感受，這玩意一個佇列，非常具備佇列的特性。
佇列在什麼樣的情況下可以提升整體效能
- 如果在一個 stage 批處理請求可以節省時間。
- 需要快取的場景，例如：批量日誌刷盤，熱資料快取等。
# context 包
在前文中經常會定義 done channel 的做法，防止 goroutine 洩露，或者主動中斷需要停止的 pipeline 。難道我們每次構建 pipeline 的時候都要建立 done channel 嗎？答案是否定的，Go 團隊為我們準備了 context 包，專用於幹類似的工作。
```go
type Context interface {
// 當該 context 工作的 work 被取消時，返回超時時間
Deadline() (deadline time.Time, ok bool)
// done 返回停止 pipeline 的 channel
Done()