Golang 經常使用併發編程技巧

《Golang 經常使用併發編程技巧》 最先發布在 blog.hdls.me/15726777274…

Golang 是最先將 CSP 原則歸入其核心的語言之一，並將這種併發編程風格引入到大衆中。CSP 指的是 Communicating Sequential Processes ，即通訊順序進程，每一個指令都須要指定具體是一個輸出變量（從一個進程中讀取一個變量的狀況），仍是一個目的地（將輸入發送到一個進程的狀況）。

Golang 不只提供了 CSP 樣式的併發方式，還支持經過內存訪問同步的傳統方式，本文對最經常使用的 Golang 併發編程工具作一個總結。

sync 包

sync 包包含了對低級別內存訪問同步最有用的併發原語，是 「內存訪問同步」 的最有利工具，也是傳統併發模型解決臨界區問題的經常使用工具。

WaitGroup

WaitGroup 是等待一組併發操做完成的方法，包含了三個函數：

func (wg *WaitGroup) Add(delta int) func (wg *WaitGroup) Done() func (wg *WaitGroup) Wait() 複製代碼

其中，Add() 用來添加 goroutine 的個數，Done() 是 goroutine 用來代表執行完成並退出，將計數減一，而 Wait() 用來等待全部 goroutine 退出。

用法以下：

func main() {
	wg := sync.WaitGroup{}

	wg.Add(1)
	go func() {
		defer wg.Done()
		fmt.Printf("goroutine 結束\n")
	}()

	wg.Wait()
}
複製代碼

須要注意的是，Add() 方法須要在 goroutine 以前執行。

互斥鎖和讀寫鎖

互斥是保護程序中臨界區的一種方式。一個互斥鎖只能同時被一個 goroutine 鎖定，其它 goroutine 將阻塞直到互斥鎖被解鎖（從新爭搶對互斥鎖的鎖定）。

用法以下：

func main() {
	var lock sync.Mutex
	var count int
	var wg sync.WaitGroup

	wg.Add(1)
	// count 加 1
	go func() {
		defer wg.Done()
		lock.Lock()
		defer lock.Unlock()
		count++
		fmt.Println("count=", count)
	}()

	// count 減 1
	wg.Add(1)
	go func() {
		defer wg.Done()
		lock.Lock()
		defer lock.Unlock()
		count--
		fmt.Println("count=", count)
	}()

	wg.Wait()
	fmt.Println("count=", count)
}
複製代碼

須要注意的是，在 goroutine 裏用 defer 來調用 Unlock 是個常見的習慣用法，確保了即便出現了 panic，調用也老是執行，防止出現死鎖。

讀寫鎖在概念上跟互斥鎖是同樣的：保護對內存的訪問，讀寫鎖讓你對內存有更多的控制。讀寫鎖與互斥鎖最大的不一樣就是能夠分別對讀、寫進行鎖定。通常用在大量讀操做、少許寫操做的狀況。

讀寫鎖的 Lock() 和 Unlock() 是對寫操做的鎖定和解鎖；Rlock() 和 RUnlock() 是對讀操做的鎖定和解鎖，須要配對使用。而讀鎖和寫鎖的關係：

1. 同時只能有一個 goroutine 可以得到寫鎖定。
2. 同時能夠有任意多個 gorouinte 得到讀鎖定。
3. 同時只能存在寫鎖定或讀鎖定（讀和寫互斥）。

Channel

Channel 是 CSP 派生的同步原語之一，是 Golang 推崇的 「使用通訊來共享內存，而不是經過共享內存來通訊」 理念的最有利的工具。

Channel 的基本使用這裏不展開講，但對不一樣狀態下的 Channel 不一樣操做的結果作一個總結：

操做	Channel 狀態	結果
Read	nil	阻塞
	打開非空	輸出值
	打開但空	阻塞
	關閉	<默認值>, false
	只寫	編譯錯誤
Write	nil	阻塞
	打開但填滿	阻塞
	打開不滿	寫入值
	關閉	panic
	只讀	編譯錯誤
Close	nil	panic
	打開非空	關閉 Channel; 讀取成功，直到 Channel 耗盡，讀取產生值的默認值
	打開但空	關閉 Channel；讀到生產者的默認值
	關閉	panic
	只讀	編譯錯誤

for-select

select 語句是將 Channel 綁定在一塊兒的粘合劑，可以讓一個 goroutine 同時等待多個 Channel 達到準備狀態。

select 語句是針對 Channel 的操做，語法上看上去與 switch 很像，但不一樣的是，select 塊中的 case 語句沒有測試順序，若是沒有知足任何條件，執行也不會失敗。用法以下：

var c1, c2 <-chan interface{}
select {
  case <- c2:
    // 某段邏輯
  case <- c2:
    // 某段邏輯
}
複製代碼

上面這個 select 控制結構會等待全部 case 條件語句任意一個的返回，不管哪個返回都會馬上執行 case 中的代碼，不過若是了 select 中的兩個 case 同時被觸發，就會隨機選擇一個 case 執行。

for-select 是一個很常見的用法，一般在 「向 Channel 發送迭代變量」 和 「循環等待中止」 兩種狀況下會用到，用法以下：

向 Channel 發送迭代變量：

func main() {
	c := make(chan int, 3)
	for _, s := range []int{1, 2, 3} {
		select {
		case c <- s:
		}
	}
}
複製代碼

循環等待中止：

