Go并發編程實踐

2020-04-01 19:07:28

字體：大中小

來源：轉載

供稿：網友

前言

并發編程一直是Golang區別與其他語言的很大優勢，也是實際工作場景中經常遇到的。近日筆者在組內分享了我們常見的并發場景，及代碼示例，以期望大家能在遇到相同場景下，能快速的想到解決方案，或者是拿這些方案與自己實現的比較，取長補短?，F整理出來與大家共享。

簡單并發場景

很多時候，我們只想并發的做一件事情，比如測試某個接口的是否支持并發。那么我們就可以這么做：

func RunScenario1() {    count := 10    var wg sync.WaitGroup    for i := 0; i < count; i++ {       wg.Add(1)       go func(index int) {           defer wg.Done()           doSomething(index)       }(i)    }    wg.Wait()}

使用goroutine來實現異步，使用WaitGroup來等待所有goroutine結束。這里要注意的是要正確釋放WaitGroup的counter(在goroutine里調用Done()方法)。

但此種方式有個弊端，就是當goroutine的量過多時，很容易消耗完客戶端的資源，導致程序表現不佳。

規定時間內的持續并發模型

我們仍然以測試某個后端API接口為例，如果我們想知道這個接口在持續高并發情況下是否有句柄泄露，這種情況該如何測試呢？

這種時候，我們需要能控制時間的高并發模型：

func RunScenario2() {  timeout := time.Now().Add(time.Second * time.Duration(10))  n := runtime.NumCPU()  waitForAll := make(chan struct{})  done := make(chan struct{})  concurrentCount := make(chan struct{}, n)  for i := 0; i < n; i++ {    concurrentCount <- struct{}{}  }  go func() {    for time.Now().Before(timeout) {      <-done      concurrentCount <- struct{}{}    }    waitForAll <- struct{}{}  }()  go func() {    for {      <-concurrentCount      go func() {        doSomething(rand.Intn(n))        done <- struct{}{}      }()    }  }()  <-waitForAll}

上面的代碼里，我們通過一個buffered channel來控制并發的數量(concurrentCount)，然后另起一個channel來周期性的發起新的任務，而控制的條件就是 time.Now().Before(timeout)，這樣當超過規定的時間，waitForAll 就會得到信號，而使整個程序退出。

這是一種實現方式，那么還有其他的方式沒？我們接著往下看。

基于大數據量的并發模型

前面說的基于時間的并發模型，那如果只知道數據量很大，但是具體結束時間不確定，該怎么辦呢？

比如，客戶給了個幾TB的文件列表，要求把這些文件從存儲里刪除。再比如，實現個爬蟲去爬某些網站的所有內容。

而解決此類問題，最常見的就是使用工作池模式了(Worker Pool)。以刪文件為例，我們可以簡單這樣來處理：

Go并發

Jobs - 可以從文件列表里讀取文件，初始化為任務，然后發給worker
Worker - 拿到任務開始做事
Collector - 收集worker處理后的結果
Worker Pool - 控制并發的數量

雖然這只是個簡單Worker Pool模型，但已經能滿足我們的需求：

func RunScenario3() {    numOfConcurrency := runtime.NumCPU()    taskTool := 10    jobs := make(chan int, taskTool)    results := make(chan int, taskTool)    var wg sync.WaitGroup    // workExample    workExampleFunc := func(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {       defer wg.Done()       for job := range jobs {           res := job * 2           fmt.Printf("Worker %d do things, produce result %d /n", id, res)           time.Sleep(time.Millisecond * time.Duration(100))           results <- res       }    }    for i := 0; i < numOfConcurrency; i++ {       wg.Add(1)       go workExampleFunc(i, jobs, results, &wg)    }    totalTasks := 100 // 本例就要從文件列表里讀取    wg.Add(1)    go func() {       defer wg.Done()       for i := 0; i < totalTasks; i++ {           n := <-results           fmt.Printf("Got results %d /n", n)       }       close(results)    }()    for i := 0; i < totalTasks; i++ {       jobs <- i    }    close(jobs)    wg.Wait()}

在Go里，分發任務，收集結果，我們可以都交給Channel來實現。從實現上更加的簡潔。

仔細看會發現，本模型也是適用于按時間來控制并發。只要把totalTask的遍歷換成時間控制就好了。

等待異步任務執行結果

goroutine和channel的組合在實際編程時經常會用到，而加上Select更是無往而不利。

func RunScenario4() {    sth := make(chan string)    result := make(chan string)    go func() {       id := rand.Intn(100)       for {           sth <- doSomething(id)       }    }()    go func() {       for {           result <- takeSomthing(<-sth)       }    }()    select {    case c := <-result:       fmt.Printf("Got result %s ", c)    case <-time.After(time.Duration(30 * time.Second)):       fmt.Errorf("指定時間內都沒有得到結果")    }}

在select的case情況，加上time.After()模型可以讓我們在一定時間范圍內等待異步任務結果，防止程序卡死。

定時反饋異步任務結果

上面我們說到持續的壓測某后端API，但并未實時收集結果。而很多時候對于性能測試場景，實時的統計吞吐率，成功率是非常有必要的。

func RunScenario5() {  concurrencyCount := runtime.NumCPU()  for i := 0; i < concurrencyCount; i++ {    go func(index int) {      for {        doUploadMock()      }    }(i)  }  t := time.NewTicker(time.Second)  for {    select {    case <-t.C:      // 計算并打印實時數據    }  } }

這種場景就需要使用到Ticker，且上面的Example模型還能控制并發數量，也是非常實用的方式。

知識點總結

上面我們共提到了五種并發模式：