// 第一種
for {
  select {
  case <- done:
    return
  default:
    // 進行非搶佔式任務
  }
}
// 第二種
for {
  select {
  case <- done:
    return
  default:
  }
  // 進行非搶佔式任務
}
複製代碼

第一種是指，當咱們輸入 select 語句時，若是完成的 Channel 還沒有關閉，咱們將執行 default 語句；第二種是指，若是已經完成的 Channel 未關閉，咱們將退出 select 語句並繼續執行 for 循環的其他部分。

done channel

雖然 goroutine 廉價且易於利用，運行時能夠將多個 goroutine 複用到任意數量的操做系統線程，但咱們須要知道的是 goroutine 是須要消耗資源的，而且是不會被運行時垃圾回收的。若是出現 goroutine 泄露的狀況，嚴重的時候會致使內存利用率的降低。

而 done channel 就是防止 goroutine 泄露的利器。用 done channel 在父子 goroutine 之間創建一個 「信號通道」，父 goroutine 能夠將該 channel 傳遞給子 goroutine ，而後在想要取消子 goroutine 的時候關閉該 channel。用法以下：

func main() {
	doneChan := make(chan interface{})

	go func(done <-chan interface{}) {
	   for {
		  select {
		  case <-done:
		    return
		  default:
		  }
		}
	}(doneChan)

	// 父 goroutine 關閉子 goroutine
	close(doneChan)
}
複製代碼

確保 goroutine 不泄露的方法，就是規定一個約定：若是 goroutine 負責建立 goroutine，它也負責確保它能夠中止 goroutine。

Context 包

Context 包是專門用來簡化對於處理單個請求的多個 goroutine 之間與請求域的數據、取消信號、截止時間等相關操做，這些操做可能涉及多個 API 調用。Context 包的目的主要有兩個：提供一個能夠取消你的調用圖中分支的 API，提供用於經過呼叫傳輸請求範圍數據的數據包。

若是使用 Context 包，那麼位於頂級併發調用下游的每一個函數都會將 context 做爲其第一個參數。

Context 的類型以下：

type Context interface {
  Deadline() (deadline time.Time, ok bool)
  Done() <-chan struct{}
  Err() error
  Value(key interface{}) interface{}
}
複製代碼

其中，Deadline 函數用於指示在必定時間後 goroutine 是否會被取消；Done 方法返回當咱們的函數被搶佔時關閉的 Channel；Err 方法返回取消的錯誤緣由，由於什麼 Context 被取消；Value 函數返回與此 Context 關聯的 key 或 nil。

Context 雖然是個接口，可是咱們在使用它的時候並不須要實現，context 包內置的兩個方法來建立上下文的實例：

func Background() Context func TODO() Context 複製代碼

Background 主要用於 main 函數、初始化以及測試代碼中，做爲Context 這個樹結構的最頂層的 Context，不能被取消；TODO，若是咱們不知道該使用什麼 Context 的時候，可使用這個，可是實際應用中，暫時尚未使用過這個 TODO。

而後以此做爲最頂層的父 Context，衍生出子 Context 啓動調用鏈。而這些 Context 對象造成了一棵樹，當父 Context 對象被取消時，它的全部子 Context 都會被取消。context 包還提供了一系列函數用以產生子 Context：

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc) func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) func WithValue(parent Context, key, val interface{}) Context 複製代碼

其中，WithCancel 返回一個新的 Context，在調用返回的 cancel 函數時關閉其 done channel；WithDeadline 返回一個新的 Context，當機器的時鐘超過給定的最後期限時，它關閉完成的 channel；WithTimeout 返回一個新的 Context，在給定的超時時間後關閉其完成的 channel；WithValue 生成一個綁定了一個鍵值對數據的 Context，這個綁定的數據能夠經過 Context.Value 方法訪問到。

下面來看使用方法：

WithCancel

func main() {
	wg := sync.WaitGroup{}
	ctx, cancel := context.WithCancel(context.Background())

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		for {
			select {
			case <-ctx.Done():
				fmt.Println("Err:", ctx.Err())
				return
			default:
			}
		}
	}(ctx)

	cancel()
	wg.Wait()
}
複製代碼

WithDeadline

func main() {
	d := time.Now().Add(1 * time.Second)
	wg := sync.WaitGroup{}
	ctx, cancel := context.WithDeadline(context.Background(), d)
	defer cancel()

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		for {
			select {
			case <-ctx.Done():
				fmt.Println("Err:", ctx.Err())
				return
			default:
			}
		}
	}(ctx)

	wg.Wait()
}
複製代碼

WithTimeout

func main() {
	wg := sync.WaitGroup{}
	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		for {
			select {
			case <-ctx.Done():
				fmt.Println("Err:", ctx.Err())
				return
			default:
			}
		}
	}(ctx)

	wg.Wait()
}
複製代碼

WithValue

func main() {
	wg := sync.WaitGroup{}
	ctx, cancel := context.WithCancel(context.Background())
	valueCtx := context.WithValue(ctx, "key", "add value")

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		for {
			select {
			case <-ctx.Done():
				fmt.Println("Err:", ctx.Err())
				return
			default:
				fmt.Println(ctx.Value("key"))
				time.Sleep(1 * time.Second)
			}
		}
	}(valueCtx)

	time.Sleep(5*time.Second)
	cancel()
	wg.Wait()
}
複製代碼

參考：《Go 語言併發之道》